Производителност на BlockingCollection(T).

Тук има няколко потенциални възможности.

Първо, BlockingCollection<T> в Reactive Extensions е backport и не е точно същото като окончателната версия на .NET 4. Не бих се изненадал, ако производителността на този бекпорт се различава от .NET 4 RTM (въпреки че не съм профилирал специално тази колекция). Голяма част от TPL се представя по-добре в .NET 4, отколкото в .NET 3.5 backport.

Като се има предвид това, подозирам, че вашата реализация ще надхвърли производителността BlockingCollection<T>, ако имате една нишка на производител и една нишка на потребител. С един производител и един потребител вашето заключване ще има по-малко въздействие върху общата производителност, а събитието за нулиране е много ефективно средство за изчакване от страна на потребителя.

Въпреки това, BlockingCollection<T> е проектиран да позволи на много нишки на производителите да "поставят" данни много добре. Това няма да се представи добре с вашето внедряване, тъй като спорът за заключване ще започне да става проблематичен доста бързо.

Като се има предвид това, бих искал също да посоча едно погрешно схващане тук:

...вероятно ще предложи по-добра производителност от всичко, което аз или моите съотборници сме писали.

Това често не е вярно. Класовете за колекция от рамки обикновено се представят много добре, но често не са най-производителната опция за даден сценарий. Като се има предвид това, те са склонни да се представят добре, като същевременно са много гъвкави и много здрави. Те често са склонни да се мащабират много добре. „Домашно написаните“ класове за събиране често превъзхождат рамковите колекции в конкретни сценарии, но са склонни да бъдат проблематични, когато се използват в сценарии извън този, за който са специално проектирани. Подозирам, че това е една от тези ситуации.

Опитах BlockingCollection срещу ConurrentQueue/AutoResetEvent комбо (подобно на решението на OP, но без заключване) в .Net 4 и последното комбо беше толкова много по-бързо за моя случай на употреба, че се отказах от BlockingCollection. За съжаление това беше преди почти година и не можах да намеря резултатите от бенчмарка.

Използването на отделно AutoResetEvent не прави нещата много по-сложни. Всъщност човек може дори да го абстрахира, веднъж завинаги, в BlockingCollectionSlim....

BlockingCollection вътрешно разчита и на ConcurrentQueue, но прави малко допълнително жонглиране с тънки семафори и токени за анулиране, което дава допълнителни функции, но на цена, дори когато не се използва. Трябва също да се отбележи, че BlockingCollection не е женен за ConcurrentQueue, но може да се използва и с други имплементатори на IProducerConsumerCollection вместо това.

Неограничена, доста гола реализация на BlockingCollectionSlim:

class BlockingCollectionSlim<T>
{
    private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
    private readonly AutoResetEvent _autoResetEvent = new AutoResetEvent(false);
    public void Add(T item)
    {
        _queue.Enqueue(item);
        _autoResetEvent.Set();
    }
    public bool TryPeek(out T result)
    {
        return _queue.TryPeek(out result);
    }
    public T Take()
    {
        T item;
        while (!_queue.TryDequeue(out item))
            _autoResetEvent.WaitOne();
        return item;
    }
    public bool TryTake(out T item, TimeSpan patience)
    {
        if (_queue.TryDequeue(out item))
            return true;
        var stopwatch = Stopwatch.StartNew();
        while (stopwatch.Elapsed < patience)
        {
            if (_queue.TryDequeue(out item))
                return true;
            var patienceLeft = (patience - stopwatch.Elapsed);
            if (patienceLeft <= TimeSpan.Zero)
                break;
            else if (patienceLeft < MinWait)
            // otherwise the while loop will degenerate into a busy loop,
            // for the last millisecond before patience runs out
                patienceLeft = MinWait;
            _autoResetEvent.WaitOne(patienceLeft);
        }
        return false;
    }
    private static readonly TimeSpan MinWait = TimeSpan.FromMilliseconds(1);

Попаднах на същите проблеми с производителността с BlockingCollection в .Net 4.7.2 и намерих тази публикация. Моят случай е MultipleProducers-MultipleConsumers, по-специално малки парчета данни се четат от много източници и трябва да се обработват от много филтри. Бяха използвани няколко (Env.ProcessorCount) BlockingCollections и завърших с профилиращ инструмент за производителност, който ми каза, че BlockingCollection.GetConsumingEnumerable.MoveNext() изяжда повече процесорно време от действителното филтриране!

Благодаря ти, @Eugene Beresovsky, за твоя код. FYI: В моята среда беше почти два пъти по-бавен от BlockingCollection. И така, ето моята SpinLocked BlockingCollection:

public class BlockingCollectionSpin<T>
{
    private SpinLock _lock = new SpinLock(false);
    private Queue<T> _queue = new Queue<T>();

