Глава 13. Планирование

Каждая часть должна выполняться в каком-то потоке. Планировщик (scheduler) — объект, который решает, где должен выполняться тот или иной код. В фреймворке .NET существует несколько разных типов планировщиков, которые по-разному используются параллельным кодом и кодом потоков данных.

Рекомендую при возможности не задавать планировщика; обычно настройки по умолчанию работают правильно. Например, оператор await в асинхронном коде автоматически возобновит выполнение метода в том же контексте, если только вы не переопределите значение по умолчанию, как описано в рецепте 2.7. У реактивного кода тоже имеются разумные контексты по умолчанию для выдачи событий, хотя их можно переопределить с помощью ObserveOn, как описано в рецепте 6.2.

Если другой код должен выполняться в конкретном контексте (например, в контексте UI-потока или в контексте запроса ASP.NET), то рецепты этой главы помогут в планировании кода.

13.1. Планирование работы в пуле потоков

Задача

Имеется фрагмент кода, который должен выполняться в потоке из пула потоков.

Решение

В большинстве случаев следует использовать Task.Run; это достаточно просто. Следующий пример блокирует поток из пула потоков на 2 секунды:

Task task = Task.Run(() =>

{

  Thread.Sleep(TimeSpan.FromSeconds(2));

});

Task.Run также поддерживает возвращаемые значения и асинхронные лямбда-выражения. Задача, возвращаемая Task.Run в следующем коде, завершится через 2 секунды с результатом 13:

Task<int> task = Task.Run(async () =>

{

  await Task.Delay(TimeSpan.FromSeconds(2));

  return 13;

});

Task.Run возвращает объект Task (или Task<T>), который может естественным образом потребляться асинхронным или реактивным кодом.

Пояснение

Task.Run идеально подходит для UI-приложений с продолжительной работой, которая не должна выполняться в UI-потоке. Например, в рецепте 8.4 Task.Run используется для вынесения параллельной обработки в поток из пула потоков. Тем не менее не используйте Task.Run в ASP.NET, если только вы не уверены в том, что делаете. В ASP.NET код обработки запросов уже выполняется в потоке из пула потоков, так что перенесение его в другой поток из пула потоков обычно нерационально.

Task.Run является фактической заменой для BackgroundWorker, Delegate.BeginInvoke и ThreadPool.QueueUserWorkItem. Ни один из этих старых API не следует использовать в новом коде; код с Task.Run намного проще пишется и сопровождается со временем. Более того, Task.Run справляется с большинством задач, для которых используется Thread, так что в большинстве случаев Thread может заменяться Task.Run (за редким исключением потоков из модели однопоточного подразделения).

Параллельный код и код потоков данных выполняется в пуле потоков по умолчанию, поэтому обычно Task.Run не нужно использовать с кодом, выполняемым Parallel, библиотекой TPL Dataflow или Parallel LINQ.

Если вы применяете динамический параллелизм, используйте Task.Factory.StartNew  вместо Task.Run. Это необходимо из-за того, что у объекта Task, возвращаемого Task.Run, параметры по умолчанию настроены для асинхронного использования (т. е. для потребления в асинхронном или реактивном коде). Кроме того, он не поддерживает такие расширенные возможности, как задачи «родитель/потомок», типичные для динамического параллельного кода.

Дополнительная информация

В рецепте 8.6 рассматривается потребление асинхронного кода (например, задачи, возвращенной Task.Run) из реактивного кода.

В рецепте 8.4 рассматривается асинхронное ожидание параллельного кода, проще всего реализуемое с использованием Task.Run.

В рецепте 4.4 рассматривается динамический параллелизм — сценарий, в котором следует использовать Task.Factory.StartNew вместо Task.Run.

13.2. Выполнение кода с помощью планировщика задач

Задача

Есть несколько частей кода, которые требуется выполнить определенным способом. Например, все части кода должны выполняться в UI-потоке или же в любой момент времени должно выполняться только определенное количество частей.

В этом рецепте показано, как определить и сконструировать планировщик для этих частей кода. Применению планировщика будут посвящены следующие два рецепта.

Решение

В .NET есть немало типов, предназначенных для работы с планированием; в этом рецепте мы сосредоточимся на типе TaskScheduler, потому что он портируем и относительно прост в использовании.

