Многопоточное программирование (стр. 5 )

Состав Parallel Extensions

·  TPL (Task Parallel Library)

o  Parallel. For, Parallel. ForEach, Parallel. Invoke

·  PLINQ (Parallel Language-Integrated Query)

o  LINQ to Objects, LINQ to XML

o  AsParallel()

·  CDS (Coordination Data Structures)

o  Типы структур данных, используемых для синхронизации и координации выполнения параллельных задач

Особенности Parallel Extensions

·  Использование задач (Task)

o  Задача – это небольших участок кода, реализованный в виде лямбда-функции, который может исполняться независимо от остальных задач

o  TPL предусматривает разбиение блоков кода или итераций циклов на задачи

o  Задачи могут исполняться параллельно, при использовании общих ресурсов задачи должны обеспечивать синхронизацию

·  Использование специального планировщика задач

o  Он обеспечивает распределение задач по рабочим потокам

o  Количество рабочих потоков по умолчанию равно количеству вычислительных ядер

o  Каждый рабочий поток имеет собственную очередь задач, ожидающих исполнения

o  Каждый процессор имеет собственный пул потоков

o  Каждый процессор может заимствовать задачи из других процессоров

Состав Task Parallel Library (TPL)

·  System. Threading. Tasks. Task http://msdn. /en-us/library/system. threading. tasks. task. aspx

o  Класс, представляющий собой задачу, не возвращающую значение

·  System. Threading. Tasks. Task<TResult> http://msdn. /en-us/library/dd321424.aspx

o  Класс, представляющий собой задачу, имеющую возвращаемое значение

·  System. Threading. Tasks. Parallel http://msdn. /en-us/library/system. threading. tasks. parallel. aspx

o  Предназначен для распараллеливания циклов и последовательностей блоков кода за счёт использования задач

o  Имеет набор статических методов For, ForEach и Invoke

o  Parallel. For и Parallel. ForEach корректно использовать если ни в одной итерации цикла не используется результаты работы предыдущих итераций, например, изменённое значение предыдущей ячейки массива

o  Допустимо использовать Parallel. For и Parallel. ForEach в случае когда между итерациями передаётся общая информация, например, при расчёте суммы элементов массива

o  Parallel. Invoke позволяет распараллелить исполнение независимых блоков операторов, например, для рекурсивных алгоритмов

·  System. Threading. Tasks. TaskScheduler http://msdn. /en-us/library/system. threading. tasks. taskscheduler. aspx

o  Класс, позволяющий управлять экземпляром очереди выполнения задач за счёт постановки задачи в очередь, извлечения задачи из очереди и т. п.

Пример использования класса Task

Task taskExample = new Task(() =>

{

for (int i = 0; i < 10; i++)

{

Console. WriteLine("Task loop iteration: {0}.", i);

}

throw new ArgumentException("This is sample exception.");

});

taskExample. Start();

for (int i = 0; i < 10; i++)

{

Console. WriteLine("Main thread loop iteration: {0}.", i);

}

try

{

taskExample. Wait();

}

catch (AggregateException e)

{

Console. WriteLine("Exception occured: {0} {1}", e. GetType(), e. Message);

foreach (var ie in e. InnerExceptions)

{

Console. WriteLine("Inner exception: {0} {1}", ie. GetType(), ie. Message);

}

}

Пример использования класса Task<Result>

Task<int> taskExample = new Task<int>(() =>

{

int j = 0;

for (int i = 0; i < 10; i++)

{

Console. WriteLine("Task loop iteration: {0}.", i);

j += 1 << i;

}

return j;

});

taskExample. Start();

// Some code

taskExample. Wait();

Console. WriteLine("Task result: {0}", taskExample. Result);

Создание параллельного цикла на основе Parallel.For

¡  Обычный цикл (однопоточный)

int[] arraySample = { 1, 2, 3, 4, 5 };

for (int i = 0; i < arraySample. Length; i++)

{

arraySample[i] += 1;

Console. WriteLine("arraySample[{0}]: {1}",

i, arraySample[i]);

}

·  Параллельный цикл на основе Parallel. For

o  Из-за одновременного запуска задач в разных потоках порядок вывода на консоль информации об элементах массива может отличаться от традиционного: , и быть, например:

int[] arraySample = { 1, 2, 3, 4, 5 };

Parallel. For(0, arraySample. Length, i =>

{

arraySample[i] += 1;

Console. WriteLine("arraySample[{0}]: {1}",

i, arraySample[i]);

});

Дополнения области имён System. Threading

·  System. Threading. CountdownEvent http://msdn. /en-us/library/system. threading. countdownevent. aspx

o  Вызывает событие после указанного количества вызовов метода Decrement

·  System. Threading. LazyInitializer http://msdn. /en-us/library/system. threading. lazyinitializer. aspx

o  Позволяет выполнять потокобезопасную ленивую инициализацию

Состав Coordination Data Structures (CDS)

·  System. Collections. Concurrent. ConcurrentStack<T> http://msdn. /en-us/library/dd267331.aspx

o  Потокобезопасный вариант стека LIFO

·  System. Collections. Concurrent. ConcurrentQueue<T> http://msdn. /en-us/library/dd267265.aspx

o  Потокобезопасный вариант очереди FIFO

·  System. Collections. Concurrent. BlockingCollection<T> http://msdn. /en-us/library/dd381957.aspx

o  Потокобезопасный вариант структуры данных, реализующей шаблон «издатель-потребитель»: издатель формирует данные, потребитель их по очереди обрабатывает

·  System. Collections. Concurrent. ConcurrentDictionary<TKey, TValue> http://msdn. /en-us/library/dd287191.aspx

o  Потокобезопасный вариант структуры коллекции «ключ-значение»

·  System. Collections. Concurrent. ConcurrentBag<T> http://msdn. /en-us/library/dd381779.aspx

o  Потокобезопасный вариант несортированного динамического массива объектов

·  System. Collections. Concurrent. Partitioner http://msdn. /en-us/library/system. collections. concurrent. partitioner. aspx

·  System. Collections. Concurrent. Partitioner<TSource> http://msdn. /en-us/library/dd381768.aspx

·  System. Collections. Concurrent. OrderablePartitioner<TSource> http://msdn. /en-us/library/dd394988.aspx

o  Классы, предназначенные для разделения структур данных (массивов, списков) на разделы (partitions), позволяющие осуществить балансировку вычислительной нагрузки при их обработке

Использование PLINQ

·  Основан на классе System. Linq. ParallelEnumerable http://msdn. /en-us/library/system. linq. parallelenumerable. aspx

·  Основные методы расширения класса ParallelEnumerable

o  AsParallel http://msdn. /en-us/library/system. linq. parallelenumerable. asparallel. aspx

§  Включает распараллеливание выполнения запроса (по возможности)

o  AsSequential<TSource> http://msdn. /en-us/library/dd383949.aspx

§  Отключает распараллеливание для запроса

·  При возможности распараллеливания выполнение частей запроса осуществляется посредством задач (класс Task)

·  Дополнительная информация http://msdn. /en-us/library/dd997425.aspx

Пример использования PLINQ

Обычная выборка данных (однопоточная)

int[] arraySample = { 1, 2, 3, 4, 5 };

string[] strings =

(from i in arraySample

where i > 2

select i. ToString()).ToArray();

Выборка данных при использовании PLINQ

int[] arraySample = { 1, 2, 3, 4, 5 };

string[] strings =

(from i in arraySample.AsParallel()

where i > 2

select i. ToString()).ToArray();

7.1  Совершенствование пулов потоков

Преимущества использования пула потоков

·  Возможность одновременно обрабатывать ограниченное количество запросов клиентов с постановкой в очередь ожидающих запросов

·  Эффективно при наличии значительного числа операций ввода-вывода и ожидания внешних сервисов

·  Повышение эффективности использования аппаратных ресурсов системы

·  Возможность быстро запускать обработку поступившего запроса при наличии одного или нескольких свободных потоков в пуле

Изменения в работе пула потоков в.NET 4

·  Причины

o  Введение задач (класс Task), не порождающих отдельных потоков, а распределяемых между потоками в пуле

o  Необходимость диспетчеризации и управления задачами, а не потоками

o  Необходимость балансировки вычислительной нагрузки в многоядерных системах

