Implement asynchronous methods.

[lukebakken]

Fixes:

• Port "Implement Full Async Channel" PR to main #1308
• Xunit parallel test execution #1412

Related to:

• Implement QueueDeclareAsync #1345
• Port "Implement Full Async Channel" PR to main #1308
• Proposal: Full async channel interface (IModel) #970
• AMQP 0-9-1 Channel (IModel) API with async methods #843

[stebet]

It's awesome seeing these trickle in. Look @davidfowl, we're going full async with Channels and Pipelines :D

[lukebakken]

@stebet I'd rather not keep a long-running branch. I know there's much more to do for version 7, but I would like to get this into main.

So, if you could give it a quick review, I'll merge it and move on to the next item, which will be reviewing all of the code doc blocks to make sure they are complete and accurate.

[lukebakken]

@stebet @michaelklishin

I am planning to merge this soon and continue work on version 7 via other PRs to main. I don't want to have long-running feature branches.

One serious issue is that Test.Integration fails when I enable Xunit's parallel test execution, which to me sounds like there are some subtle bugs in the code (either in the new code, or pre-existing). I'm planning to fix that issue next. I have moved all tests that must execute sequentially (because they stop/start RabbitMQ itself, for instance) to Test.SequentialIntegration.

[lukebakken]

@stebet I figured out why tests were failing when Xunit's parallelization was enabled. There is something incorrect with how ArrayPool<byte>.Shared is being used. If I disable the use of that class, and just allocate new byte arrays, everything works great:

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/lukebakken/async-operations/projects/RabbitMQ.Client/util/ClientArrayPool.cs#L39

The errors I see on the RabbitMQ side are all framing related. "Invalid frame end marker" is a common error, which suggests that the written frame's size is too large, or has corrupt data.

The fact that rented arrays can be larger than the requested size is a big hint, so I'm focusing on that.

[paulomorgado]

@stebet I figured out why tests were failing when Xunit's parallelization was enabled. There is something incorrect with how ArrayPool<byte>.Shared is being used. If I disable the use of that class, and just allocate new byte arrays, everything works great:

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/lukebakken/async-operations/projects/RabbitMQ.Client/util/ClientArrayPool.cs#L39

The errors I see on the RabbitMQ side are all framing related. "Invalid frame end marker" is a common error, which suggests that the written frame's size is too large, or has corrupt data.

The fact that rented arrays can be larger than the requested size is a big hint, so I'm focusing on that.

Rented arrays should only used with Span or Memory to avoid those issues.

Or ArraySegment<T>,

[lukebakken]

Hi @paulomorgado! Thanks for taking a look.

Is that documented anywhere?

@stebet here is what I think the problem is -

We get a rented array here, for instance -

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/lukebakken/async-operations/projects/RabbitMQ.Client/client/impl/Frame.cs#L157-L162

Then, we return the rented array here -

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/lukebakken/async-operations/projects/RabbitMQ.Client/client/impl/SocketFrameHandler.cs#L305-L313

But, in the case where the rented array is larger than the requested size, the array being returned won't have the same size, since it has been wrapped and unwrapped by ReadOnlyMemory. Thus, the returned array may corrupt the ArrayPool somehow.

That's my theory, at least.

[lukebakken]

@paulomorgado @stebet so much for my theory, this commit does not fix the issue when ArrayPool is used - f823dc7

@paulomorgado we are using ReadOnlyMemory to "capture" the size of the frame to be written, but somehow when ArrayPool is used, there are framing issues.

Additional theories or suggestions are very welcome!

[paulomorgado]

@lukebakken,

It's kind of common sense.

ArraySegment is in .NET Framework since version 2,0, but it's not common knowledge.

For .NET 6+, I'd go with Span/ReadOnlySpan or Memory/ReadOnlyMemory, because the BCL and the runtime know about it.

But they all can wrap slice and warp an array without heap allocations.

With spans, can even allocate small array in the stack.

https://github.com/rabbitmq/rabbitmq-dotnet-client/blob/lukebakken/async-operations/projects/RabbitMQ.Client/client/impl/Frame.cs#L157-L162 is already using ReadOnlyMemory.

I'm a bit concerned about renting arrays in one place and returning in another. One can loose track of those and loose the benefits.

You might want to look at MemoryPool.

I never used it myself, but you can check wher .NET is using it; https://source.dot.net/#System.Memory/System/Buffers/MemoryPool.cs,790a0fe9f42069bf,references

[paulomorgado]

@paulomorgado @stebet so much for my theory, this commit does not fix the issue when ArrayPool is used - f823dc7

@paulomorgado we are using ReadOnlyMemory to "capture" the size of the frame to be written, but somehow when ArrayPool is used, there are framing issues.

Additional theories or suggestions are very welcome!

If the array is returned and something is still holding on to it, it will be rented again and will be used by more than one "client".

That's why you need to keep track of the rented arrays at all times and not let them get out of your sight.

[lukebakken]

@paulomorgado thank you for your input. I am not as familiar with .NET / C# as many of the other contributors to this library so I appreciate it.

These two statements lead me to the source of the problem:

I'm a bit concerned about renting arrays in one place and returning in another. One can loose track of those and loose the benefits.
If the array is returned and something is still holding on to it, it will be rented again and will be used by more than one "client".

I added code to the ClientArrayPool class to keep track of rentals / returns and figured out where the issue was.

[lukebakken]

Thank you @danielmarbach for your review and continued interest in this library.

[danielmarbach]

You're welcome. I might be able to have a more thorough look at some point. These are only quick observations from scrolling through the changeset

[lukebakken]

@danielmarbach I don't plan on releasing version 7 without your and several other key contributors' collective thumbs-up 👍

[paulomorgado]

I think this PR is too big.

If everything is functionally working, it should be merged and issues open for improvements,

[lukebakken]

I think this PR is too big. If everything is functionally working, it should be merged and issues open for improvements

@paulomorgado yep, I said as much two weeks ago. I am going to give @Gsantomaggio a tour of these changes tomorrow, merge this PR, and move on to other version 7 issues that I have filed.

[lukebakken]

Thanks for stopping by @Tornhoof

[plewam]

Hi @lukebakken,
I am switching my dev environment to version 7.0.0. Has the batch message functionality been removed?
IBasicPublishBatch batchMessage = this.Channel.CreateBasicPublishBatch();

[lukebakken]

@plewam, yes that API was removed here: #1028

If you would like to submit a pull request against the main branch, with tests, to restore that API's function, please do so and it will be considered.

[plewam]

@lukebakken Thats completely fine as it is, have just been using this functionality but if it works with the same performance through single message publish I am happy