有限的并发级别任务调度器(具有任务优先级)处理包装的任务

I'm having a hard time finding a task scheduler on which I can schedule prioritised tasks but can also handle "wrapped" tasks. It is something like what Task.Run tries to solve, but you cannot specify a task scheduler to Task.Run. I have been using a QueuedTaskScheduler from the Parallel Extensions Extras Samples to solve the task priority requirement (also suggested by this post).

我很难找到一个任务调度程序，我可以调度优先任务，但也可以处理“包装”任务。就像什么任务。运行尝试解决问题，但是不能指定要执行的任务调度程序。我一直在使用并行扩展附加示例中的QueuedTaskScheduler来解决任务优先级需求(本文也建议这样做)。

Here is my example:

这是我的例子:

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<Task> 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);
    }
}

The typical output of the example above is:

上面示例的典型输出是:

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

What I want is:

我想要的是:

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

EDIT:

编辑:

I think I'm looking for a task scheduler, similar to QueuedTaskScheduler, that will solve this problem. But any other suggestions are welcome.

我想我正在寻找一个任务调度程序，类似于QueuedTaskScheduler，可以解决这个问题。但任何其他建议都是受欢迎的。

3 个解决方案

#1

Unfortunately, this can't be solved with a TaskScheduler, because they always work at the Task level, and an async method almost always contains multiple Tasks.

不幸的是，这不能通过TaskScheduler解决，因为它们总是在任务级别上工作，异步方法几乎总是包含多个任务。

You should use a SemaphoreSlim in conjunction with a prioritizing scheduler. Alternatively, you could use AsyncLock (which is also included in my AsyncEx library).

您应该与优先级调度程序一起使用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<Task> 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);
  }
}

#2

The best solution I could find is to make my own version of the QueuedTaskScheduler (original found in the Parallel Extensions Extras Samples source code).

我能找到的最好的解决方案是制作我自己的QueuedTaskScheduler版本(最初在Parallel Extensions Extras中找到)。

I added a bool awaitWrappedTasks parameter to the constructors of the QueuedTaskScheduler.

我在QueuedTaskScheduler的构造函数中添加了bool awaitWrappedTasks参数。

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)
    ...
}

I then modified the ProcessPrioritizedAndBatchedTasks() method to be async

然后，我将processprized和batchedtasks()方法修改为异步的

private async void ProcessPrioritizedAndBatchedTasks()

I then modified the code just after the part where the scheduled task is executed:

然后我在执行计划任务的部分之后修改了代码:

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<Task, Exception>)(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;
                }
            }
        }
    }
}

The change of method ThreadBasedDispatchLoop was a bit different, in that we cannot use the async keyword or else we will break the functionality of executing scheduled tasks in the dedicated thread(s). So here is the modified version of ThreadBasedDispatchLoop

方法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<Task, Exception>)(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;
    }
}

I have tested this and it gives the desired output. This technique could also be used for any other scheduler. E.g. LimitedConcurrencyLevelTaskScheduler and OrderedTaskScheduler

我已经测试过这个，它给出了想要的输出。这种技术也可以用于任何其他调度器。例如,LimitedConcurrencyLevelTaskScheduler OrderedTaskScheduler

#3

I think it is impossible to achieve this goal. A core problem seems to be that a TaskScheduler can only be used to run code. But there are tasks that do not run code, such as IO tasks or timer tasks. I don't think the TaskScheduler infrastructure can be used to schedule those.

我认为实现这个目标是不可能的。一个核心问题似乎是TaskScheduler只能用于运行代码。但是有些任务不运行代码，比如IO任务或计时器任务。我不认为TaskScheduler基础结构可以用来安排这些计划。

From the perspective of a TaskScheduler it looks like this:

从TaskScheduler的角度来看，它是这样的:

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

Step (2) is synchronous which means that the Task to be executed must start and finish as part of step (2). This means that this Task cannot do async IO because that would be non-blocking. In that sense, TaskScheduler only supports blocking code.

步骤(2)是同步的，这意味着要执行的任务必须作为步骤(2)的一部分启动和完成，这意味着该任务不能进行异步IO，因为这将是非阻塞的。在这个意义上，TaskScheduler只支持阻塞代码。

I think you would be served best by implementing yourself a version of AsyncSemaphore that releases waiters in priority order and does throttling. Your async methods can await that semaphore in a non-blocking way. All CPU work can run on the default thread-pool so there is no need to start custom threads inside of a custom TaskScheduler. IO tasks can continue to use non-blocking IO.

我认为您最好是实现自己的一个版本的AsyncSemaphore，以优先顺序释放服务人员，并进行节流。异步方法可以以非阻塞方式等待信号量。所有CPU工作都可以在默认的线程池上运行，因此不需要在自定义任务调度程序中启动自定义线程。IO任务可以继续使用非阻塞IO。