AMQP close-reason, code = 541, text = 'Unexpected Exception', SocketException(10054)

[fish-li]

This exception has been encountered a lot recently and has been found to be related to the number of threads.
See the following code example.

If the threadCount is less than 10, the code works fine.
My machine has an I5-7500 CPU and it has four cores.

public static class TestRabbitPostIssue
{
    private static readonly string s_exchange = "amq.direct";
    private static readonly string s_queueName = "issue_demo";

    private static readonly int s_threadCount = 30;

    [MenuMethod(Title = "RabbitMQ-issue-DEMO-1")]
    public static void Test1()
    {
        Stopwatch watch = Stopwatch.StartNew();
        Task[] tasks = new Task[s_threadCount];

        for( int i = 0; i < s_threadCount; i++ ) {
            tasks[i] = Task.Run(Execute);
        }
        Task.WaitAll(tasks);

        watch.Stop();
        Console.WriteLine(watch.Elapsed.ToString());
    }

    private static void Execute()
    {
        using( IConnection  connection = CreateConnection() ) {
            using( IModel channel = connection.CreateModel() ) {

                // prepare queue
                channel.QueueDeclare(s_queueName, true, false, false, null);
                channel.QueueBind(s_queueName, s_exchange, s_queueName);

                // prepare message
                string message = new string('a', 512);
                byte[] body = Encoding.UTF8.GetBytes(message);

                IBasicProperties props = channel.CreateBasicProperties();
                props.ContentType = "text/plain";
                props.Persistent = true;

                // send message
                for( int i = 0; i < 1_0000; i++ ) {                        
                    channel.BasicPublish(s_exchange, s_queueName, props, body);
                }
            }
        }
    }

    private static IConnection CreateConnection()
    {
        ConnectionFactory factory = new ConnectionFactory();

        factory.AutomaticRecoveryEnabled = true;
        factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(1);

        // modify the following parameters as required
        factory.HostName = "RabbitHost";
        factory.UserName = "fishli";
        factory.Password = "1qaz7410";
        factory.VirtualHost = "nbtest";

        return factory.CreateConnection();    // there will be throw an exception
    }
}

exception text:

System.Reflection.TargetInvocationException: Exception has been thrown by the target of an invocation.
 ---> System.AggregateException: One or more errors occurred. (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable) (None of the specified endpoints were reachable)
 ---> RabbitMQ.Client.Exceptions.BrokerUnreachableException: None of the specified endpoints were reachable
 ---> RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=541, text='Unexpected Exception', classId=0, methodId=0, cause=System.IO.IOException: Unable to read data from the transport connection: 远程主机强迫关闭了一个现有的连接。.
 ---> System.Net.Sockets.SocketException (10054): 远程主机强迫关闭了一个现有的连接。
   at System.Net.Sockets.NetworkStream.Read(Byte[] buffer, Int32 offset, Int32 size)
   --- End of inner exception stack trace ---
   at System.Net.Sockets.NetworkStream.Read(Byte[] buffer, Int32 offset, Int32 size)
   at System.IO.BufferedStream.ReadByteSlow()
   at System.IO.BufferedStream.ReadByte()
   at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
   at RabbitMQ.Client.Impl.SocketFrameHandler.ReadFrame()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
   at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
   at RabbitMQ.Client.Impl.ModelBase.ConnectionOpen(String virtualHost, String capabilities, Boolean insist)
   at RabbitMQ.Client.Framing.Impl.Connection.Open(Boolean insist)
   at RabbitMQ.Client.Framing.Impl.Connection..ctor(IConnectionFactory factory, Boolean insist, IFrameHandler frameHandler, String clientProvidedName)
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.Init(IFrameHandler fh)
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.Init(IEndpointResolver endpoints)
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(IEndpointResolver endpointResolver, String clientProvidedName)
   --- End of inner exception stack trace ---
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(IEndpointResolver endpointResolver, String clientProvidedName)
   at RabbitMQ.Client.ConnectionFactory.CreateConnection(String clientProvidedName)
   at RabbitMQ.Client.ConnectionFactory.CreateConnection()
   at PerformanceTest.MQ.TestRabbitPostIssue.CreateConnection() in D:\WorkCode\Nebula.net\test\PerformanceTest\MQ\TestRabbitPostIssue.cs:line 71
   at PerformanceTest.MQ.TestRabbitPostIssue.Execute() in D:\WorkCode\Nebula.net\test\PerformanceTest\MQ\TestRabbitPostIssue.cs:line 36
   at System.Threading.Tasks.Task.InnerInvoke()
   at System.Threading.Tasks.Task.<>c.<.cctor>b__277_0(Object obj)
   at System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(Thread threadPoolThread, ExecutionContext executionContext, ContextCallback callback, Object state)
--- End of stack trace from previous location ---
   at System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(Thread threadPoolThread, ExecutionContext executionContext, ContextCallback callback, Object state)
   at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot, Thread threadPoolThread)
   --- End of inner exception stack trace ---
   at System.Threading.Tasks.Task.WaitAllCore(Task[] tasks, Int32 millisecondsTimeout, CancellationToken cancellationToken)
   at System.Threading.Tasks.Task.WaitAll(Task[] tasks)
   at PerformanceTest.MQ.TestRabbitPostIssue.Test1() in D:\WorkCode\Nebula.net\test\PerformanceTest\MQ\TestRabbitPostIssue.cs:line 28

[michaelklishin]

I will convert this issue to a GitHub discussion. Currently GitHub will automatically close and lock the issue even though your question will be transferred and responded to elsewhere. This is to let you know that we do not intend to ignore this but this is how the current GitHub conversion mechanism makes it seem for the users :(

[michaelklishin]

You need to inspect server logs.

Another obvious thing that stands out is that you publish without using publisher confirms.

[michaelklishin]

What also stands out is that the exception happens when you try to open a connection. So it's either a thread pool exhaustion on the .NET side or you hit a limit of some kind (e.g. the open file handle limit) elsewhere but 10 is a low number of connections even with all defaults. Nonetheless, you won't get a complete picture just from the client connection exception.

[fish-li]

Failed to open connection related to thread pool ?

[add0212]

Has this issue got resolved? I'm facing similar issue when trying to connect to RabbitMQ from windows machine.

Exception: <class 'pika.exceptions.StreamLostError'> Stream connection lost: ConnectionResetError(10054, 'An existing connection was forcibly closed by the remote host', None, 10054, None)

[michaelklishin]

ConnectionResetError with the details of An existing connection was forcibly closed by the remote host means that the node closed this connection, either because it was shut down or because it used the pause-minority or autoheal partition handling mode, and had to pause itself on the minority side or restart.

There is nothing a client can do about this. Applications should be prepared to handle node shutdown by recovering connections, the topology, and so on, and using publisher confirms.

[michaelklishin]

If you are trying to connect and run into a similar exception, you may be running into this default security limitation or a network connectivity issue.