Runtime: BufferBlock slow?

[electricessence]

Been doing some bench-marking and I thought a BufferBlock<T> would be a great underlying collection for an Object Pool. But it just seems to be dramatically slower than a ConcurrentQueue<T>.

Is this simply the nature of the beast? Or isn't there a way to boost performance?

The reason why I'm not just simply using ConcurrentQueue (because it's super fast), is that some instances of object construction and recycling can take extra time, and I'd prefer something like a Dataflow pipeline to handle it.

[stephentoub]

It's largely the "nature of the beast". ConcurrentQueue<T>, especially in .NET Core 2.0, is very fast, with no locks (unless it needs to expand the buffer), very few interlocks, minimized contention between producers and consumers, no interface calls, etc. BufferBlock<T> has all of that, in particular in order to play nicely with other dataflow blocks, and that adds non-trivial cost.

Depending on your needs, you might try Channel.CreateUnbounded<T> from the corefxlab repo (https://github.com/dotnet/corefxlab/tree/master/src/System.Threading.Tasks.Channels, https://github.com/dotnet/corefxlab/issues/1731). It doesn't integrate with dataflow, but as a result can get away with less expenses (and we're still working on improving it, help welcome), but still supports awaiting for items to be available.

[electricessence]

Thanks so much for the insight!

Channels sounds exactly like what I've been looking for! I originally was using Dataflow blocks as async P-C queues.

Will try it out and report back. :)

[electricessence]

So definitely Channel<T> is magnitudes faster. Even to the point that I removed the BufferBlock<T> from my tests because it was taking too long.

Here are the latest results:
https://github.com/electricessence/Open.Disposable.ObjectPools/blob/Channel-Experiment/benchmarking/BenchmarkResult.txt

You'll also notice at the end that it defeats an Optimistic-Array after size 1000 but is still much slower than a ConcurrentQueue<T>.

I like how the API is defined... But I have one criticism: The terminology could be confusing...

pool.Reader.TryRead(out T item);

This to me doesn't really scream "take". It seems more like "yes there's something here for you and you can read it." Just my thoughts. It actually made me worry that I wasn't owning the item afterwards.

[electricessence]

Is there an extension for TryReadAsync()? Seems pretty trivial to add one.

[electricessence]

And is there an ActionBlock<T> equivalent for a Channel<T>?

[stephentoub]

But I have one criticism: The terminology could be confusing

So you'd want TryAdd / TryTake instead of TryWrite / TryRead? And Adder/Taker instead of Writer/Reader? There's something natural I think about the read/write terminology, but I do understand the concern that read could potentially be viewed more as a peek than as a take. Might just need to be clear about it in docs.

And is there an ActionBlock equivalent for a Channel?

One of the key differences between the channels vs dataflow lib is whether execution is built in or not. The dataflow library integrates the storage and processing of that storage, whereas channels is really just the storage (and associated synchronization); any processing is left to the user. So the equivalent for ActionBlock<T> would just be a loop consuming data from the channel, e.g. pseudo-code
```C#
public static async Task ProcessAsync(ChannelReader reader, Action action)
{
while (await reader.WaitToReadAsync())
while (reader.TryRead(out T item))
action(item);
}

> Is there an extension for TryReadAsync()? Seems pretty trivial to add one.

Up until a week or so ago, the base channel had a virtual ReadAsync method.  We removed it out of concerns that it was baking in control flow around channel completion, but I still have mixed feelings about that and might push to put it back.  Regardless, TryReadAsync would be very similar, e.g. pseudo-code
```C#
public static async ValueTask<(bool, T)> TryReadAsync(this ChannelReader<T> reader)
{
    while (await reader.WaitToReadAsync())
    {
        if (reader.TryRead(out T item)) return (true, item);
    }
    return (false, default(T));
}

plus whatever optimizations you wanted if necessary.

[electricessence]

I totally get keeping it simple and not baking stuff in that isn't really core functionality. What you wrote was almost exactly what I was thinking. I've never seen the strange ValueTask syntax. Thanks for that as well.

Love the ProcessAsync method. Will use that in my Recycler.

So yes... I obviously wrestled with this recently with Object Pools because so many people have their own ideas of API and what makes sense...
But I definitely sided with TryTake and TryGive for object pools. These "Channels" _are_ P-C queues so there needs to be a notion of "ownership" IMO. The consumer is "taking" when reading. For write, I think it's hard to say. Again, as a consumer, I'm "giving" to that channel. But I think .writer.TryWrite(...) works because you know you're sending something there even if you hold a reference to it. You know what it did. "It's in there." TryRead on the other hand doesn't really scream "I've given it to you".

Just my thoughts. You could even change .Reader to .Available and .Writer to .Receiver...
Or .In and .Out... The list is long obviously.. I feel your pain. :P

[electricessence]

Again, thanks for the help. ;) Like what you've done. Is there an expected release date, etc?

[stephentoub]

Is there an expected release date, etc?

I don't have any firm dates, but it's in the works.

Thanks for the feedback.

[electricessence]

I will continue to look into your code. I'm very curious how to optimize an incoming buffer.

[svick]

Another possible option would be to use AsyncCollection from AsyncEx. It's not not part of .Net, but it is released (unlike Channels). It's basically an async version of BlockingCollection, so its speed should be comparable with ConcurrentQueue.

[stephentoub]

so its speed should be comparable with ConcurrentQueue

In terms of raw Enqueue/TryDequeue speed, that's unlikely.

[electricessence]

@svick ... It's interesting that you bring that up. I was just reviewing @StephenCleary 's code the other day. At this point I'm comfortable with ConcurrentQueue as it's very very fast. The problem with over-async (maybe @stephentoub can comment too) is that you could queue up a lot of tasks waiting to get access and the overall performance you were looking for will be negatively affected. It just depends...

[electricessence]

@stephentoub ... I think he's referring to the speed of ConcurrentQueue since the default underlying collection of AsyncCollection is ... wait for it... a ConcurrentQueue LOL. So you can decide when you want async operations to occur on it... which is nearly what I was referring to.

[StephenCleary]

Just chiming in... if ConcurrentQueue works for your needs, then go ahead and use it. You won't get much faster than that.

BlockingCollection provides an abstraction over a concurrent collection (defaulting to ConcurrentQueue) that allows producers/consumers to wait for a notfull/notempty signal. AsyncCollection provides (almost) the same abstraction over a concurrent collection (defaulting to ConcurrentQueue) that allows producers/consumers to asynchronously wait for a notfull/notempty signal.

If you don't need the signals, then you don't need that abstraction, and can just use ConcurrentQueue directly.

BufferBlock is a much more "batteries included" type of queue. Neither BufferBlock nor BlockingCollection / AsyncCollection attempt to compare performance-wise with ConcurrentQueue, but they bring their own different advantages.

[stephentoub]

I think he's referring to the speed of ConcurrentQueue since the default underlying collection of AsyncCollection is ... wait for it... a ConcurrentQueue

Yes, but as with any component A that layers functionality over B, A is going to be more expensive, e.g.
```C#
using Nito.AsyncEx;
using System;
using System.Collections.Concurrent;
using System.Diagnostics;

class Program
{
static void Main()
{
const int Iters = 1000000;
var sw = new Stopwatch();

    while (true)
    {
        var ac = new AsyncCollection<int>();
        sw.Restart();
        for (int i = 0; i < Iters; i++)
        {
            ac.Add(i);
            ac.Take();
        }
        sw.Stop();
        Console.WriteLine("AC: " + sw.Elapsed);

        var cq = new ConcurrentQueue<int>();
        sw.Restart();
        for (int i = 0; i < Iters; i++)
        {
            cq.Enqueue(i);
            cq.TryDequeue(out int _);
        }
        sw.Stop();
        Console.WriteLine("CQ: " + sw.Elapsed);

        Console.WriteLine();
    }
}

}

yields results like:

AC: 00:00:00.8890268
CQ: 00:00:00.0198012
```
This isn't a slight against AsyncCollection; anything that layered on such functionality would incur cost. I'm just highlighting why I wrote what I wrote.

[electricessence]

At both Stephens. Thank you for chiming in! :)