Планировщик задач с ограниченным уровнем параллелизма (с приоритетом задачи), обрабатывающий упакованные задачи
Мне трудно найти планировщик задач, на котором я могу планировать приоритетные задачи, но я также могу обрабатывать "обернутые" задачи. Это что-то вроде того, что пытается решить Task.Run, но вы не можете указать планировщик задач для Task.Run, Я использовал QueuedTaskScheduler из примеров дополнительных возможностей Parallel Extensions для решения задачи приоритета задачи (также предложенной в этом посте).
Вот мой пример:
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);
}
}
Типичный вывод приведенного выше примера:
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, что решит эту проблему. Но любые другие предложения приветствуются.
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<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 было немного по-другому, в том, что мы не можем использовать 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;
}
}
Я проверил это, и это дает желаемый результат. Этот метод также может быть использован для любого другого планировщика. Например LimitedConcurrencyLevelTaskScheduler а также OrderedTaskScheduler
К сожалению, это не может быть решено с TaskScheduler потому что они всегда работают на Task уровень и async метод почти всегда содержит несколько Task s.
Вы должны использовать 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);
}
}
Я думаю, что невозможно достичь этой цели. Основная проблема заключается в том, что 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, Задачи ввода-вывода могут продолжать использовать неблокирующие операции ввода-вывода.