Ограничение количества одновременных задач в .NET 4.5

Наблюдайте следующую функцию:

public Task RunInOrderAsync<TTaskSeed>(IEnumerable<TTaskSeed> taskSeedGenerator,
    CreateTaskDelegate<TTaskSeed> createTask,
    OnTaskErrorDelegate<TTaskSeed> onError = null,
    OnTaskSuccessDelegate<TTaskSeed> onSuccess = null) where TTaskSeed : class
{
    Action<Exception, TTaskSeed> onFailed = (exc, taskSeed) =>
    {
        if (onError != null)
        {
            onError(exc, taskSeed);
        }
    };

    Action<Task> onDone = t =>
    {
        var taskSeed = (TTaskSeed)t.AsyncState;
        if (t.Exception != null)
        {
            onFailed(t.Exception, taskSeed);
        }
        else if (onSuccess != null)
        {
            onSuccess(t, taskSeed);
        }
    };

    var enumerator = taskSeedGenerator.GetEnumerator();
    Task task = null;
    while (enumerator.MoveNext())
    {
        if (task == null)
        {
            try
            {
                task = createTask(enumerator.Current);
                Debug.Assert(ReferenceEquals(task.AsyncState, enumerator.Current));
            }
            catch (Exception exc)
            {
                onFailed(exc, enumerator.Current);
            }
        }
        else
        {
            task = task.ContinueWith((t, taskSeed) =>
            {
                onDone(t);
                var res = createTask((TTaskSeed)taskSeed);
                Debug.Assert(ReferenceEquals(res.AsyncState, taskSeed));
                return res;
            }, enumerator.Current).TaskUnwrap();
        }
    }

    if (task != null)
    {
        task = task.ContinueWith(onDone);
    }

    return task;
}

Где TaskUnwrap - это сохраняющая состояние версия стандарта Task.Unwrap:

public static class Extensions
{
    public static Task TaskUnwrap(this Task<Task> task, object state = null)
    {
        return task.Unwrap().ContinueWith((t, _) =>
        {
            if (t.Exception != null)
            {
                throw t.Exception;
            }
        }, state ?? task.AsyncState);
    }
}

Метод RunInOrderAsync позволяет выполнять N задач асинхронно, но последовательно - один за другим. По сути, он запускает задачи, созданные из заданных семян, с пределом concurrency равным 1.

Предположим, что задачи, созданные из семян делегатом createTask, не соответствуют нескольким параллельным задачам.

Теперь я хотел бы добавить параметр maxConcurrencyLevel, поэтому подпись функции будет выглядеть так:

Task RunInOrderAsync<TTaskSeed>(int maxConcurrencyLevel,
  IEnumerable<TTaskSeed> taskSeedGenerator,
  CreateTaskDelegate<TTaskSeed> createTask,
  OnTaskErrorDelegate<TTaskSeed> onError = null,
  OnTaskSuccessDelegate<TTaskSeed> onSuccess = null) where TTaskSeed : class

И здесь я немного застрял.

У SO есть такие вопросы:

Что в основном предлагает два способа атаки на проблему:

Используя Parallel.ForEach с ParallelOptions, указав значение свойства MaxDegreeOfParallelism как равное требуемому максимальному уровню concurrency.
Использование пользовательского TaskScheduler с желаемым значением MaximumConcurrencyLevel.

Второй подход не сокращает его, потому что все задействованные задачи должны использовать один и тот же экземпляр планировщика задач. Для этого все методы, используемые для возврата Task, должны иметь перегруз, принимающий пользовательский экземпляр TaskScheduler. К сожалению, Microsoft не очень согласна с этим. Например, SqlConnection.OpenAsync не принимает такой аргумент (но TaskFactory.FromAsync делает).

Первый подход подразумевает, что мне придется преобразовать задачи в действия, примерно так:

() => t.Wait()

Я не уверен, что это хорошая идея, но я буду рад получить больше информации об этом.

Другой подход - использовать TaskFactory.ContinueWhenAny, но это грязно.

Любые идеи?

РЕДАКТИРОВАТЬ 1

Я хотел бы уточнить причины, требующие ограничения. Наши задачи в конечном счете выполняют SQL-запросы на одном сервере SQL. Мы хотим, чтобы ограничить количество параллельных исходящих SQL-запросов. Вполне возможно, что другие SQL-операторы будут выполняться одновременно с другими фрагментами кода, но это пакетный процессор и потенциально может наводнить сервер.

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

Вот почему решения doom day, такие как ThreadPool.SetMaxThreads(), не имеют значения.

Теперь о SqlConnection.OpenAsync. Он был сделан асинхронным по какой-либо причине - он мог бы сделать обратный переход к серверу и, следовательно, может быть подвержен сетевой латентности и другим прекрасным побочным эффектам распределенной среды. Таким образом, это не отличается от других асинхронных методов, которые принимают параметр TaskScheduler. Я склонен думать, что не принимать его - это просто ошибка.

РЕДАКТИРОВАТЬ 2

Я хотел бы сохранить асинхронный дух исходной функции. Поэтому я хочу избежать каких-либо явных блокирующих решений.

РЕДАКТИРОВАТЬ 3

