Add cancellation token during subscribe

[Gsantomaggio]

Fixes #205

In this way, the thread is not locked, and the:

new MessageContext(message.MessageOffset, TimeSpan.FromMilliseconds(chunk.Timestamp)),
                            message).Wait(_token);

is in wait(...), so there is not the thread pool exhaustion.

@ricardSiliuk @ricsiLT it should not change anything. Please feel free to add any comments.

@jonnepmyra do you have a chance to test it?

Signed-off-by: Gabriele Santomaggio G.santomaggio@gmail.com

[codecov]

Codecov Report

Base: 92.17% // Head: 91.98% // Decreases project coverage by -0.19% ⚠️

Coverage data is based on head (1944a18) compared to base (ce2e847).
Patch coverage: 65.38% of modified lines in pull request are covered.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #206      +/-   ##
==========================================
- Coverage   92.17%   91.98%   -0.20%     
==========================================
  Files          92       92              
  Lines        7966     7985      +19     
  Branches      640      641       +1     
==========================================
+ Hits         7343     7345       +2     
- Misses        492      507      +15     
- Partials      131      133       +2     

Impacted Files	Coverage Δ
RabbitMQ.Stream.Client/RawConsumer.cs	84.16% <65.38%> (-4.46%)	⬇️
RabbitMQ.Stream.Client/WireFormatting.cs	71.18% <0.00%> (-3.39%)	⬇️
RabbitMQ.Stream.Client/Reliable/ReliableBase.cs	96.47% <0.00%> (+1.17%)	⬆️

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

☔ View full report at Codecov.
📢 Do you have feedback about the report comment? Let us know in this issue.

[jonnepmyra]

Hi!

Thank your for the quick response and patch @Gsantomaggio

We've been testing it during the day with various workloads and it's definitely alot better!

At first we could still se about 0.5% of all closed connections get deadlocked with an error:
Error removing the consumer id: 0 from the server

... but after tweaking our consumer messagehandlers that contained TaskCompletionSources with TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); it works better and we're no longer able to reporduce the problem!!

The reason why I'm mentioning the TaskCompletionSource tweak in our application code is that even if it works well now, and we're super grateful for this being fixed, the code still seems to be able to deadlock under certain conditions but I might be wrong...

The Wait() in:

rabbitmq-stream-dotnet-client/RabbitMQ.Stream.Client/RawConsumer.cs

Lines 160 to 162 in 9d34170

	_config.MessageHandler(this,
	new MessageContext(message.MessageOffset, TimeSpan.FromMilliseconds(chunk.Timestamp)),
	message).Wait(_token);

feels like some sort of a flaw (as mentioned before in #106 (comment) ) but on the other hand, I've no better suggestion at the very moment if WaitAsync() leads to thread stravation.

Thanks again!

[Gsantomaggio]

@jonnepmyra let me clarify some points:

• We can't use WaitAsync since we will end up on Add GetAwaiter().GetResult(); on MessageHandler #184 (comment)

• We can't use aync Tasks here:

  rabbitmq-stream-dotnet-client/RabbitMQ.Stream.Client/RawConsumer.cs

  Lines 160 to 162 in 9d34170

  	_config.MessageHandler(this,
  	new MessageContext(message.MessageOffset, TimeSpan.FromMilliseconds(chunk.Timestamp)),
  	message).Wait(_token);

  
  Because with ref SequenceReader, it is not possible:

Parameters of type `SequenceReader<byte>` cannot be declared in async local functions

• We introduced SequenceReader<byte> for performance problems here: SequenceReader replaces the ReadOnlySequence to read the bytes from the socket/stream #180

• .Wait(_token); should not go in dead-lock (unless there is some blocking code in your consumer).

Have questions:

• code still seems to be able to deadlock

  • Have you tried without your consumer code to reproduce the lock?
  • What do you mean by deadlock? Do you still have a memory leak?
  • Have you tried the last commit 9d34170 ?

Thank you for trying the PR. :)

p.s. If you want, you can find me at https://rabbitmq-slack.herokuapp.com/

[jonnepmyra]

Hi!

Thanks for taking time to explain.

• Have you tried without your consumer code to reproduce the lock?

Yes, and I have not been able to reproduce it.

• What do you mean by deadlock? Do you still have a memory leak?

The Memoryleak seems to be fixed now, no connections left unclosed after tweaking application specific consumer code!

• Have you tried the last commit 9d34170 ?

Yes, it seems fine

Cheers!

[Gsantomaggio]

Great! We will merge it soon!
Ty

[Gsantomaggio]

Let's wait for some feedback from @ricardSiliuk @ricsiLT then, we can merge and release.

[Gsantomaggio]

I did some memory/threads test with millions of messages:

Task.Run(() =>
            {
                while (true)
                {
                    ThreadPool.GetMaxThreads(out var max, out _);
                    ThreadPool.GetAvailableThreads(out var available, out _);
                    var running = max - available;
                    Console.WriteLine(
                        $"{DateTime.Now:HH:mm:ss.fff} - Threads: {System.Diagnostics.Process.GetCurrentProcess().Threads.Count} " +
                        $"-  {running}");
                    Thread.Sleep(1000);
                }
            });
....



  MessageHandler = async (consumer, ctx, message) =>
                {
                    if (++consumed % 10_000 == 0)
                    {
                        var timeMilliseconds = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
                        Console.WriteLine(
                            $"Client Consumed {consumed} messages in {timeMilliseconds - startTimestamp} ms stream: {Stream}");
                    }

                    await Task.Delay(1);
                }

Threads are stable:

11:02:35.849 - Threads: 23 -  4
11:02:36.849 - Threads: 23 -  4
.....
11:05:27.923 - Threads: 23 -  4
11:05:28.924 - Threads: 23 -  4

Memory is stable: