Планировщик задач с ограниченным уровнем параллелизма (с приоритетом задачи), обрабатывающий упакованные задачи

Мне трудно найти планировщик задач, на котором я могу планировать приоритетные задачи, но также могу справиться "завернутые» задачи. Это что-то вроде тогоTask.Run пытается решить, но вы не можете указать планировщик задач дляTask.Run, Я использовалQueuedTaskScheduler отПримеры параллельных расширений для решения задачи приоритета задачи (также предлагается этимсообщение).

Вот мой пример:

class Program
{
    private static QueuedTaskScheduler queueScheduler = new QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);
    private static TaskScheduler ts_priority1;
    private static TaskScheduler ts_priority2;
    static void Main(string[] args)
    {
        ts_priority1 = queueScheduler.ActivateNewQueue(1);
        ts_priority2 = queueScheduler.ActivateNewQueue(2);

        QueueValue(1, ts_priority2);
        QueueValue(2, ts_priority2);
        QueueValue(3, ts_priority2);
        QueueValue(4, ts_priority1);
        QueueValue(5, ts_priority1);
        QueueValue(6, ts_priority1);

        Console.ReadLine();           
    }

    private static Task QueueTask(Func f, TaskScheduler ts)
    {
        return Task.Factory.StartNew(f, CancellationToken.None, TaskCreationOptions.HideScheduler | TaskCreationOptions.DenyChildAttach, ts);
    }

    private static Task QueueValue(int i, TaskScheduler ts)
    {
        return QueueTask(async () =>
        {
            Console.WriteLine("Start {0}", i);
            await Task.Delay(1000);
            Console.WriteLine("End {0}", i);
        }, ts);
    }
}

Типичный вывод приведенного выше примера:

Start 4
Start 5
Start 6
Start 1
Start 2
Start 3
End 4
End 3
End 5
End 2
End 1
End 6

Что я хочу это:

Start 4
End 4
Start 5
End 5
Start 6
End 6
Start 1
End 1
Start 2
End 2
Start 3
End 3

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

Я думаю я'ищу планировщик задач, похожий наQueuedTaskScheduler, что решит эту проблему. Но любые другие предложения приветствуются.

 Francois Nel14 нояб. 2012 г., 13:52
@Keknew QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1); выше делает именно это (ограничить количество одновременных потоков до 1)
 Kek14 нояб. 2012 г., 13:02
Ну, что вы хотите, это обрабатывать приоритет задач, а не запускать их в параллельном режиме? Не могли бы вы просто ограничить количество одновременных потоков в вашем планировщике?

Ответы на вопрос(3)

Решение Вопроса

Лучшее решение, которое я смог найти, - это сделать свою собственную версиюQueuedTaskScheduler (оригинал найден вПримеры параллельных расширений исходный код).

Я добавилbool awaitWrappedTasks параметр для конструкторов.QueuedTaskScheduler

public QueuedTaskScheduler(
        TaskScheduler targetScheduler,
        int maxConcurrencyLevel,
        bool awaitWrappedTasks = false)
{
    ...
    _awaitWrappedTasks = awaitWrappedTasks;
    ...
}

public QueuedTaskScheduler(
        int threadCount,
        string threadName = "",
        bool useForegroundThreads = false,
        ThreadPriority threadPriority = ThreadPriority.Normal,
        ApartmentState threadApartmentState = ApartmentState.MTA,
        int threadMaxStackSize = 0,
        Action threadInit = null,
        Action threadFinally = null,
        bool awaitWrappedTasks = false)
{
    ...
    _awaitWrappedTasks = awaitWrappedTasks;

    // code starting threads (removed here in example)
    ...
}

Затем я изменилProcessPrioritizedAndBatchedTasks() метод бытьasync

private async void ProcessPrioritizedAndBatchedTasks()

Затем я изменил код сразу после части, где выполняется запланированное задание:

