Lower throughput with a workload that heavily churns channels when upgrading from 5.1.2 to 6.1.0

[aberkromb]

We updated the version of RabbitMQ.Client library (from 5.1.2 to 6.1.0) and met a decrease in performance.
Below a minimal repro.

Framework: netcoreapp 3.1(our internal library netstandard 2.0)
RabbitMQ.Client library version: 5.1.2 and 6.1.0
RabbitMQ version: 3.8
Docker version: 19.03.8
Docker Image: rabbitmq:3.8-management-alpine
OS version: Win10 1903

attached our scripts to run a local cluster for tests
scripts.zip

For 5.1.2 i've got ~5 sec, for 6.1.0 ~58 sec

static void Main(string[] args)
        {
            var ids = Enumerable.Range(1, 512).Select(_ => Guid.NewGuid());
           
            var (factory, resolver) = ConfigureFactoryAndResolver();

            var sw = Stopwatch.StartNew();
            using (var connection = factory.CreateConnection(resolver, "example"))
            {
                Parallel.ForEach(ids,
                    new ParallelOptions {MaxDegreeOfParallelism = ids.Count()},
                    id => Publish(connection, id));
            }

            sw.Stop();
            Console.WriteLine(sw.Elapsed);
        }

        private static void Publish(IConnection connection, Guid message)
        {
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare("test-exchange", "topic", true);
                channel.QueueDeclare("test-queue", true, false, false);
                channel.QueueBind("test-queue", "test-exchange", "test-rk");
                channel.BasicPublish("test-exchange", "test-rk", null, Encoding.UTF8.GetBytes(message.ToString()));
            }
        }
        
        
        /// <summary>
        ///     Create Factory for v5.1.2
        /// </summary>
        private static (ConnectionFactory factory, DefaultEndpointResolver resolver) ConfigureFactoryAndResolver()
        {
            var connectionFactory = new ConnectionFactory
            {
                HostName = "127.0.0.11",
                VirtualHost = "/",
                Port = 5672,
                UserName = "admin",
                Password = "admin",
                AutomaticRecoveryEnabled = true,
                TopologyRecoveryEnabled = false,
                RequestedHeartbeat = (ushort) TimeSpan.FromSeconds(10).TotalSeconds,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(5),
                RequestedConnectionTimeout = (int) TimeSpan.FromSeconds(5).TotalMilliseconds,
                ContinuationTimeout = TimeSpan.FromSeconds(20),
                HandshakeContinuationTimeout = TimeSpan.FromSeconds(10),
                SocketReadTimeout = (int) TimeSpan.FromSeconds(30).TotalMilliseconds,
                SocketWriteTimeout = (int) TimeSpan.FromSeconds(30).TotalMilliseconds,
                UseBackgroundThreadsForIO = true
            };

            var endpoints = new[] {"127.0.0.13", "127.0.0.12", "127.0.0.11"}.Select(h =>
                new AmqpTcpEndpoint(h, connectionFactory.Port, connectionFactory.Ssl));

            return (connectionFactory, new DefaultEndpointResolver(endpoints));
        }
        
        /// <summary>
        ///     Create Factory for v6.1.0
        /// </summary>     
        // private static (ConnectionFactory factory, DefaultEndpointResolver resolver) ConfigureFactoryAndResolver()
        // {
        //     var connection_factory = new ConnectionFactory();
        //     connection_factory.HostName = "127.0.0.13";
        //     connection_factory.VirtualHost = "/";
        //     connection_factory.Port = 5672;
        //     connection_factory.UserName = "admin";
        //     connection_factory.Password = "admin";
        //     connection_factory.AutomaticRecoveryEnabled = true;
        //     connection_factory.TopologyRecoveryEnabled = false;
        //     connection_factory.RequestedHeartbeat = TimeSpan.FromSeconds(10); // пинг идёт два раза за указанное время
        //     connection_factory.NetworkRecoveryInterval = TimeSpan.FromSeconds(5);
        //     connection_factory.RequestedConnectionTimeout = TimeSpan.FromSeconds(5);
        //     connection_factory.ContinuationTimeout = TimeSpan.FromSeconds(20);
        //     connection_factory.HandshakeContinuationTimeout = TimeSpan.FromSeconds(10);
        //     connection_factory.SocketReadTimeout = TimeSpan.FromSeconds(30);
        //     connection_factory.SocketWriteTimeout = TimeSpan.FromSeconds(30);
        //     connection_factory.UseBackgroundThreadsForIO = true;
        //
        //     var endpoints = new[] {"127.0.0.13", "127.0.0.12", "127.0.0.11"}.Select(h =>
        //         new AmqpTcpEndpoint(h, connection_factory.Port, connection_factory.Ssl));
        //
        //     return (connection_factory, new DefaultEndpointResolver(endpoints));
        // }

