Concurrent Publish using shared model fails

[bard83]

Hi,
one test in our code base - that was previously working - doesn't work anymore. N messages were published - correctly - over the MQ and then consumed by a consumer components (the component under test). After upgrading RabbitMQ and also the client the messages were not published anymore. Did the #878 introduced a change? Am I doing something wrong?

Expectations: all messages are published and confirmed and available to be consumed.
Results: few messages are published and confirmed, few messages are not confirmed and many get exceptions on publish.

Here the tests:

using NUnit.Framework;
using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

    public class RabbitMqTests
    {
        [Test]
        public async Task TestMultithreadPublishWithBarrier()
        {
            var count = 500;
            var barrier = new AsyncBarrier(count);
            var exchangeName = nameof(TestMultithreadPublishWithBarrier);
            var factory = new ConnectionFactory
            {
                UserName = "rabbitmq",
                Password = "rabbitmq",
                HostName = "localhost",
                Port = 5672,
                VirtualHost = "/",
            };

            var connection = factory.CreateConnection(nameof(TestMultithreadPublishWithBarrier));
            var tasksList = new List<Task<(int Code, string? Error)>>();
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, true);
                channel.ConfirmSelect();
                for (int i = 0; i < count; i++)
                {
                    var task = PublishAsync(channel, $"Message {i}", exchangeName, $"rabbit.test.{i}", barrier);
                    tasksList.Add(task);
                }

                await Task.WhenAll(tasksList);
                Console.Error.WriteLine($"# publishing error: {tasksList.Where(t => t.Result.Code == 1).Count()}");
                Console.Error.WriteLine($"# confirms error: {tasksList.Where(t => t.Result.Code == 2).Count()}");
                Console.Error.WriteLine($"# nack-ed: {tasksList.Where(t => t.Result.Code == 3).Count()}");
                Console.Error.WriteLine($"# not confirmed: {tasksList.Where(t => t.Result.Code == 4).Count()}");
                Console.Error.WriteLine($"# publishing error: {tasksList.Where(t => t.Result.Code == 1).First().Result.Error}");
                Assert.IsTrue(tasksList.Where(t => t.Result.Code == 0).Count() == count);
            }
        }

        [Test]
        public async Task TestMultithreadPublish()
        {
            var count = 500;
            var exchangeName = nameof(TestMultithreadPublish);
            var factory = new ConnectionFactory
            {
                UserName = "rabbitmq",
                Password = "rabbitmq",
                HostName = "localhost",
                Port = 5672,
                VirtualHost = "/",
            };

            var connection = factory.CreateConnection(nameof(TestMultithreadPublish));
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, true);
                channel.ConfirmSelect();
                var tasksList = new List<Task<(int Code, string? Error)>>();
                for (int i = 0; i < count; i++)
                {
                    var task = Task.Run(() => Publish(channel, $"Message {i}", exchangeName, $"rabbit.test.{i}"));
                    tasksList.Add(task);
                }

                await Task.WhenAll(tasksList);
                Console.Error.WriteLine($"# publishing error: {tasksList.Where(t => t.Result.Code == 1).Count()}");
                Console.Error.WriteLine($"# confirms error: {tasksList.Where(t => t.Result.Code == 2).Count()}");
                Console.Error.WriteLine($"# nack-ed: {tasksList.Where(t => t.Result.Code == 3).Count()}");
                Console.Error.WriteLine($"# not confirmed: {tasksList.Where(t => t.Result.Code == 4).Count()}");
                Assert.IsTrue(tasksList.Where(t => t.Result.Code == 0).Count() == count);
            }
        }

        private async Task<(int Code, string? Error)> PublishAsync(IModel channel, string message, string exchangeName, string routingKey, AsyncBarrier barrier)
        {
            await barrier.SignalAndWaitAsync();

            return Publish(channel, message, exchangeName, routingKey);
        }

        private (int Code, string? Error) Publish(IModel channel, string message, string exchangeName, string routingKey)
        {
            byte[] byteMessage = Encoding.UTF8.GetBytes(message);

            IBasicProperties props = channel.CreateBasicProperties();
            props.ContentType = "application/json";
            props.Persistent = true;

            bool isConfirmed;
            bool timedOut;
            lock (channel)
            {
                try
                {
                    channel.BasicPublish(exchangeName, routingKey, props, byteMessage);
                }
                catch (Exception e)
                {
                    return (1, $"{message}: {e.ToString()}");
                }

                try
                {
                    isConfirmed = channel.WaitForConfirms(new TimeSpan(0, 1, 0), out timedOut);
                }
                catch (Exception e)
                {
                    return (2, $"{message}: {e.ToString()}");
                }
            }

            if (!isConfirmed)
            {
                return (3, $"{message}: got NACK-ed");
            }

            if (timedOut)
            {
                return (4, $"{message}: got time-out on waiting for message confirmaiton");
            }

            return (0, null);
        }

        private class AsyncBarrier
        {
            private int _count;
            private TaskCompletionSource<bool> _tcs;
            private int _waiting;

            public AsyncBarrier(int count)
            {
                _count = count;
                _waiting = 0;
                // Compiler doesn't recognize the initialization in a method...
                _tcs = new TaskCompletionSource<bool>();
            }

            public Task SignalAndWaitAsync()
            {
                var tcs = _tcs;
                if (Interlocked.Increment(ref _waiting) == _count)
                {
                    // Reset();
                    tcs.SetResult(true);
                }

                return tcs.Task;
            }

            private void Reset()
            {
                _waiting = 0;
                _tcs = new TaskCompletionSource<bool>();
            }
        }
    }

