Task scheduler: when awaiting in a Task.Factory.StartNew, is the thread returned to the pool?

user_376409 0
Viewed 1354

I'm implementing a worker engine with an upper limit to concurrency. I'm using a semaphore to wait until concurrency drops below the maximum, then use Task.Factory.StartNew to wrap the async handler in a try/catch, with a finally which releases the semaphore.

I realise this creates threads on the thread pool - but my question is, when one of those task-running threads actually awaits (on a real IO call or wait handle), is the thread returned to the pool, as I'd hope it would be?

If there's a better way to implement a task scheduler with limited concurrency where the work handler is an async method (returns Task), I'd love to hear it too. Or, let's say ideally, if there's a way to queue up an async method (again, it's a Task-returning async method) that feels less dodgy than wrapping it in a synchronous delegate and passing it to Task.Factory.StartNew, that would seem perfect..?

(This also makes me think that there are two kinds of parallelism here: how many tasks are being processed overall, but also how many continuations are running on different threads concurrently. Might be cool to have configurable options for both, though not a fixed requirement..)

Edit: snippet:

                    concurrencySemaphore.Wait(cancelToken);
                    deferRelease = false;
                    try
                    {
                        var result = GetWorkItem();
                        if (result == null)
                        { // no work, wait for new work or exit signal
                            signal = WaitHandle.WaitAny(signals);
                            continue;
                        }

                        deferRelease = true;
                        tasks.Add(Task.Factory.StartNew(() =>
                        {
                            try
                            {
                                DoWorkHereAsync(result); // guess I'd think to .GetAwaiter().GetResult() here.. not run this yet
                            }
                            finally
                            {
                                concurrencySemaphore.Release();
                            }
                        }, cancelToken));
                    }
                    finally
                    {
                        if (!deferRelease)
                        {
                            concurrencySemaphore.Release();
                        }
                    }
task-parallel-library task asynchronous .net c#
2 Answers

You can think that thread is returned to a ThreadPool even thought it is not actauly a return. The thread simply picks next queued item when async operation starts.

I would suggest you to look at Task.Run instead of Task.Factory.StartNew Task.Run vs Task.Factory.StartNew.

And also have a look at TPL DataFlow. I think it will match your requirements.

user_21389590

Here an example of a TaskWorker, that will not produce countless worker threads.

The magic is done by awaiting SemaphoreSlim.WaitAsync() which is an IO task (and there is no thread).

class TaskWorker
{
    private readonly SemaphoreSlim _semaphore;

    public TaskWorker(int maxDegreeOfParallelism)
    {
        if (maxDegreeOfParallelism <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism));
        }

        _semaphore = new SemaphoreSlim(maxDegreeOfParallelism, maxDegreeOfParallelism);
    }

    public async Task RunAsync(Func<Task> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        // No ConfigureAwait(false) here to keep the SyncContext if any
        // for the real task
        await _semaphore.WaitAsync(cancellationToken);
        try
        {
            await taskFactory().ConfigureAwait(false);
        }
        finally
        {
            _semaphore.Release(1);
        }
    }

    public async Task<T> RunAsync<T>(Func<Task<T>> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        await _semaphore.WaitAsync(cancellationToken);
        try
        {
            return await taskFactory().ConfigureAwait(false);
        }
        finally
        {
            _semaphore.Release(1);
        }
    }
}

and a simple console app to test

class Program
{
    static void Main(string[] args)
    {
        var worker = new TaskWorker(1);
        var cts = new CancellationTokenSource();
        var token = cts.Token;

        var tasks = Enumerable.Range(1, 10)
            .Select(e => worker.RunAsync(() => SomeWorkAsync(e, token), token))
            .ToArray();

        Task.WhenAll(tasks).GetAwaiter().GetResult();
    }

    static async Task SomeWorkAsync(int id, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Some Started {id}");
        await Task.Delay(2000, cancellationToken).ConfigureAwait(false);
        Console.WriteLine($"Some Finished {id}");
    }
}

Update

TaskWorker implementing IDisposable

class TaskWorker : IDisposable
{
    private readonly CancellationTokenSource _cts = new CancellationTokenSource();
    private readonly SemaphoreSlim _semaphore;
    private readonly int _maxDegreeOfParallelism;

    public TaskWorker(int maxDegreeOfParallelism)
    {
        if (maxDegreeOfParallelism <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism));
        }

        _maxDegreeOfParallelism = maxDegreeOfParallelism;
        _semaphore = new SemaphoreSlim(maxDegreeOfParallelism, maxDegreeOfParallelism);
    }

    public async Task RunAsync(Func<Task> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        ThrowIfDisposed();

        using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token))
        {
            // No ConfigureAwait(false) here to keep the SyncContext if any
            // for the real task
            await _semaphore.WaitAsync(cts.Token);
            try
            {
                await taskFactory().ConfigureAwait(false);
            }
            finally
            {
                _semaphore.Release(1);
            }
        }
    }

    public async Task<T> RunAsync<T>(Func<Task<T>> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        ThrowIfDisposed();

        using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token))
        {
            await _semaphore.WaitAsync(cts.Token);
            try
            {
                return await taskFactory().ConfigureAwait(false);
            }
            finally
            {
                _semaphore.Release(1);
            }
        }
    }

    private void ThrowIfDisposed()
    {
        if (disposedValue)
        {
            throw new ObjectDisposedException(this.GetType().FullName);
        }
    }

    #region IDisposable Support
    private bool disposedValue = false;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                _cts.Cancel();
                // consume all semaphore slots
                for (int i = 0; i < _maxDegreeOfParallelism; i++)
                {
                    _semaphore.WaitAsync().GetAwaiter().GetResult();
                }
                _semaphore.Dispose();
                _cts.Dispose();
            }
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        Dispose(true);
    }
    #endregion
}
user_17441640
Related