private async void ProcessPrioritizedAndBatchedTasks()
{
    bool continueProcessing = true;
    while (!_disposeCancellation.IsCancellationRequested && continueProcessing)
    {
        try
        {
            // Note that we're processing tasks on this thread
            _taskProcessingThread.Value = true;

            // Until there are no more tasks to process
            while (!_disposeCancellation.IsCancellationRequested)
            {
                // Try to get the next task.  If there aren't any more, we're done.
                Task targetTask;
                lock (_nonthreadsafeTaskQueue)
                {
                    if (_nonthreadsafeTaskQueue.Count == 0) break;
                    targetTask = _nonthreadsafeTaskQueue.Dequeue();
                }

                // If the task is null, it's a placeholder for a task in the round-robin queues.
                // Find the next one that should be processed.
                QueuedTaskSchedulerQueue queueForTargetTask = null;
                if (targetTask == null)
                {
                    lock (_queueGroups) FindNextTask_NeedsLock(out targetTask, out queueForTargetTask);
                }

                // Now if we finally have a task, run it.  If the task
                // was associated with one of the round-robin schedulers, we need to use it
                // as a thunk to execute its task.
                if (targetTask != null)
                {
                    if (queueForTargetTask != null) queueForTargetTask.ExecuteTask(targetTask);
                    else TryExecuteTask(targetTask);

                    // ***** MODIFIED CODE START ****
                    if (_awaitWrappedTasks)
                    {
                        var targetTaskType = targetTask.GetType();
                        if (targetTaskType.IsConstructedGenericType && typeof(Task).IsAssignableFrom(targetTaskType.GetGenericArguments()[0]))
                        {
                            dynamic targetTaskDynamic = targetTask;
                            // Here we await the completion of the proxy task.
                            // We do not await the proxy task directly, because that would result in that await will throw the exception of the wrapped task (if one existed)
                            // In the continuation we then simply return the value of the exception object so that the exception (stored in the proxy task) does not go totally unobserved (that could cause the process to crash)
                            await TaskExtensions.Unwrap(targetTaskDynamic).ContinueWith((Func)(t => t.Exception), TaskContinuationOptions.ExecuteSynchronously);
                        }
                    }
                    // ***** MODIFIED CODE END ****
                }
            }
        }
        finally
        {
            // Now that we think we're done, verify that there really is
            // no more work to do.  If there's not, highlight
            // that we're now less parallel than we were a moment ago.
            lock (_nonthreadsafeTaskQueue)
            {
                if (_nonthreadsafeTaskQueue.Count == 0)
                {
                    _delegatesQueuedOrRunning--;
                    continueProcessing = false;
                    _taskProcessingThread.Value = false;
                }
            }
        }
    }
}

Смена методаThreadBasedDispatchLoop было немного по-другому, в том, что мы не можем использоватьasync Ключевое слово или мы нарушим функциональность выполнения запланированных задач в выделенном потоке (ах). Так вот модифицированная версияThreadBasedDispatchLoop

