RabbitMq client hangs on consumer connection dispose

[ig-sinicyn]

RabbitMq.Client: 6.1.0 + net.core 3.1

We have a sample that was expected to measure consumer throughput but instead it hangs on consumer connection dispose.
Full code: RabbitMqHangsOnDispose.zip

Problem part:

private static async Task<string[]> ReceiveAsync(
    IConnectionFactory connectionFactory,
    int parallelCount,
    int expectedCount)
{
    string[] result;
    using (var connection = connectionFactory.CreateConnection())
    {
        using (var channel = connection.CreateModel())
        {
            channel.BasicQos(0, (ushort) parallelCount, false);

            var completed = new TaskCompletionSource<string[]>();
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (s, e) => ...
            var tag = channel.BasicConsume(...);
            result = await completed.Task;
            channel.BasicCancel(tag);

            Console.WriteLine("Before close channel");
        }

        Console.WriteLine("Before close connection");
    }                                                  // < hangs on Dispose() call

    Console.WriteLine("All closed - never called!");

    return result;
}

Expected behavior: connection disposal should complete successfully.

[michaelklishin]

We need a thread dump in order to suggest anything. This code looks like a short lived "one-off consumer" and this is not how this client is typically used. This is mailing list material at this point.

[ig-sinicyn]

Ok, reposted.

[ig-sinicyn]

Memory dump

[stebet]

I'll see if I can take a look at this tonight.

[lukebakken]

I can confirm the provided code reproduces the issue using 6.1.0 as well as the latest master code -

https://github.com/lukebakken/rabbitmq-dotnet-client-874

[stebet]

I found the issue in the example. The problems is this code:

using (var connection = connectionFactory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        channel.BasicQos(0, (ushort)parallelCount, false);

        var received = new ConcurrentQueue<string>();
        /*
        Here is the problem. A TaskCompletionSource is being created in a synchronous handler which is ok,
        but you have to force it to execute asynchronously, or it might up fighting with itself since it isn't awaiting
        anything else and will therefore very likely execute the continuation synchronously, causing a deadlock with
        other operations trying to get the "write lock" I'm guessing.
        */
        var completed = new TaskCompletionSource<string[]>(); //<-- This is the problem right here, replace with the line below and the awaited task will correctly execute asynchronously.
        //var completed = new TaskCompletionSource<string[]>(TaskCreationOptions.RunContinuationsAsynchronously);

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (s, e) =>
        {
            var body = Encoding.UTF8.GetString(e.Body.ToArray());
            received.Enqueue(body);
            channel.BasicAck(e.DeliveryTag, false);

            Console.WriteLine("Received:" + body);

            if (received.Count >= expectedCount)
            {
                completed.SetResult(received.ToArray());
            }
        };

        var tag = channel.BasicConsume(
            Queue,
            false,
            "",
            true,
            false,
            new Dictionary<string, object>(),
            consumer);

        result = await completed.Task;
        channel.BasicCancel(tag);

        Console.WriteLine("Before close channel");
    }

    Console.WriteLine("Before close connection");
}

[mdymel]

I was hitting the same issue and am also using TaskCompletionSource. Creating it with TaskCreationOptions.RunContinuationsAsynchronously seems to have fixed the problem.

Thanks a lot!

[michaelklishin]

Thank you, @stebet! I assume there is nothing for us to do in the client then.