Достоинство неизменяемых объектов с точки зрения многопоточности заключается в том, что работа с ними требует коротких блокировок, обычно обрамляющих операции присваивания объектов:
public class WorkWithImmutable
{
private readonly object _locker = new object(); private ProgressStatus _status;
public void SetFields()
{
// создаём и настраиваем временный объект
var status = new ProgressStatus(50, "Working on it");
// переносим информацию, используя короткую блокировку lock (_locker) _status = status;
}
public void ReadInfo()
{
//используя короткую блокировку, создаём временную копию
ProgressStatus statusCopy;
lock (_locker) statusCopy = _status;
//работаем с копией
int pc = statusCopy.PercentComplete; string msg = statusCopy.StatusMessage;
}
}
33. Библиотека параллельных задач
Библиотека параллельных задач (Task Parallel Library, TPL) является ча-
стью платформы .NET 4 и представляет собой набор типов в пространствах
имён System.Threading.Tasks и System.Threading. Эта библиотека предназна-
чена для повышения производительности разработчиков за счёт упрощения добавления параллелизма в приложения. Она динамически масштабирует степень параллелизма для наиболее эффективного использования всех доступных процессорных ядер.
TPL обеспечивает три уровня организации параллелизма:
1.Параллелизм на уровне задач. Библиотека обеспечивает высокоуровневую работу с пулом потоков, позволяя явно структурировать параллельно исполняющийся код с помощью легковесных задач. Планировщик библиотеки выполняет диспетчеризацию задач, а также предоставляет единообразный механизм отмены задач и обработки исключительных ситуаций.
2.Параллелизм при императивной обработке данных. Библиотека содер-
жит параллельные реализации основных итеративных операторов, таких как
136
циклы for и foreach. При этом выполнение автоматически распределяется на все процессорные ядра вычислительной системы.
3. Параллелизм при декларативной обработке данных реализуется при по-
мощи параллельного интегрированного языка запросов (PLINQ). PLINQ выполняет запросы LINQ параллельно, обеспечивая масштабируемость и загрузку процессорных ядер.
33.1. Параллелизм на уровне задач
Параллелизм на уровне задач – базовый элемент TPL. Задача (task) – сущность, которая в целом подобна потоку. Основное различие заключается в том, что исполнением задач управляет специальный планировщик, опирающийся в своей работе на пул потоков.
Для представления задач используются классы Task и Task<T>, размещённые в пространстве имён System.Threading.Tasks. Табл. 20 содержит описание элементов класса Task.
|
Таблица 20 |
|
|
Элементы класса Task |
|
|
|
|
Имя элемента |
Описание |
|
AsyncState |
Объект, заданный при создании задачи как аргумент Action<object> |
|
ContinueWith(), |
Используются для указания метода, выполняемого после завершения |
|
ContinueWith<T>() |
текущей задачи |
|
CreationOptions |
Опции, указанные при создании задачи (тип TaskCreationOptions) |
|
CurrentId |
Статическое свойство типа int?, которое возвращает целочисленный |
|
идентификатор текущей задачи |
||
|
||
Dispose() |
Освобождение ресурсов, связанных с задачей |
|
Exception |
Возвращает объект типа AggregateException, который соответствует |
|
исключению, прервавшему выполнение задачи |
||
|
||
Factory |
Доступ к фабрике, содержащей методы создания Task и Task<T> |
|
Id |
Целочисленный идентификатор задачи |
|
IsCanceled |
Булево свойство, указывающее, была ли задача отменена |
|
IsCompleted |
Свойство равно true, если выполнение задачи успешно завершилось |
|
IsFaulted |
Свойство равно true, если задача сгенерировала исключение |
|
RunSynchronously() |
Запуск задачи синхронно |
|
Start() |
Запуск задачи асинхронно |
|
Status |
Возвращает текущий статус задачи (объект типа TaskStatus) |
|
Wait() |
Приостанавливает текущий поток до завершения задачи |
|
WaitAll() |
Статический метод; приостанавливает текущий поток до завершения |
|
всех указанных задач |
||
|
||
WaitAny() |
Статический метод; приостанавливает текущий поток до завершения |
|
любой из указанных задач |
||
|
Для создания задачи используется один из перегруженных конструкторов класса Task. При этом указывается аргумент типа Action – метод, выполняемый в задаче. Если необходим метод с параметром, используется Action<object> и дополнительный аргумент типа object.
137
Action m1 = () => {
Thread.Sleep(2000); Console.WriteLine("Done");
};
Action<object> m2 = obj => {
Thread.Sleep(1000); Console.WriteLine(obj.ToString());
};
var t1 = new Task(m1); var t2 = new Task(m2, 25);
Перегруженные конструкторы класса Task принимают опциональные ар-
гументы типа CancellationToken и TaskCreationOptions. Перечисление
TaskCreationOptions задаёт вид задачи (например, LongRunning – долгая задача). Структура CancellationToken применяется для прерывания задачи.
var t1 = new Task(m1, TaskCreationOptions.LongRunning);
Созданная задача ставится в очередь планировщика и запускается при помощи методов Start() или RunSynchronously(). Второй метод запускает задачу в текущем потоке. Оба метода могут принимать аргумент типа TaskScheduler (пользовательский планировщик задач).
// используем задачи t1 и t2, |
объявленные |
выше |
|
t1.Start(); |
// |
асинхронный |
запуск |
t2.RunSynchronously(); |
// |
синхронный запуск |
|
Console.WriteLine("Tasks were |
started"); |
// напечатано через 1 сек. |
|
Метод ContinueWith() позволяет создать цепочку задач. Этот метод принимает объект-делегат, инкапсулирующий метод продолжения. Метод ContinueWith() можно вызвать как до, так и после старта задачи.
t1.ContinueWith(task => Console.WriteLine("After task " + task.Id));
Методы Wait(), WaitAll() и WaitAny() останавливают основной поток до завершения задачи (или задач). Перегруженные версии методов позволяют задать период ожидания завершения и токен отмены.
t1.Wait(1000); Task.WaitAll(t1, t2);
Класс Task<T> наследуется от Task и описывает задачу, возвращающую значение типа T. Дополнительно к элементам базового класса, Task<T> объявляет свойство Result для хранения вычисленного значения. Конструкторы класса Task<T> принимают аргументы типа Func<T> и Func<object, T> (опционально –
аргументы типа CancellationToken и TaskCreationOptions).
138
Func<int> func = () => {
Thread.Sleep(2000); |
|
return 100; |
|
}; |
|
var task = new Task<int>(func); |
|
Console.WriteLine(task.Status); |
// Created |
task.Start(); |
|
Console.WriteLine(task.Status); |
// WaitingToRun |
task.Wait(); |
|
Console.WriteLine(task.Result); |
// 100 |
Класс TaskFactory содержит набор методов, соответствующих некоторым
сценариям использования задач – StartNew(), FromAsync(), ContinueWhenAll(),
ContinueWhenAny(). Экземпляр TaskFactory доступен через статическое свой-
ство Task.Factory.
Task.Factory.StartNew(() => {
Thread.Sleep(2000); Console.WriteLine("Done");
});
33.2. Параллелизм при императивной обработке данных
Класс System.Threading.Tasks.Parallel позволяет распараллеливать циклы и последовательность блоков кода. Эта функциональность реализована как набор статических методов For(), ForEach() и Invoke().
Методы Parallel.For() и Parallel.ForEach() являются параллельными
аналогами циклов for и foreach. Их использование корректно в случае независимости итераций цикла, то есть, если ни в одной итерации не используется результаты работы предыдущих итераций.
Существует несколько перегруженных вариантов метода Parallel.For(), однако любой из них подразумевает указание начального и конечного значения счётчика (типа int или long) и тела цикла в виде объекта делегата. В качестве примера использования Parallel.For() приведём код, перемножающий две квадратные матрицы (m1 и m2 – исходные матрицы, result – их произведение, size – размер матриц):
Parallel.For(0, size, i =>
{
for (int j = 0; j < size; j++)
{
result[i, j] = 0;
for (int k = 0; k < size; k++) result[i, j] += m1[i, k] * m2[k, j];
}
});
139