RabbitMQ instance

version: '2.4'

services:
  mq:
    image: rabbitmq:3.8.14-management-alpine
    ports:
    - '5672:5672'
    - '15672:15672'
    environment:
         RABBITMQ_NODENAME=eventStoreMq
        RABBITMQ_DEFAULT_USER=rabbitmq
        RABBITMQ_DEFAULT_PASS=rabbitmq
    healthcheck:
      test: ["CMD-SHELL", "rabbitmq-diagnostics -q ping"]
      interval: 1s
      timeout: 5s
      retries: 5

RabbitMQ logs

mq_1                | 2021-04-13 18:49:07.840 [info] <0.1050.0> Connection <0.1050.0> (172.19.0.1:42592 -> 172.19.0.2:5672) has a client-provided name: TestConcurrentPublish
mq_1                | 2021-04-13 18:49:07.845 [info] <0.1050.0> connection <0.1050.0> (172.19.0.1:42592 -> 172.19.0.2:5672 - TestConcurrentPublish): user 'rabbitmq' authenticated and granted access to vhost '/'
mq_1                | 2021-04-13 18:54:07.849 [error] <0.1050.0> closing AMQP connection <0.1050.0> (172.19.0.1:42592 -> 172.19.0.2:5672 - TestConcurrentPublish):
mq_1                | missed heartbeats from client, timeout: 60s
mq_1                | 2021-04-13 18:54:07.849 [info] <0.1462.0> Closing all channels from connection '172.19.0.1:42592 -> 172.19.0.2:5672' because it has been closed

dotnet --info

.NET SDK (reflecting any global.json):
 Version:   5.0.202
 Commit:    db7cc87d51

Runtime Environment:
 OS Name:     ubuntu
 OS Version:  18.04
 OS Platform: Linux
 RID:         ubuntu.18.04-x64
 Base Path:   /snap/dotnet-sdk/120/sdk/5.0.202/

Host (useful for support):
  Version: 5.0.5
  Commit:  2f740adc14

.NET SDKs installed:
  5.0.202 [/snap/dotnet-sdk/120/sdk]

.NET runtimes installed:
  Microsoft.AspNetCore.App 5.0.5 [/snap/dotnet-sdk/120/shared/Microsoft.AspNetCore.App]
  Microsoft.NETCore.App 5.0.5 [/snap/dotnet-sdk/120/shared/Microsoft.NETCore.App]

To install additional .NET runtimes or SDKs:
  https://aka.ms/dotnet-download

[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]

From .NET client guide's Concurrency Considerations section:

This is a hard requirement for publishers: sharing a channel (an IModel instance) for concurrent publishing will lead to incorrect
frame interleaving at the protocol level. Channel instances must not be shared by threads that publish on them.

