Parallel Task Queue with return value

Question

I have an implementation question centered around queuing long-running tasks that return a value. Because of the CPU usage in this task, I want to limit the number of tasks running at the same time to a constant number.

I'm returning an async task to the caller that it will await on, and hopefully the implementation will still let that happen.

This is the task that I would like to queue, and let the caller wait on:

public async Task<string> LongTask(string filename)
{
    return await Task.Run(() =>
    {
        //does something for awhile
        return "my result";
    });
}

This is how it is called from the caller, which will wait on the result:

string result = await LongTask("test");

Thanks in advance.

Possible duplicate of [Have a set of Tasks with only X running at a time](https://stackoverflow.com/questions/14075029/have-a-set-of-tasks-with-only-x-running-at-a-time) — Cory Nelson, Jan 26 '18 at 05:30
I did find that question previously, but it does not help me figure out how to add to a queue. — Joe, Jan 26 '18 at 15:57

David · Answer 1 · 2018-01-26T05:15:23.977

You can try below functions:

int nMaxConcurrentTasks = 5;
public void DoSomethingALotWithTasksThrottled()
{
    var listOfTasks = new List<Task>();
    for (int i = 0; i < 100; i++)
    {
        var count = i;
        // Note that we create the Task here, but do not start it.
        listOfTasks.Add(new Task(() => Something(count)));
    }
    Tasks.StartAndWaitAllThrottled(listOfTasks, nMaxConcurrentTasks); // 5 max
}

public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxActionsToRunInParallel, CancellationToken cancellationToken = new CancellationToken())
{
    StartAndWaitAllThrottled(tasksToRun, maxActionsToRunInParallel, -1, cancellationToken);
}


public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxActionsToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken())
{
    // Convert to a list of tasks so that we don't enumerate over it multiple times needlessly.
    var tasks = tasksToRun.ToList();

    using (var throttler = new SemaphoreSlim(maxActionsToRunInParallel))
    {
        var postTaskTasks = new List<Task>();

        // Have each task notify the throttler when it completes so that it decrements the number of tasks currently running.
        tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release())));

        // Start running each task.
        foreach (var task in tasks)
        {
            // Increment the number of tasks currently running and wait if too many are running.
            throttler.Wait(timeoutInMilliseconds, cancellationToken);

            cancellationToken.ThrowIfCancellationRequested();
            task.Start();
        }

        // Wait for all of the provided tasks to complete.
        // We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler&#39;s using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object.
        Task.WaitAll(postTaskTasks.ToArray(), cancellationToken);
    }
}

Original article here.

I'd like it to be a queue that I can routinely add to, this looks like I'd have to know all my tasks ahead of time? — Joe, Jan 26 '18 at 05:27

Parallel Task Queue with return value

1 Answers1