·  Сделанные изменения

o  Каждому потоку из пула потоков назначается локальная очередь, содержащая задачи, ожидающие исполнения в этом потоке

o  Добавлен механизм заимствования задач между локальными очередями потоков, по правилу LIFO

7.2  Транзакционная память

·  Транзакционная память – модель работы с памятью

·  Обеспечивает механизм легковесных транзакций для потоков управления, выполняемых в общем адресном пространстве, при обращении к памяти

·  Гарантирует атомарность и изолированность параллельно выполняемых задач

·  Альтернатива использованию дорогостоящих блокировок при работе с данными

·  В основе системы лежат два механизма

o  Управление версиями данных (data versioning)

o  Обнаружение конфликтов (conflict detection)

Семантика

·  Для каждой транзакции система исполнения обеспечивает ACID-свойства

·  Если транзакция противоречит некоторой транзакции, зафиксированной в процессе своего выполнения (то есть, читает или пишет измененный элемент данных), то транзакция откатывается – результаты её действий отменяются, и она может быть перезапущена (чаще всего системы STM автоматически перезапускают транзакцию в случае конфликтов)

·  Если необработанное исключение покидает пределы транзакции, то транзакция откатывается и не перезапускается

·  Вложенные транзакции фиксируются вместе с родительской транзакцией (то есть, относительно фиксации дерево транзакций выпрямляется в список), но откатываются по отдельности (например, исключение, выброшенное вложенной транзакцией, но обработанное в родительской, отменит только вложенную)

Управление версиями данных

·  Каждой транзакции система выделяет приватную область памяти, недоступную другим транзакциями, в которой транзакция хранит результаты своей работы

·  Существуют две основные реализации области памяти

o  Работа осуществляется с копией объекта и результат становится виден только при завершении транзакции

o  Работа осуществляется непосредственно с объектом данных, при этом создаются копии каждой изменённой ячейки памяти

Обнаружение конфликтов

·  Коллизии возникают в тех случаях, когда две или более транзакции, выполняемые параллельно, обращаются к одному и тому же фрагменту данных в тот момент, когда одна из них записывает новую версию этих данных

·  Средствами обнаружения и разрешения конфликтов обеспечивается комплекс характеристик ACID, эти средства включают в себя два набора адресов, соответствующих каждой транзакции; в один записываются адреса, откуда она читает, во второй — адреса, куда она пишет

·  При разрешении коллизий этот механизм может действовать по одному из двух сценариев

o  Пессимистический сценарий – предполагает, что конфликтные ситуации возникают непосредственно в процессе выполнения транзакций, они обнаруживаются на ранних моментах и могут быть разрешены путем приостановления одной из транзакций или прерыванием одной из транзакций с тем, чтобы возобновить ее позднее

o  Оптимистический сценарий – исходит из того, что вероятность бесконфликтного завершения транзакций велика, поэтому текущей транзакции позволяют завершиться до конца, а затем только производится проверка

·  Обнаружение конфликтов возможно на уровне объектов данных либо на уровне отдельных слоёв (участки памяти, поля объекта)

Реализации

·  Существует два способа реализации:

o  Программная транзакционная память (STM)

o  Специальная аппаратная поддержка (HTM)

Программная транзакционная память

·  Альтернатива традиционным способам блокировок используемым в многопоточно среде

·  Реализуется за счет запуска сложной системы исполнения, работа системы исполнения схожа с работой системы сборки мусора

·  Доступна при использовании сторонних библиотек или расширений языка и/или виртуальной машины

Пример использования STM для .NET

·  Проверка целостности справочника

bool ValidateMapping(string name, string phone)

{

string outPhone = null;

string outName = null;

atomic

{

outPhone = name2phone. TryGetValue(name);

outName = phone2name. TryGetValue(phone);

}

// Код проверки соответствия

// номера телефона и абонента

}

Пример использования STM для Java

Библиотеки STM

·  Java

