Параллельный C #: стоит ли использовать много AutoResetEvent?

Поскольку все клиенты должны ожидать выполнения задания в одном потоке, нет реальной необходимости в использовании очереди. Поэтому я предлагаю вместо этого использовать класс Monitor, и, в частности, функции Wait/Pulse. Хотя это немного низкоуровнево и многословно.

class Worker<TResult> : IDisposable
{
    private readonly object _outerLock = new object();
    private readonly object _innerLock = new object();
    private Func<TResult> _currentJob;
    private TResult _currentResult;
    private Exception _currentException;
    private bool _disposed;

    public Worker()
    {
        var thread = new Thread(MainLoop);
        thread.IsBackground = true;
        thread.Start();
    }

    private void MainLoop()
    {
        lock (_innerLock)
        {
            while (true)
            {
                Monitor.Wait(_innerLock); // Wait for client requests
                if (_disposed) break;
                try
                {
                    _currentResult = _currentJob.Invoke();
                    _currentException = null;
                }
                catch (Exception ex)
                {
                    _currentException = ex;
                    _currentResult = default;
                }
                Monitor.Pulse(_innerLock); // Notify the waiting client that the job is done
            }
        } // We are done
    }

    public TResult DoWork(Func<TResult> job)
    {
        TResult result;
        Exception exception;
        lock (_outerLock) // Accept only one client at a time
        {
            lock (_innerLock) // Acquire inner lock
            {
                if (_disposed) throw new InvalidOperationException();
                _currentJob = job;
                Monitor.Pulse(_innerLock); // Notify worker thread about the new job
                Monitor.Wait(_innerLock); // Wait for worker thread to process the job
                result = _currentResult;
                exception = _currentException;
                // Clean up
                _currentJob = null;
                _currentResult = default;
                _currentException = null;
            }
        }
        // Throw the exception, if occurred, preserving the stack trace
        if (exception != null) ExceptionDispatchInfo.Capture(exception).Throw();
        return result;
    }

    public void Dispose()
    {
        lock (_outerLock)
        {
            lock (_innerLock)
            {
                _disposed = true;
                Monitor.Pulse(_innerLock); // Notify worker thread to exit loop
            }
        }
    }
}

Пример использования:

var worker = new Worker<int>();
int result = worker.DoWork(() => 1); // Accepts a function as argument
Console.WriteLine($"Result: {result}");
worker.Dispose();

Выход:

Result: 1

Обновление. Предыдущее решение не поддерживает ожидание, поэтому вот решение, которое допускает надлежащее ожидание. Он использует TaskCompletionSource для каждого задания, хранящегося в BlockingCollection.

class Worker<TResult> : IDisposable
{
    private BlockingCollection<TaskCompletionSource<TResult>> _blockingCollection
        = new BlockingCollection<TaskCompletionSource<TResult>>();

    public Worker()
    {
        var thread = new Thread(MainLoop);
        thread.IsBackground = true;
        thread.Start();
    }

    private void MainLoop()
    {
        foreach (var tcs in _blockingCollection.GetConsumingEnumerable())
        {
            var job = (Func<TResult>)tcs.Task.AsyncState;
            try
            {
                var result = job.Invoke();
                tcs.SetResult(result);
            }
            catch (Exception ex)
            {
                tcs.TrySetException(ex);
            }
        }
    }

    public Task<TResult> DoWorkAsync(Func<TResult> job)
    {
        var tcs = new TaskCompletionSource<TResult>(job,
            TaskCreationOptions.RunContinuationsAsynchronously);
        _blockingCollection.Add(tcs);
        return tcs.Task;
    }

    public TResult DoWork(Func<TResult> job) // Synchronous call
    {
        var task = DoWorkAsync(job);
        try { task.Wait(); } catch { } // Swallow the AggregateException
        // Throw the original exception, if occurred, preserving the stack trace
        if (task.IsFaulted) ExceptionDispatchInfo.Capture(task.Exception.InnerException).Throw();
        return task.Result;
    }

    public void Dispose()
    {
        _blockingCollection.CompleteAdding();
    }
}

Пример использования

var worker = new Worker<int>();
int result = await worker.DoWorkAsync(() => 1); // Accepts a function as argument
Console.WriteLine($"Result: {result}");
worker.Dispose();

Выход:

Result: 1

С точки зрения синхронизации это работает нормально.

Но кажется бесполезным делать это таким образом. Если вы хотите выполнять задания одно за другим, вы можете просто использовать блокировку:

lock (lockObject) {
  RunJob();
}

Каковы ваши намерения с этим кодом?

Также возникает вопрос эффективности, потому что каждая задача создает событие ОС и ожидает его. Если вы используете более современный TaskCompletionSource, он будет использовать то же самое под капотом, если вы синхронно ожидаете этой задачи. Вы можете использовать асинхронное ожидание (await myTCS.Task;), чтобы немного повысить эффективность. Конечно, это заражает весь стек вызовов с помощью async/await. Если это операция с довольно небольшим объемом, вы не выиграете много.

В целом, я думаю, это сработает, хотя, когда вы говорите, что "многие" потоки вызывают Do(), это может плохо масштабироваться... приостановленные потоки используют ресурсы.

Другая проблема с этим кодом заключается в том, что во время простоя у вас будет «жесткий цикл» в «workerThread», который заставит ваше приложение возвращать время высокой загрузки ЦП. Вы можете добавить этот код в "workerThread":

if (concurrentQueue.IsEmpty) Thread.Sleep(1);

Вы также можете ввести тайм-аут для вызова WaitOne, чтобы избежать застревания в журнале.