[aberkromb]

repro project
ConsoleApp1.zip

[michaelklishin]

This code opens a new channel and then sets up a topology for every message published. This is not a scenario we will optimize for as it is not how channels are expected to be used and is incredibly wasteful.

[michaelklishin]

In fact, this workload is more or less what was covered in #860.

[aberkromb]

Thank you!

[aberkromb]

This code opens a new channel and then sets up a topology for every message published. This is not a scenario we will optimize for as it is not how channels are expected to be used and is incredibly wasteful.

I tried to use 1 connection and 1 channel.
And got exception

 ---> (Inner Exception #57) RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=505, text='UNEXPECTED_FRAME - expected content
header for class 60, got non content header frame instead', classId=60, methodId=40
   at RabbitMQ.Client.Impl.SessionBase.Transmit(Command cmd)
   at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)
   at RabbitMQ.Client.Framing.Impl.Model._Private_ExchangeDeclare(String exchange, String type, Boolean passive, Boolean durable, Boolean autoDelete, Boolean internal, Boolean nowait, IDictionary`2 arguments)
   at RabbitMQ.Client.Impl.AutorecoveringModel.ExchangeDeclare(String exchange, String type, Boolean durable, Boolean autoDelete, IDictionary`2 arguments)
   at ConsoleApp1.Program.Publish(IModel channel, Guid message) in C:\Users\d.nekrasov\RiderProjects\ClassLibrary1\ConsoleApp1\Program.cs:line 50
   at ConsoleApp1.Program.<>c__DisplayClass0_0.<Main>b__1(Guid id) in C:\Users\d.nekrasov\RiderProjects\ClassLibrary1\ConsoleApp1\Program.cs:line 27
   at System.Threading.Tasks.Parallel.<>c__DisplayClass44_0`2.<PartitionerForEachWorker>b__1(IEnumerator& partitionState, Int32 timeout, Boolean& replicationDelegateYieldedBeforeCompletion)
--- End of stack trace from previous location where exception was thrown ---
   at System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw(Exception source)
   at System.Threading.Tasks.Parallel.<>c__DisplayClass44_0`2.<PartitionerForEachWorker>b__1(IEnumerator& partitionState, Int32 timeout, Boolean& replicationDelegateYieldedBeforeCompletion)
   at System.Threading.Tasks.TaskReplicator.Replica.Execute()<---

Tried reuse channel from pool.
Result: 5.1.2 ~0.5 sec 6.1.0 ~15 sec

static void Main(string[] args)
        {
            var ids = Enumerable.Range(1, 100).Select(_ => Guid.NewGuid());
        
            var sw = Stopwatch.StartNew();
            var (factory, resolver) = ConfigureFactoryAndResolver();
        
            using (var connection = factory.CreateConnection(resolver, "example"))
            {
                var pool = CreatePool(connection, 20);
        
                Parallel.ForEach(ids,
                    new ParallelOptions {MaxDegreeOfParallelism = ids.Count()},
                    id => Publish(pool, id));
            }
        
            sw.Stop();
            Console.WriteLine(sw.Elapsed);
        }
        
        private static ConcurrentQueue<IModel> CreatePool(IConnection connection, int size) =>
            Enumerable.Range(1, size).Aggregate(new ConcurrentQueue<IModel>(), (queue, i) =>
            {
                queue.Enqueue(connection.CreateModel());
                return queue;
            });
        
        private static void Publish(ConcurrentQueue<IModel> pool, Guid message)
        {
            IModel channel;
            while (!pool.TryDequeue(out channel))
            {
                Thread.Sleep(500);
            }
        
            channel.ExchangeDeclare("test-exchange", "topic", true);
            channel.QueueDeclare("test-queue", true, false, false);
            channel.QueueBind("test-queue", "test-exchange", "test-rk");
            channel.BasicPublish("test-exchange", "test-rk", null, Encoding.UTF8.GetBytes(message.ToString()));
        
            pool.Enqueue(channel);
        }

[michaelklishin]

@aberkromb channels must not be shared between threads for publishing. The channel pool approach avoids such sharing.

There is no reason to declare an exchange, declare a queue then binding them for every message published. That adds three round trips for three synchronous protocol operations to what is supposed to be only a publish (an entirely asynchronous protocol operation). Topology declaration should be done once on application startup when possible and in this example, it absolutely is (all names and bindings used constant values).

Please use a profiler to compare the two versions and share your findings on the mailing list and in #860 (which most external contributors to this client have already been notified of).