So yes, you are doing something wrong. Concurrent publishing on a shared channel was always explicitly not supported
by this client. Publishing a message means publishing 2 or more (usually 3 or more) frames, and when done concurrently,
their interleaving on the wire will be arbitrary.

[michaelklishin]

#868 and earlier changes that attempt to send all N frames in a single TCP frame generally reduce the probability of incorrect interleaving by trying to pack N AMQP 0-9-1 frames into a single stream of data or even a single TCP frame However, it is still not guaranteed and it has never been our goal to support concurrent publishing on a shared channel.

If you need concurrent publishing on a connection, use multiple channels or synchronise as you see fit. It should perhaps be noted that opening a new channel is a network round-trip so channels should be opened and used for a period of time. Opening a channel per message published from a task will be less efficient than sequential publishing on a single channel.

[bard83]

Hi,

thanks for your reply. My bad.
I've probably messed things up. Originally I've read the concurrent considerations, looking back to my code, I had a lock(channel) instruction and the tests worked. After the Rabbitmq instanced has been upgraded, I've got a lot connections timeout when the channel were created (var channel = connection.CreateModel()). I went over the doc - forgetting about the concurrent considerations and the shared channel - and I've read only about the connection and channel lifespan. Now I'm on my original problem where I got timeout exceptions when a channel is created. I can image that I should delay the channel creation between tasks, right?

Below you can find my current exception and the updated code.

