6

優先度の高いタスクをスケジュールできるタスク スケジューラを見つけるのに苦労していますが、「ラップされた」タスクも処理できます。Task.Runが解決しようとしているようなものですが、 にタスク スケジューラを指定することはできませんTask.Run。私はQueuedTaskSchedulerParallel Extensions Extras Samplesの を使用して、タスクの優先度要件を解決しています (この投稿でも提案されています)。

これが私の例です：

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この問題を解決するに似たタスク スケジューラを探していると思います。しかし、他の提案は大歓迎です。

4

3 に答える 3

4

残念ながら、これは では解決できませんTaskScheduler。それらは常にTaskレベルで機能し、asyncメソッドにはほとんどの場合複数Taskの が含まれているためです。

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);
  }
}
于 2012-11-14T13:44:01.917 に答える
3

私が見つけた最善の解決策は、自分のバージョン（ Parallel Extensions Extrasサンプルのソースコード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

于 2012-11-16T10:08:48.510 に答える
0

この目標を達成することは不可能だと思います。核となる問題は、 aTaskSchedulerはコードの実行にしか使用できないことです。ただし、IO タスクやタイマー タスクなど、コードを実行しないタスクもあります。TaskSchedulerインフラストラクチャを使用してそれらをスケジュールできるとは思いません。

TaskScheduler の観点からは、次のようになります。

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

ステップ (2) は同期的です。つまり、Task実行される は、ステップ (2) の一部として開始および終了する必要があります。これは、Task非ブロッキングであるため、非同期 IO を実行できないことを意味します。その意味でTaskSchedulerは、ブロッキング コードのみをサポートします。

AsyncSemaphore優先順位に従ってウェイターを解放し、スロットリングを行うバージョンを自分で実装することで、最も役立つと思います。非同期メソッドは、ブロックしない方法でそのセマフォを待機できます。すべての CPU 作業はデフォルトのスレッドプールで実行できるため、 custom 内でカスタム スレッドを開始する必要はありませんTaskScheduler。IO タスクは引き続きノンブロッキング IO を使用できます。

于 2012-11-14T11:44:09.260 に答える