private void ThreadBasedDispatchLoop(Action threadInit, Action threadFinally)
{
    _taskProcessingThread.Value = true;
    if (threadInit != null) threadInit();
    try
    {
        // If the scheduler is disposed, the cancellation token will be set and
        // we'll receive an OperationCanceledException.  That OCE should not crash the process.
        try
        {
            // If a thread abort occurs, we'll try to reset it and continue running.
            while (true)
            {
                try
                {
                    // For each task queued to the scheduler, try to execute it.
                    foreach (var task in _blockingTaskQueue.GetConsumingEnumerable(_disposeCancellation.Token))
                    {
                        Task targetTask = task;
                        // If the task is not null, that means it was queued to this scheduler directly.
                        // Run it.
                        if (targetTask != null)
                        {
                            TryExecuteTask(targetTask);
                        }
                        // If the task is null, that means it's just a placeholder for a task
                        // queued to one of the subschedulers.  Find the next task based on
                        // priority and fairness and run it.
                        else
                        {
                            // Find the next task based on our ordering rules...                                    
                            QueuedTaskSchedulerQueue queueForTargetTask;
                            lock (_queueGroups) FindNextTask_NeedsLock(out targetTask, out queueForTargetTask);

                            // ... and if we found one, run it
                            if (targetTask != null) queueForTargetTask.ExecuteTask(targetTask);
                        }

                        if (_awaitWrappedTasks)
                        {
                            var targetTaskType = targetTask.GetType();
                            if (targetTaskType.IsConstructedGenericType && typeof(Task).IsAssignableFrom(targetTaskType.GetGenericArguments()[0]))
                            {
                                dynamic targetTaskDynamic = targetTask;
                                // Here we wait for the completion of the proxy task.
                                // We do not wait for the proxy task directly, because that would result in that Wait() will throw the exception of the wrapped task (if one existed)
                                // In the continuation we then simply return the value of the exception object so that the exception (stored in the proxy task) does not go totally unobserved (that could cause the process to crash)
                                TaskExtensions.Unwrap(targetTaskDynamic).ContinueWith((Func)(t => t.Exception), TaskContinuationOptions.ExecuteSynchronously).Wait();
                            }
                        }
                    }
                }
                catch (ThreadAbortException)
                {
                    // If we received a thread abort, and that thread abort was due to shutting down
                    // or unloading, let it pass through.  Otherwise, reset the abort so we can
                    // continue processing work items.
                    if (!Environment.HasShutdownStarted && !AppDomain.CurrentDomain.IsFinalizingForUnload())
                    {
                        Thread.ResetAbort();
                    }
                }
            }
        }
        catch (OperationCanceledException) { }
    }
    finally
    {
        // Run a cleanup routine if there was one
        if (threadFinally != null) threadFinally();
        _taskProcessingThread.Value = false;
    }
}

Я проверил это, и это дает желаемый результат. Этот метод также может быть использован для любого другого планировщика. Например.LimitedConcurrencyLevelTaskScheduler а такжеOrderedTaskScheduler

 Francois Nel16 февр. 2017 г., 12:10
@ springy76 на самом деле это работало очень хорошо, но с тех пор я отошел от любых явных реализаций TaskScheduler, когда обнаружил правильное использованиеActionBlock в платформе TPL DataFlow
 usr16 нояб. 2012 г., 12:56
Ожидание задачи в планировщике уничтожает значение асинхронного ввода-вывода. Если вы неВ любом случае, асинхронный ввод-вывод не требуется, вы можете переключаться на синхронные тела задач.
 usr16 нояб. 2012 г., 15:56
+1 тогда. Я многому научился в этом вопросе. Не совсем уверен, что это решение предпочтительнееAsyncSemaphore но я'подумаю об этом
 springy7615 февр. 2017 г., 10:22
Вы выполняетеasync-void метод изнутриTaskScheduler реализация? Страшно, интересно, что @StephenCleary ничего не говорит по этому поводу.

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

С точки зрения TaskScheduler это выглядит так:

1. Select a registered task for execution
2. Execute its code on the CPU
3. Repeat

Шаг (2) является синхронным, что означает, чтоTask для выполнения должен начинаться и заканчиваться как часть шага (2). Это означает, что этоTask не может сделать асинхронный ввод-вывод, потому что это было бы неблокирующим. В этом смысле,TaskScheduler поддерживает только код блокировки.

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

 usr14 нояб. 2012 г., 14:39
Я верю тебеверно. Я добавил некоторые мысли и предложения. Пожалуйста, дайте мне знать, что вы думаете.
 usr14 нояб. 2012 г., 16:46
Вы можете создать собственный планировщик, который будет знать, что выполнение задачи завершено только после завершения перенесенной задачи, а также задачи переноса. Это должно было бы сделать приведение во время выполнения, чтобы видеть, поставлена ли в очередьTask на самом делеTask и в этом случае добавить продолжение и так далее.
 Francois Nel16 нояб. 2012 г., 11:18