Простейшая разновидность TaskScheduler — TaskScheduler.Default — ставит работу в очередь пула потоков. Вам редко придется использовать TaskScheduler.Default в своем коде, но важно о нем помнить, потому что этот планировщик используется по умолчанию во многих сценариях планирования. TaskScheduler.Default используется Task.Run в параллельном коде и в коде потоков данных.

Вы можете сохранить конкретный контекст и позднее спланировать работу в этом контексте с помощью TaskScheduler.FromCurrent­Synchro­nizationContext:

TaskScheduler scheduler =

   TaskScheduler.FromCurrentSynchronizationContext();

Этот код создает объект TaskScheduler, чтобы сохранить текущий объект SynchronizationContext и спланировать выполнение кода в этом контексте. Тип SynchronizationContext представляет контекст планирования общего назначения. В фреймворке .NET предусмотрено несколько разных контекстов; многие UI-фреймворки предоставляют контекст SynchronizationContext, представляющий UI-поток, а в ASP.NET до Core предоставлялся контекст SynchronizationContext, представляющий контекст запроса HTTP.

ConcurrentExclusiveSchedulerPair — еще один высокоэффективный тип, появившийся в .NET 4.5; в действительности это два планировщика, связанных друг с другом. Компонент ConcurrentScheduler содержит планировщик, позволяющий нескольким задачам выполняться одновременно — при условии, что ни одна задача не выполняется в ExclusiveScheduler. ExclusiveScheduler выполняет только по одной задаче за раз и только в том случае, если в настоящее время никакие задачи не выполняются в ConcurrentScheduler:

var schedulerPair = new ConcurrentExclusiveSchedulerPair();

TaskScheduler concurrent = schedulerPair.ConcurrentScheduler;

TaskScheduler exclusive = schedulerPair.ExclusiveScheduler;

Одно из частых применений ConcurrentExclusiveSchedulerPair — простое использование ExclusiveScheduler, гарантирующее, что в любой момент времени будет выполняться только одна задача. Код, выполняемый в ExclusiveScheduler, будет выполняться в пуле потоков, но будет ограничен монопольным выполнением без всего остального кода с использованием экземпляра ExclusiveScheduler.

Также ConcurrentExclusiveSchedulerPair может выполняться в качестве регулирующего планировщика. Вы можете создать объект Concurrent­Exclusive­SchedulerPair, который будет ограничивать собственный уровень параллелизма. В этом сценарии ExclusiveScheduler обычно не используется:

var schedulerPair = new ConcurrentExclusiveSchedulerPair(

    TaskScheduler.Default, maxConcurrencyLevel: 8);

TaskScheduler scheduler = schedulerPair.ConcurrentScheduler;

Учтите, что такая регулировка влияет на код только во время его выполнения; она сильно отличается от логической регулировки, рассмотренной в рецепте 12.5. В частности, асинхронный код не считается выполняемым во время ожидания операции. ConcurrentScheduler регулирует выполняющийся код; другие виды регулировки (такие, как SemaphoreSlim) осущест­в­ляют регулировку на более высоком уровне (т. е. всего async-метода.)

Пояснение

Возможно, вы заметили, что в последнем примере конструктору Concurrent­ExclusiveSchedulerPair передается объект TaskScheduler.Default. Это объясняется тем, что ConcurrentExclusiveSchedulerPair применяет свою конкурентную/монопольную логику к существующему TaskScheduler.

В этом рецепте представлен метод TaskScheduler.FromCurrent­Synchro­nizationContext, используемый для выполнения кода в сохраненном контексте. Также возможно напрямую использовать SynchronizationContext для выполнения кода в этом контексте; тем не менее я не рекомендую применять этот подход. Там, где это возможно, используйте оператор await для возобновления в неявно сохраненном контексте или обертку TaskScheduler.

Никогда не используйте платформенно-зависимые типы для выполнения кода в UI-потоке. WPF, Silverlight, iOS и Android предоставляют тип Dispatcher, Universal Windows использует тип CoreDispatcher, а в Windows Forms существует интерфейс ISynchronizeInvoke (т. е. Control.Invoke). Не используйте эти типы в новом коде; просто считайте, что их вообще нет. Эти типы только без всякой необходимости привязывают код к конкретной платформе. SynchronizationContext — абстракция общего назначения на базе этих типов.