System.TimeoutException: The operation has timed out.
   at RabbitMQ.Util.BlockingCell`1.WaitForValue(TimeSpan timeout)
   at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
   at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)
   at RabbitMQ.Client.Framing.Impl.Model._Private_ChannelOpen(String outOfBand)
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateNonRecoveringModel()
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()
   at GeneralHandler.IntegrationTests.RabbitMqTests.Publish(IConnection connection, String message, String exchangeName, String routingKey) in /home/bard83/jungle_project/general-sql-handler/GeneralHandler.IntegrationTests/RabbitMqTests.cs:line 93

   public class RabbitMqTests
    {
        [Test]
        public async Task TestConcurrentPublishWithBarrier()
        {
            var count = 100;
            var barrier = new AsyncBarrier(count);
            var exchangeName = nameof(TestConcurrentPublishWithBarrier);
            var factory = new ConnectionFactory
            {
                UserName = "rabbitmq",
                Password = "rabbitmq",
                HostName = "localhost",
                Port = 5672,
                VirtualHost = "/",
            };

            var connection = factory.CreateConnection(nameof(TestConcurrentPublishWithBarrier));
            var tasksList = new List<Task<(int Code, string? Error)>>();

            for (int i = 0; i < count; i++)
            {
                var task = PublishAsync(connection, $"Message {i}", exchangeName, $"rabbit.test.{i}", barrier);
                tasksList.Add(task);
            }

            await Task.WhenAll(tasksList);
            Console.Error.WriteLine($"# create channel error: {tasksList.Where(t => t.Result.Code == -1).Count()}");
            Console.Error.WriteLine($"# publishing error: {tasksList.Where(t => t.Result.Code == 1).Count()}");
            Console.Error.WriteLine($"# confirms error: {tasksList.Where(t => t.Result.Code == 2).Count()}");
            Console.Error.WriteLine($"# nack-ed: {tasksList.Where(t => t.Result.Code == 3).Count()}");
            Console.Error.WriteLine($"# not confirmed: {tasksList.Where(t => t.Result.Code == 4).Count()}");
            Assert.IsTrue(tasksList.Where(t => t.Result.Code == 0).Count() == count);
        }

        [Test]
        public async Task TestConcurrentPublish()
        {
            var count = 100;
            var exchangeName = nameof(TestConcurrentPublish);
            var factory = new ConnectionFactory
            {
                UserName = "rabbitmq",
                Password = "rabbitmq",
                HostName = "localhost",
                Port = 5672,
                VirtualHost = "/",
            };

            var connection = factory.CreateConnection(nameof(TestConcurrentPublish));
            var tasksList = new List<Task<(int Code, string? Error)>>();
            for (int i = 0; i < count; i++)
            {
                var task = Task.Run(() => Publish(connection, $"Message {i}", exchangeName, $"rabbit.test.{i}"));
                tasksList.Add(task);
            }

            await Task.WhenAll(tasksList);
            Console.Error.WriteLine($"# create channel error: {tasksList.Where(t => t.Result.Code == -1).Count()}");
            Console.Error.WriteLine($"# publishing error: {tasksList.Where(t => t.Result.Code == 1).Count()}");
            Console.Error.WriteLine($"# confirms error: {tasksList.Where(t => t.Result.Code == 2).Count()}");
            Console.Error.WriteLine($"# nack-ed: {tasksList.Where(t => t.Result.Code == 3).Count()}");
            Console.Error.WriteLine($"# not confirmed: {tasksList.Where(t => t.Result.Code == 4).Count()}");
            Assert.IsTrue(tasksList.Where(t => t.Result.Code == 0).Count() == count);
        }

        private async Task<(int Code, string? Error)> PublishAsync(IConnection connection, string message, string exchangeName, string routingKey, AsyncBarrier barrier)
        {
            await barrier.SignalAndWaitAsync();

            return Publish(connection, message, exchangeName, routingKey);
        }

        private (int Code, string? Error) Publish(IConnection connection, string message, string exchangeName, string routingKey)
        {
            byte[] byteMessage = Encoding.UTF8.GetBytes(message);
            bool isConfirmed;
            bool timedOut;
            IModel channel_;
            try
            {
                channel_ = connection.CreateModel();
            }
            catch (Exception e)
            {
                return (-1, e.ToString());
            }

            lock (channel_)
            {
                using (var channel = channel_)
                {
                    channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, true);
                    channel.ConfirmSelect();
                    IBasicProperties props = channel.CreateBasicProperties();
                    props.ContentType = "application/json";
                    props.Persistent = true;

                    lock (channel)
                    {
                        try
                        {
                            channel.BasicPublish(exchangeName, routingKey, props, byteMessage);
                        }
                        catch (Exception e)
                        {
                            return (1, $"{message}: {e.ToString()}");
                        }

                        try
                        {
                            isConfirmed = channel.WaitForConfirms(new TimeSpan(0, 1, 0), out timedOut);
                        }
                        catch (Exception e)
                        {
                            return (2, $"{message}: {e.ToString()}");
                        }
                    }
                }
            }

            if (!isConfirmed)
            {
                return (3, $"{message}: got NACK-ed");
            }

            if (timedOut)
            {
                return (4, $"{message}: got time-out on waiting for message confirmaiton");
            }

            return (0, null);
        }

        private class AsyncBarrier
        {
            private int _count;
            private TaskCompletionSource<bool> _tcs;
            private int _waiting;

            public AsyncBarrier(int count)
            {
                _count = count;
                _waiting = 0;
                // Compiler doesn't recognize the initialization in a method...
                _tcs = new TaskCompletionSource<bool>();
            }

            public Task SignalAndWaitAsync()
            {
                var tcs = _tcs;
                if (Interlocked.Increment(ref _waiting) == _count)
                {
                    // Reset();
                    tcs.SetResult(true);
                }

                return tcs.Task;
            }

            private void Reset()
            {
                _waiting = 0;
                _tcs = new TaskCompletionSource<bool>();
            }
        }
    }

[bollhals]

I didn't see it until now, but I'd argue that the issue is that the threadpool is blocked, hence it runs into a timeout.
Why is it blocked?

You create 500 (100 in your update) tasks that all publish a message and then call the blocking method WaitForConfirms. Without modifying the max threads of the pool this means you have 500 (100) threadpool threads blocked in the WaitForConfirms method, so that no other task (including for example the heartbeat task) can run anymore, as no threads are available to run on. => Runs into timeout exception.

This is nice but why does it then work with the AsyncBarrier?
=> This awaits a task before the blocking call to WaitForConfirms is done, hence only at max 1 Task / Thread is in the blocking wait.

[lukebakken]

Hello everyone who has commented here -

If you have time to test version 7 alphas, it would be really appreciated

https://www.nuget.org/packages/RabbitMQ.Client/7.0.0-alpha.6

I'm hoping to get the final release done this month, but only if there is enough community testing (with reports back!)