Я добавил свойответ, Пожалуйста, посмотрите и посмотрите, что вы думаете.
 Francois Nel14 нояб. 2012 г., 16:08
Спасибо за ваше обновление. Ваше предложение использовать блокировку семафора - это именно то, что пользователь предложил в следующемответ (см. мои комментарии). Ваше предположение о том, что планировщик выполняет только свои задачи синхронно, несколько верно, но что, если планировщик ожидает "завернутые» задача каждой задачи, прежде чем она выполнит любые другие задачи в очереди. Я думаю, что это дало мне идею ... спасибо (сообщу, если я что-нибудь придумаю).
 Francois Nel14 нояб. 2012 г., 14:11
то, что вы объяснили здесь, я уже попробовал, и у него в основном тот же результат (как в исходной задаче). В вашем предложенииfirstPartTask запланирован в планировщике задач в очереди, но завершается, как только он достигает первогоawait а планировщик просто выполняет следующий "первая часть" в очереди, даже если предыдущийвнутренняя задача " (сброс задания после первогоawait) не завершена. Я могу только думать, что это будет решенопланировщик который обрабатывает этот сценарий, яищет и не может быть решена какой-то магией за пределами планировщика.

К сожалению, это можетне может быть решена сTaskSchedulerпотому что они всегда работают наTask уровень иasync Метод почти всегда содержит несколькоTasks.

Вы должны использоватьSemaphoreSlim в сочетании с планировщиком приоритетов. В качестве альтернативы, вы можете использоватьAsyncLock (который также включен в мойБиблиотека AsyncEx).

class Program
{
  private static QueuedTaskScheduler queueScheduler = new QueuedTaskScheduler(targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);
  private static TaskScheduler ts_priority1;
  private static TaskScheduler ts_priority2;
  private static SemaphoreSlim semaphore = new SemaphoreSlim(1);
  static void Main(string[] args)
  {
    ts_priority1 = queueScheduler.ActivateNewQueue(1);
    ts_priority2 = queueScheduler.ActivateNewQueue(2);

    QueueValue(1, ts_priority2);
    QueueValue(2, ts_priority2);
    QueueValue(3, ts_priority2);
    QueueValue(4, ts_priority1);
    QueueValue(5, ts_priority1);
    QueueValue(6, ts_priority1);

    Console.ReadLine();           
  }

  private static Task QueueTask(Func f, TaskScheduler ts)
  {
    return Task.Factory.StartNew(f, CancellationToken.None, TaskCreationOptions.HideScheduler | TaskCreationOptions.DenyChildAttach, ts).Unwrap();
  }

  private static Task QueueValue(int i, TaskScheduler ts)
  {
    return QueueTask(async () =>
    {
      await semaphore.WaitAsync();
      try
      {
        Console.WriteLine("Start {0}", i);
        await Task.Delay(1000);
        Console.WriteLine("End {0}", i);
      }
      finally
      {
        semaphore.Release();
      }
    }, ts);
  }
}
 Stephen Cleary14 нояб. 2012 г., 16:13
В этом случае выВам понадобится фактическая блокировка на основе приоритетов, которая не существует, потому что AFAIK никому больше не нужна. Вы'Я буду строить свой собственный.
 Francois Nel16 нояб. 2012 г., 11:18
Я добавил свойответ, Пожалуйста, посмотрите и посмотрите, что вы думаете.
 Francois Nel14 нояб. 2012 г., 15:53
Это выглядит как интересное решение. Однако я вижу одну проблему с этим. Хотя решение (сначала) приведет к правильному выводу (как в этом вопросе), но оно нарушит приоритет выполненных задач. Планировщик будет выполнять все задачи (с правильным приоритетом) доawait semaphore.WaitAsync() но задачи с более высоким приоритетом не будут освобождены из блокировки до задач с более низким приоритетом. Это особенно верно, если задачи с более высоким приоритетом запланированы после задач с более низким приоритетом (которые все еще ожидают освобождения из блокировки).

Ваш ответ на вопрос