Эффективность BlockingCollection (T)

Некоторое время в моей компании мы использовали домашнюю версию ObjectPool<T>, которая обеспечивает блокировку доступа к ее содержимому. Это довольно просто: a Queue<T>, a object для блокировки и AutoResetEvent для подачи потока "заимствования" при добавлении элемента.

Мясо класса действительно эти два метода:

public T Borrow() {
    lock (_queueLock) {
        if (_queue.Count > 0)
            return _queue.Dequeue();
    }

    _objectAvailableEvent.WaitOne();

    return Borrow();
}

public void Return(T obj) {
    lock (_queueLock) {
        _queue.Enqueue(obj);
    }

    _objectAvailableEvent.Set();
}

Мы использовали этот и несколько других классов коллекций вместо тех, которые были предоставлены System.Collections.Concurrent, потому что мы используем .NET 3.5, а не 4.0. Но недавно мы обнаружили, что, поскольку мы используем Reactive Extensions, у нас фактически есть пространство имен Concurrent, доступное нам (в System.Threading.dll).

Естественно, я понял, что поскольку BlockingCollection<T> является одним из основных классов в пространстве имен Concurrent, он, вероятно, будет предлагать лучшую производительность чем что-либо, что написал я или мои товарищи по команде.

Поэтому я попытался написать новую реализацию, которая работает очень просто:

public T Borrow() {
    return _blockingCollection.Take();
}

public void Return(T obj) {
    _blockingCollection.Add(obj);
}

К моему удивлению, согласно некоторым простым тестам (заимствование/возврат в пул несколько тысяч раз из нескольких потоков), наша первоначальная реализация значительно превосходит BlockingCollection<T> с точки зрения производительности. Они оба работают правильно; это просто то, что наша первоначальная реализация кажется намного быстрее.

Мой вопрос:

Зачем это было? Возможно ли это потому, что BlockingCollection<T> обеспечивает большую гибкость (я понимаю, что это работает, обертывая IProducerConsumerCollection<T>), что обязательно приводит к издержкам производительности?
Это просто ошибочное использование класса BlockingCollection<T>?
Если это подходящее использование BlockingCollection<T>, я просто не правильно использую? Например, подход Take/Add слишком упрощен, и существует гораздо более эффективный способ получить одну и ту же функциональность?

Если у кого-то есть какое-то понимание в ответ на этот третий вопрос, похоже, что мы будем придерживаться нашей первоначальной реализации на данный момент.

Ответ 1

Здесь есть пара потенциальных возможностей.

Во-первых, BlockingCollection<T> в Reactive Extensions является backport, а не точно такой же, как окончательная версия .NET 4. Я не удивлюсь, если производительность этого backport отличается от RTF.NET 4 (хотя я не профилировал эту коллекцию, в частности). Большая часть TPL лучше работает в .NET 4, чем в backport. 3.5.

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

Однако BlockingCollection<T> предназначен для того, чтобы многие потоки производителей могли "выдавать" данные очень хорошо. Это не будет хорошо работать с вашей реализацией, так как конфликт блокировки начнет становиться проблематичным довольно быстро.

Сказав это, я также хотел бы указать на одно заблуждение:

... он, вероятно, будет предлагать лучшую производительность, чем все, что написал я или мои товарищи по команде.

Это часто неверно. Классы сбора фреймворка обычно работают очень хорошо, но часто не являются наиболее эффективными для данного сценария. Это, как говорится, они имеют тенденцию работать хорошо, будучи очень гибкими и очень надежными. Они часто имеют тенденцию масштабироваться очень хорошо. "Домашние" классы коллекций часто превосходят коллекции фреймов в конкретных сценариях, но, как правило, они проблематичны при использовании в сценариях за пределами того, для которого они были специально разработаны. Я подозреваю, что это одна из таких ситуаций.

Ответ 2

Я попробовал BlockingCollection против комманды ConurrentQueue/AutoResetEvent (аналогично решению OP, но без блокировки) в .Net 4, и последнее комбо было намного быстрее для моего варианта использования, и я удалил BlockingCollection. К сожалению, это было почти год назад, и я не смог найти результаты тестов.

Использование отдельного AutoResetEvent не делает вещи слишком сложными. Фактически, можно даже отвлечь его, раз и навсегда, на BlockingCollectionSlim....

BlockingCollection также полагается на ConcurrentQueue, но делает несколько дополнительных манипуляций с тонкими семафорами и маркерами отмены, что дает дополнительные возможности, но по стоимости, даже если они не используются. Следует также отметить, что BlockingCollection не женат на ConcurrentQueue, но может также использоваться вместе с другими разработчиками IProducerConsumerCollection.

A (проверенный, но еще не ожесточенный) неограниченные голые кости. 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 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);