    public void Add(T item)
    {
        bool gotLock = false;
        try
        {
            _lock.Enter(ref gotLock);
            _queue.Enqueue(item);
        }
        finally
        {
            if (gotLock) _lock.Exit(false);
        }
    }

    public bool TryPeek(out T result)
    {
        bool gotLock = false;
        try
        {
            _lock.Enter(ref gotLock);
            if (_queue.Count > 0)
            {
                result = _queue.Peek();
                return true;
            }
            else
            {
                result = default(T);
                return false;
            }
        }
        finally
        {
            if (gotLock) _lock.Exit(false);
        }
    }

    public T Take()
    {
        var spin = new SpinWait();
        do
        {
            bool gotLock = false;
            try
            {
                _lock.Enter(ref gotLock);
                if (_queue.Count > 0)
                    return _queue.Dequeue();
            }
            finally
            {
                if (gotLock) _lock.Exit(false);
            }
            spin.SpinOnce();
        } while (true);
    }
}

А за критичен за производителността код бих предложил да се избягва readonly модификатор на поле. Той добавя проверка на всеки достъп до поле в IL. Със следния тестов код

private static void TestBlockingCollections()
{
    const int workAmount = 10000000;
    var workerCount = Environment.ProcessorCount * 2;
    var sw = new Stopwatch();
    var source = new long[workAmount];
    var rnd = new Random();
    for (int i = 0; i < workAmount; i++)
        source[i] = rnd.Next(1000000);

    var swOverhead = 0.0;
    for (int i = 0; i < workAmount; i++)
    {
        sw.Restart();
        swOverhead += sw.Elapsed.TotalMilliseconds;
    }
    swOverhead /= workAmount;

    var sum1 = new long[workerCount];
    var queue1 = new BlockingCollection<long>(10000);
    var workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        foreach (var l in queue1.GetConsumingEnumerable())
            sum1[n] += l;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue1.Add(l);
    queue1.CompleteAdding();
    Task.WaitAll(workers);
    var elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollection {0:F4}ms", elapsed / workAmount);

    var sum2 = new long[workerCount];
    var queue2 = new BlockingCollectionSlim<long?>();
    workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        long? l;
        while ((l = queue2.Take()).HasValue)
            sum2[n] += l.Value;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue2.Add(l);
    for (int i = 0; i < workerCount; i++)
        queue2.Add(null);
    Task.WaitAll(workers);
    elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollectionSlim {0:F4}ms", elapsed / workAmount);

    var sum3 = new long[workerCount];
    var queue3 = new BlockingCollectionSpin<long?>();
    workers = Enumerable.Range(0, workerCount - 1).Select(n =>
    Task.Factory.StartNew(() =>
    {
        long? l;
        while ((l = queue3.Take()).HasValue)
            sum3[n] += l.Value;
    })).ToArray();

    Thread.Sleep(1000);

    sw.Restart();
    foreach (var l in source)
        queue3.Add(l);
    for (int i = 0; i < workerCount; i++)
        queue3.Add(null);
    Task.WaitAll(workers);
    elapsed = sw.Elapsed.TotalMilliseconds - swOverhead;
    Console.WriteLine("BlockingCollectionSpin {0:F4}ms", elapsed/workAmount);

    if (sum1.Sum() != sum2.Sum() || sum2.Sum() != sum3.Sum())
        Console.WriteLine("Wrong sum in the end!");

    Console.ReadLine();
}

На Core i5-3210M с 2 ядра и активиран HT имам следния изход:

BlockingCollection     0.0006ms
BlockingCollectionSlim 0.0010ms (Eugene Beresovsky implementation)
BlockingCollectionSpin 0.0003ms

И така, SpinLocked версията е два пъти по-бърза от .Net BlockingCollection. Но бих препоръчал да използвате само него! ако наистина предпочитате производителност срещу простота на кода (и поддръжка).