Благодаря @fsimonazzi answer Теперь у меня есть рабочая реализация желаемой функциональности. Вот код:

        var sem = new SemaphoreSlim(maxConcurrencyLevel);
        var tasks = new List<Task>();

        var enumerator = taskSeedGenerator.GetEnumerator();
        while (enumerator.MoveNext())
        {
            tasks.Add(sem.WaitAsync().ContinueWith((_, taskSeed) =>
            {
                Task task = null;
                try
                {
                    task = createTask((TTaskSeed)taskSeed);
                    if (task != null)
                    {
                        Debug.Assert(ReferenceEquals(task.AsyncState, taskSeed));
                        task = task.ContinueWith(t =>
                        {
                            sem.Release();
                            onDone(t);
                        });
                    }
                }
                catch (Exception exc)
                {
                    sem.Release();
                    onFailed(exc, (TTaskSeed)taskSeed);
                }
                return task;
            }, enumerator.Current).TaskUnwrap());
        }

        return Task.Factory.ContinueWhenAll(tasks.ToArray(), _ => sem.Dispose());

Ответ 1

Вы можете использовать семафор для дросселирования обработки. Используя метод WaitAsync(), вы получаете ожидаемую асинхронность. Что-то вроде этого (обработка ошибок удалена для краткости):

private static async Task DoStuff<T>(int maxConcurrency, IEnumerable<T> items, Func<T, Task> createTask)
{
    using (var sem = new SemaphoreSlim(maxConcurrency))
    {
        var tasks = new List<Task>();

        foreach (var item in items)
        {
            await sem.WaitAsync();
            var task = createTask(item).ContinueWith(t => sem.Release());
            tasks.Add(task);
        }

        await Task.WhenAll(tasks);
    }
}

Отредактировано для удаления ошибки, в которой семафор может быть удален до того, как все операции выпуска будут выполнены.

Ответ 2

Два лучших доступных сегодня решения: Semaphoreslim (согласно @fsimonazzi answer) и блок потока TPL (т.е. ActionBlock<T> или TransformBlock<T>). Оба этих блока имеют простой способ установить уровень concurrency.

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

Кроме того, TaskScheduler здесь не будет работать. FYI, TaskScheduler "наследуется" через методы async, как я описываю мой async пост в блоге. Причина, по которой это не будет работать для вашей проблемы, состоит в том, что планировщики задач контролируют выполнение задач, а не событий, поэтому операции SQL, такие как OpenAsync, не "подсчитываются" по отношению к пределу concurrency.

Ответ 3

Вот вариант ответа @fsimonazzi без SemaphoreSlim, такой же классный, как это.

private static async Task DoStuff<T>(int maxConcurrency, IEnumerable<T> items, Func<T, Task> createTask)
{
    var tasks = new List<Task>();
    foreach (var item in items)
    {
        if (tasks.Count >= maxConcurrency)
        {
            await Task.WhenAll(tasks);
            tasks.Clear();
        }
        var task = createTask(item);
        tasks.Add(task);
    }
    await Task.WhenAll(tasks);
}

Ответ 4

Вот вариация ответа @scott-turner, такая же крутая, как и есть. Его ответ отправляет работу в куски maxConcurrency и ждет, пока каждый кусок не завершится полностью, прежде чем отправить следующий фрагмент. Это изменение представляет собой новые задачи по мере необходимости, чтобы гарантировать, что задачи maxConcurrency всегда находятся в полете. Он также демонстрирует работу с Task <T> вместо задачи.

Обратите внимание, что преимущество над версией SemaphoreSlim - с помощью SemaphoreSlim вам нужно ждать двух разных типов Задачи - семафоров и работы. Это проблематично, если работа имеет тип Task <T> вместо задачи.

    private static async Task<R[]> concurrentAsync<T, R>(int maxConcurrency, IEnumerable<T> items, Func<T, Task<R>> createTask)
    {
        var allTasks = new List<Task<R>>();
        var activeTasks = new List<Task<R>>();
        foreach (var item in items)
        {
            if (activeTasks.Count >= maxConcurrency)
            {
                var completedTask = await Task.WhenAny(activeTasks);
                activeTasks.Remove(completedTask);
            }
            var task = createTask(item);
            allTasks.Add(task);
            activeTasks.Add(task);
        }
        return await Task.WhenAll(allTasks);
    }

Ответ 5

Здесь уже много ответов. Я хочу обратиться к комментарию, сделанному вами в ответ Стивенс, о примере использования потока данных TPL для ограничения concurrency. Даже грубо вы оставили комментарий в другом ответе этого вопроса, что вы больше не используете подход, основанный на задачах, это может помочь другим людям.

Пример использования ActionBlock<T> для этого:

private static async Task DoStuff<T>(int maxConcurrency, IEnumerable<T> items, Func<T, Task> createTask)
{
    var ab = new ActionBlock<T>(createTask, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = maxConcurrency });

    foreach (var item in items)
    {
        ab.Post(item);
    }

    ab.Complete();
    await ab.Completion;
}

Более подробную информацию о потоке данных TPL можно найти здесь: https://msdn.microsoft.com/en-us/library/system.threading.tasks.dataflow(v=vs.110).aspx