La soluzione migliore che ho trovato è quello di rendere la mia versione del QueuedTaskScheduler
(originale trovata nel codice sorgente di Parallel Extensions Extras Samples).
Ho aggiunto un parametro bool awaitWrappedTasks
ai costruttori dello 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)
...
}
Ho quindi modificato il metodo ProcessPrioritizedAndBatchedTasks()
essere async
private async void ProcessPrioritizedAndBatchedTasks()
Ho poi modificato il codice subito dopo la parte in cui viene eseguita l'operazione pianificata:
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;
}
}
}
}
}
Il cambiamento di metodo ThreadBasedDispatchLoop
era un po 'diverso, nel senso che non possiamo usare la parola chiave async
altrimenti interromperà la funzionalità di ex ecuting delle attività pianificate nei thread dedicati. Così qui è la versione modificata di 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;
}
}
Ho testato questo e dà l'output desiderato. Questa tecnica potrebbe anche essere utilizzata per qualsiasi altro programmatore. Per esempio. LimitedConcurrencyLevelTaskScheduler
e OrderedTaskScheduler
Bene, ciò che si desidera è gestire la priorità delle attività, ma non eseguirle in modalità parallela? potresti non limitare solo il numero di thread simultanei nel tuo scheduler? – Kek
@Kek 'new QueuedTaskScheduler (targetScheduler: TaskScheduler.Default, maxConcurrencyLevel: 1);' sopra fa esattamente questo (limita il numero di thread simultanei a 1) –