уникальное исключение для BlockingCollection в .net 4.0

Question

Я использую BlockingCollection для шаблона Producer Consumer, и я получил исключение, я думаю написать патент на него - только два результата в Google! Исключение составляет «CompleteAdding не может использоваться одновременно с добавлениями в коллекцию», и это происходит, когда я TryAdd на th BlockingCollection следующим образом:

 public void EnqueueTask(T item)
    {
        if (!_cancellationTokenSource.IsCancellationRequested)
        {
            _workerQueue.Add(item);
        }
    }

CompleteAdding вызывается при утилизации класса оболочки Consumer-Producer:

  public void Dispose()
    {
        if (!_IsActive)
            return;
        _IsActive = false;
        _cancellationTokenSource.Cancel();
        _workerQueue.CompleteAdding();
        // Wait for the consumer's thread to finish.
        for (int i = 0; i < _workers.Length; ++i)
        {
            Task t1 = Task.Factory.StartNew(() =>
            {
                try
                {
                    if (!_workers[i].Join(4000))
                        LogWriter.Trace("Failed to join thread", "ThreadFailureOnDispose");
                }
                catch (Exception ex)
                {
                    OnLogged(ex.Message + ex.StackTrace);
                }
            });

        }


        // Release any OS resources.
    }

Кто-нибудь из Microsoft понял? я должен спать после отмены и перед вызовом CompleteAdding?

Answer 1

Посмотрите на этот фрагмент кода:

    for (int i = 0; i < _workers.Length; ++i)
    {
        Task t1 = Task.Factory.StartNew(() =>
        {
            try
            {
                if (!_workers[i].Join(4000))   << == Here
                    LogWriter.Trace("Failed to join thread", "ThreadFailureOnDispose");
            }

В _workers[i].Join(4000) значение i не то, что вы думаете.Попробуйте еще раз:

   for (int i = 0; i < _workers.Length; ++i)
    {
        int j = i;  // copy
        Task t1 = Task.Factory.StartNew(() =>
        {
            try
            {
                if (!_workers[j].Join(4000))  // j
                    LogWriter.Trace("Failed to join thread", "ThreadFailureOnDispose");
            }

В вашей версии перехватывается переменная 'i', и все задачи используют одну и ту же переменную.Все, кроме первых, увидят i == _workers.Length, потому что они выполняются после завершения цикла for.

Это классическая лямбда + захваченная задача var.