o  Multiverse (http://multiverse. codehaus. org)

o  Deuce (http://www. deucestm. org/)

o  SCAT (http://wasp. cs. washington. edu/wasp_scat. html)

·  .NET

o  NSTM (http://code. /p/nstm/)

o  (http://msdn. /en-us/devlabs/ee334183.aspx)

o  LibCMT (http:///projects/libcmt)

Аппаратная транзакционная память

·  Позволяет перенести частично или полностью функции, связанные с управлением версиями данных и предупреждением возникающих конфликтов, которые выполняет STM, на аппаратный уровень

·  В основе аппаратной поддержки – иерархия кэш-памяти (cache hierarchy) и протокол согласования кэш-памяти (cache coherence protocol); они обеспечивают управление версиями данных и обнаружение конфликтных ситуаций

Подробности реализации

·  Для отслеживания наборов прочитанных и измененных объектов каждый блок кэша помечается служебными битами R и W, которые устанавливаются при первом доступе к блоку внутри транзакции по чтению или записи соответственно

·  Блоки кэша, входящие в набор измененных объектов, действуют как буфера записи и не выталкиваются в память до фиксации транзакции

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

Минусы использования транзакционной памяти

·  Многие аспекты семантики и реализации TM все еще остаются предметом активных исследований

·  Проблемы с производительностью при малых масштабах вычислений

·  Проблемы с откатом транзакций для операций ввода-вывода

Перспективы развития

·  Система транзакций хорошо зарекомендовала себя при работе в многопоточном окружении в базах данных

·  Транзакционная модель работы с памятью повышает уровень абстракции для понимания параллельных задач и помогает избежать многочисленных коварных ошибок параллельного программирования

Приложения

Пример расчёта чисел Фибоначчи пример программы на C#.

/// <summary>

/// Represents Fibonacci number calculation routines.

/// </summary>

public class Fibonacci

{

private int stepsNumber;

private int result;

private ManualResetEvent calculationComplete;

/// <summary>

/// Gets this Fibonacci calculation argument.

/// </summary>

public int StepsNumber { get { return stepsNumber; } }

/// <summary>

/// Gets this Fibonacci calculation result.

/// </summary>

public int Result { get { return result; } }

/// <summary>

/// Creates new Fibonacci number calculation.

/// </summary>

/// <param name="stepsNumber">Fibonacci calculation

/// argument.</param>

/// <param name="calculationComplete">Event,

/// signaling calculation complete.</param>

public Fibonacci(int stepsNumber,

ManualResetEvent calculationComplete)

{

this. stepsNumber = stepsNumber;

this. calculationComplete = calculationComplete;

}

/// <summary>

/// Starts asynchronous calculation in thread pool.

/// </summary>

/// <param name="threadContext">Paramenter

/// passing from thread pool.</param>

public void ThreadPoolCallback(object threadContext)

{

int threadIndex = (int)threadContext;

Debug. Print("Thread {0} started...", threadIndex);

result = Calculate(stepsNumber);

Debug. Print("Thread {0} result calculated...",

threadIndex);

calculationComplete. Set();

}

/// <summary>

/// Synchronously calculates Fibonacci number

/// for specified argument.

/// </summary>

/// <param name="stepsNumber">Fibonacci

/// number argument.</param>

/// <returns>Fibonacci number result.</returns>

public static int Calculate(int stepsNumber)

{

if (stepsNumber <= 1)

{

return stepsNumber;

}

return Calculate(stepsNumber - 1) +

Calculate(stepsNumber - 2);

}

}

/// <summary>

/// Contains program entry point routines.

/// </summary>

public class Program

{

private const int numberOfCalculations = 64;

private static ManualResetEvent[] doneEvents =

new ManualResetEvent[numberOfCalculations];

private static int[] arguments =

CreateArguments();

private static int[] results =

new int[numberOfCalculations];

private static Fibonacci[] calculations =

new Fibonacci[numberOfCalculations];

/// <summary>

/// Program entry point.

/// </summary>

public static void Main()

{

TimeSpan elapsedTime;

Console. WriteLine("Launching {0} synchronous tasks...",

numberOfCalculations);

elapsedTime = CalculateSynchronous();

Console. WriteLine("All synchronous calculations are complete.");

PrintSynchronousResults(elapsedTime);

Console. WriteLine("Launching {0} asynchronous tasks...",

numberOfCalculations);

elapsedTime = CalculateAsynchronous();

Console. WriteLine("All asynchronous calculations are complete.");

PrintAsynchronousResults(elapsedTime);

}

private static int[] CreateArguments()

{

Random random = new Random();

int[] arguments = new int[numberOfCalculations];

for (int i = 0; i < numberOfCalculations; i++)

{

arguments[i] = random. Next(20, 40);

}

return arguments;

}

private static TimeSpan CalculateSynchronous()

{

DateTime dateStart = DateTime. Now;

for (int i = 0; i < numberOfCalculations; i++)

{

results[i] = Fibonacci. Calculate(arguments[i]);

}

DateTime dateFinish = DateTime. Now;

return dateFinish - dateStart;

}

private static TimeSpan CalculateAsynchronous()

{

DateTime dateStart = DateTime. Now;

for (int i = 0; i < numberOfCalculations; i++)

{

doneEvents[i] = new ManualResetEvent(false);

Fibonacci f = new Fibonacci(arguments[i], doneEvents[i]);

calculations[i] = f;

ThreadPool. QueueUserWorkItem(f. ThreadPoolCallback, i);

}

WaitHandle. WaitAll(doneEvents);

DateTime dateFinish = DateTime. Now;

return dateFinish - dateStart;

}

private static void PrintAsynchronousResults(

TimeSpan elapsedTime)

{

foreach (Fibonacci f in calculations)

{

PrintResult(f. StepsNumber, f. Result);

}

PrintElapsedTime(elapsedTime);

}

private static void PrintSynchronousResults(

TimeSpan elapsedTime)

{

for (int i = 0; i < numberOfCalculations; i++)

{

PrintResult(arguments[i], results[i]);

}

PrintElapsedTime(elapsedTime);

}

private static void PrintResult(

int argument, int result)

{

Console. WriteLine("Fibonacci({0}) = {1}",

argument, result);

}

private static void PrintElapsedTime(

TimeSpan elapsedTime)

{

Console. WriteLine("Elapsed time: {0}",

elapsedTime);

}

}

Пример перемножения матриц на Java.

package ru. supercorporation;

import java. util. Arrays;

import java. util. List;

import java. util. Vector;

import java. util. concurrent.*;

import java. util. concurrent. atomic. AtomicInteger;

/**

* Math matrix class.

*/

public class Matrix {

protected final int[][] data;

/**

* Get matrix table. [row][column]

*

* @return matrix table

*/

public int[][] getData() {

return data;

}

/**

* @param rawData matrix table

*/

public Matrix(int[][] rawData) {

data = rawData;

}

/**

* Matrix factory.

*

* @param rows rows

* @param columns cols

* @param lessThen fills with random numbers 0&lt;=x&lt;lessThen

* @return matrix

*/

public static Matrix createRandomMatrix(int rows, int columns, int lessThen) {

int[][] raw = new int[rows][columns];

for (int i = 0; i < rows; i++) {

for (int j = 0; j < columns; j++) {

raw[i][j] = (int) (Math. random() * lessThen);

}

}

return new Matrix(raw);

}

@Override

public boolean equals(Object obj) {

if (this == obj) {

return true;

} else if (obj instanceof Matrix) {

int[][] data2 = ((Matrix) obj).getData();

for (int i = 0; i < data. length; i++) {

for (int j = 0; j < data2[i].length; j++) {

if (data[i][j] != data2[i][j]) {

return false;

}

}

}

return true;

} else {

return false;

}

}

/**

* @return rows

*/

public int getHeight() {

return data. length;

}

/**

* @return columns

*/

public int getWidth() {

return data[0].length;

}

@Override

public String toString() {

return Arrays. deepToString(data);

}

/**

* @param a first matrix raw data

* @param b second matrix raw data

* @param result resulting matrix

* @param row row number to multiply

*/

private void multiplySingleRow(

int[][] a, int[][] b, int[][] result,

int row) {

int cols = getWidth();

int rows = getHeight();

for (int j = 0; j < cols; ++j) {

result[row][j] += a[row][j] *

b[j][row];

}

}

/**

* Multiply by another matrix in single thread.

*

* @param matrix second matrix

* @return result of multiplication

*/

public Matrix multiply(Matrix matrix) {

int cols = getWidth();

int rows = getHeight();

int[][] result = new int[rows][cols];

for (int i = 0; i < rows; ++i) {

multiplySingleRow(data, matrix. data,

result, i);

}

return new Matrix(result);

}

/**

* Multiply by another matrix with help of {@link java. util. concurrent. ExecutorService}.

*

* @param matrix second matrix

* @param threadsNumber number of simultaneous threads

* @return result of multiplication

*/

public Matrix multiplyExecutorService(final Matrix matrix, int threadsNumber)

throws InterruptedException, ExecutionException {

final int[][] result = new int[getHeight()][getWidth()];

ExecutorService es = Executors. newFixedThreadPool(threadsNumber);

Vector<Callable<Object>> tasks = new Vector<Callable<Object>>(data. length);

// task per row

for (int i = 0; i < getHeight(); ++i) {

final int row = i;

tasks. add(new Callable<Object>() {

public Object call() throws Exception {

multiplySingleRow(data, matrix. data, result, row);

return null;

}

});

}

List<Future<Object>> futures = es. invokeAll(tasks);

// wait for complete

for (Future<Object> future : futures) {

future. get();

}

es. shutdown();

return new Matrix(result);

}

/**

* Multiply by another matrix. Multithreaded.

*

* @param matrix second matrix

* @param threadsNumber number of simultaneous threads

* @return result of multiplication

*/

public Matrix multiplyLockFreeThreads(final Matrix matrix, int threadsNumber)

throws InterruptedException {

final int[][] result = new int[getHeight()][getWidth()];

final int rows = getHeight();

Thread[] threads = new Thread[threadsNumber];

for (int i = 0; i < threadsNumber; i++) {

final int start = i * (rows / threadsNumber);

final int before = i!= threadsNumber - 1 ? (i + 1) *

(rows / threadsNumber) : rows;

threads[i] = new Thread(new Runnable() {

public void run() {

for (int i = start; i < before; ++i) {

multiplySingleRow(data,

matrix. data, result, i);

}

}

});

threads[i].start();

}

// wait for complete

for (Thread thread : threads) {

thread. join();

}

return new Matrix(result);

}

}

package ru. supercorporation;

import java. util. concurrent. ExecutionException;

public class MultiplyMain {

public static void main(String[] args) {

System. out. println("Starting Matrix multiply");

Matrix a = Matrix. createRandomMatrix(8000, 1000, 10);

Matrix b = Matrix. createRandomMatrix(1000, 8000, 10);

benchmarkMultiplication(a, b, 4);

System. out. println("Completed");

}

public static void benchmarkMultiplication(Matrix a, Matrix b, int threadsNumber) {

long time = System. currentTimeMillis();

Matrix result1 = a. multiply(b);

System. out. println("Computed in single thread. Done in "

+ (System. currentTimeMillis() - time));

time = System. currentTimeMillis();

try {

Matrix result2 = a. multiplyExecutorService(b, threadsNumber);

System. out. println("Computed using ExecutorService with " +

threadsNumber + " threads. Done in "

+ (System. currentTimeMillis() - time));

assert result1.equals(result2) : "answers not equals";

} catch (InterruptedException ie) {

assert false : ie;

} catch (ExecutionException e) {

assert false : e;

}

time = System. currentTimeMillis();

try {

Matrix result3 = a. multiplySimpleThreads(

b, threadsNumber);

System. out. println(

"Computed using simple Threads with " +

threadsNumber + " threads. Done in " +

(System. currentTimeMillis() - time));

assert result1.equals(result3) : "answers not equals";

} catch (InterruptedException ie) {

assert false : ie;

}

time = System. currentTimeMillis();

try {

Matrix result4 = a. multiplySimpleThreads(

b, threadsNumber);

System. out. println(

"Computed using lock free Threads with " +

threadsNumber + " threads. Done in " +

(System. currentTimeMillis() - time));

assert result1.equals(result4) : "answers not equals";

} catch (InterruptedException ie) {

assert false : ie;

}

}

}

Пример создания процесса в Java

ProcessBuilder pb =

new ProcessBuilder(

"myCommand",

"myArg1",

"myArg2");

Process p = pb. start();