В System.Reactive (Rx) появилась еще более универсальная абстракция планировщика: IScheduler. Планировщик Rx способен инкапсулировать любую разновидность планировщика; TaskPoolScheduler инкапсулирует любой объект TaskFactory (который содержит TaskScheduler). Команда Rx также определила реализацию IScheduler, которой можно управлять вручную в целях тестирования. Если вам потребовалось использовать абстракцию планировщика, я рекомендую IScheduler из Rx; она хорошо спроектирована, четко определена и удобна для тестирования. В большинстве случаев абстракция планировщика не нужна, а более ранние библиотеки — такие, как Task Parallel Library (TPL) и TPL Dataflow, — «понимают» только тип TaskScheduler.

Дополнительная информация

В рецепте 13.3 рассматривается применение TaskScheduler в параллельном коде.

В рецепте 13.4 рассматривается применение TaskScheduler в коде потоков данных.

В рецепте 12.5 рассматривается высокоуровневая логическая регулировка.

В рецепте 6.2 рассматриваются планировщики System.Reactive для потоков событий.

В рецепте 7.6 рассматривается тестовый планировщик System.Reactive.

13.3. Планирование параллельного кода

Задача

Требуется управлять выполнением отдельных фрагментов в параллельном коде.

Решение

После того как вы создадите нужный экземпляр TaskScheduler (см. рецепт 13.2), можете включить его в набор параметров, передаваемых методу Parallel. Следующий код получает последовательность последовательностей матриц. Он запускает несколько параллельных циклов и ограничивает общий параллелизм всех циклов одновременно независимо от количества матриц в каждой последовательности:

void RotateMatrices(IEnumerable<IEnumerable<Matrix>> collections,

   float degrees)

{

  var schedulerPair = new ConcurrentExclusiveSchedulerPair(

      TaskScheduler.Default, maxConcurrencyLevel: 8);

  TaskScheduler scheduler = schedulerPair.ConcurrentScheduler;

  ParallelOptions options = new ParallelOptions { TaskScheduler =

     scheduler };

  Parallel.ForEach(collections, options,

      matrices => Parallel.ForEach(matrices, options,

          matrix => matrix.Rotate(degrees)));

}

Пояснение

Parallel.Invoke также получает экземпляр ParallelOptions, поэтому вы можете передать TaskScheduler при вызове Parallel.Invoke так же, как и для Parallel.ForEach. При выполнении динамического параллельного кода можно передать TaskScheduler непосредственно TaskFactory.StartNew или Task.ContinueWith.

Передать TaskScheduler коду Parallel LINQ (PLINQ) невозможно.

Дополнительная информация

В рецепте 13.2 рассматриваются основные планировщики задач и рекомендации по выбору между ними.

13.4. Синхронизация потоков данных с помощью планировщиков

Задача

Требуется управлять выполнением отдельных фрагментов в коде потоков данных.

Решение

После того как вы создадите нужный экземпляр TaskScheduler (см. рецепт 13.2), вы можете включить его в набор параметров, передаваемых блоку потока данных. При вызове из UI-потока следующий код создает сеть потока данных, которая умножает все свои входные значения на 2 (с использованием пула потоков), а затем присоединяет полученные значения к списку (в UI-потоке):

var options = new ExecutionDataflowBlockOptions

{

  TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext(),

};

var multiplyBlock = new TransformBlock<int, int>(item => item * 2);

var displayBlock = new ActionBlock<int>(

    result => ListBox.Items.Add(result), options);

multiplyBlock.LinkTo(displayBlock);

Пояснение

Назначение TaskScheduler особенно полезно при координации действий блоков в разных частях вашей сети потока данных. Например, можно воспользоваться ConcurrentExclusiveSchedulerPair.ExclusiveScheduler, чтобы блоки A и C никогда не выполнялись одновременно, а блок B мог выполняться тогда, когда пожелает.

Учтите, что синхронизация TaskScheduler действует только во время выполнения кода. Например, если имеется блок действия, который выполняет асинхронный код и применяет монопольный планировщик, код не считается выполняющимся во время ожидания.

TaskScheduler можно задать для любой разновидности блоков потока данных. Даже при том, что блок может не выполнять ваш код (например, BufferBlock<T>), у него все равно имеются служебные задачи, которые необходимо выполнять, и блок будет использовать предоставленный объект TaskScheduler для всей своей внутренней работы.

Дополнительная информация

В рецепте 13.2 рассматриваются основные планировщики задач и рекомендации по выбору между ними.