有限公司并发级别的任务调度器（带任务优先级）处理包裹任务 [英] Limited concurrency level task scheduler (with task priority) handling wrapped tasks
问题描述
我有一个很难找到一个任务调度上，我可以安排优先任务，而且还可以处理包装的任务。这多少有点像什么 Task.Run 试图解决，但你不能指定任务计划 Task.Run 。
我一直在使用 QueuedTaskScheduler 从并行扩展额外样品解决了任务优先级要求（同一个帖子建议）。
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).
下面是我的例子：
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
我要的是：
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 ，这将解决这个问题。但是，任何其他建议是欢迎的。
I think I'm looking for a task scheduler, similar to QueuedTaskScheduler, that will solve this problem. But any other suggestions are welcome.
推荐答案
我能找到的最好的解决办法是让我自己版本的 QueuedTaskScheduler （原来在发现并行扩展额外的样品源$ C $ C）。
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).
我添加了一个布尔awaitWrappedTasks 参数的构造函数的 QueuedTaskScheduler 。
I added a bool awaitWrappedTasks parameter to the constructors of the 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（）方法是异步
private async void ProcessPrioritizedAndBatchedTasks()
然后我修改了code只是其中执行预定任务后的部分：
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;
}
}
}
}
}
方法 ThreadBasedDispatchLoop 的变化是一个有点不同，因为我们不能使用异步关键字否则我们将打破在专用线程（S）执行计划任务的功能。因此，这里是 ThreadBasedDispatchLoop
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
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;
}
}
我已经测试了这一点，它提供了所需的输出。这种技术也可以用于任何其他的调度。例如。 LimitedConcurrencyLevelTaskScheduler 和 OrderedTaskScheduler
这篇关于有限公司并发级别的任务调度器（带任务优先级）处理包裹任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
