Камерон Хьюз - Параллельное и распределенное программирование на С++

Во всех ситуациях использовать многопоточные версии функций попросту невозможно. В отдельных случаях многопоточные версии конкретных функций недоступны для данного компилятора или среды. Иногда один интерфейс функции не в состоянии сделать ее безопасной. Кроме того, программист может столкнуться с увеличением числа потоков в среде, которая изначально использовала функции, предназначенные для функционирования в однопоточной среде. В таких условиях обычно используются мьютексы. Например, программа имеет три параллельно выполняемых потока. Два из них, thread1 и thread2, параллельно выполняют функцию funcA(), которая не является безопасной для одновременной работы потоков. Третий поток, thread3, выполняет функцию funcB (). Для решения проблемы, связанной с функцией funcA (), возможно, достаточно заключить в защитную оболочку мьютекса доступ к ней со стороны потоков threadl и thread2:

thread1 thread2 thread3

{ { {

lock() lock() funcB()

funcA() funcA() }

unlock() unlock()

} }

При реализации таких защитных мер к функции funcA () в любой момент времени может получить доступ только один поток. Но проблемы на этом не исчерпываются. Если обе функции funcA() и funcB() небезопасны для выполнения потоками, они могут обе модифицировать глобальные или статические переменные. И хотя потоки thread1 и thread2 используют мьютексы для функции funcA (), поток thread3 может выполнять функцию funcB одновременно с любым из остальных потоков. В такой ситуации вполне вероятно возникновение условий «гонок», поскольку функции funcA () и funcB () могут модифицировать одну и ту же глобальную или статическую переменную.

Проиллюстрируем еще один тип условий «гонок», возникающих при использовании библиотеки i ostream. Предположим, у нас есть два потока, А и В, которые выводят данные в стандартный выходной поток, cout, который представляет собой типа ostream. При использовании операторов ">>" и "<<" вызываются методы объекта cout. Вопрос: являются ли эти методы безопасными? Если поток A от правляет сообщение «Мы существа разумные» объекту stdout, а поток В отправляет сообщение «Люди алогичные существа», то не произойдет ли «перемешивание» выходных данных, в результате которого может получиться сообщение вроде такого- " Мы Люди существа алогичные разумные существа»? В некоторых случаях безопасные для потоков функции реализуются как атомные. Атомные функции — это функции, ко торые, если их выполнение началось, не могут быть прерваны. Если операция ">>" для объекта cout реализована как атомная, то подобное «перемешивание» не произойдет. Если есть несколько обращений к оператору ">>", то они будут выполнены последовательно. Сначала отобразится сообщение потока А, а затем сообщение потока В или наоборот, хотя они вызвали функцию вывода одновременно. Это — пример преобразования параллельных действий в последовательные, которое обеспечит безопасность выполнения потоков. Но это не единственный способ обезопасить функцию. Если функция не оказывает неблагоприятного эффекта, она может смешивать свои операции. Например, если метод добавляет или удаляет элементы из структуры, которая не отсортирована, и этот метод вызывают два различных потока, то перемешивание их операций не даст неблагоприятного эффекта.

Если неизвестно, какие функции из библиотеки являются безопасными, а какие -нет, программист может воспользоваться одним из следующих вариантов действий.

• Ограничить использование всех опасных функций одним потоком.

• Не использовать безопасные функции вообще.

• Собрать все потенциально опасные функции в один набор механизмов синхронизации.

Еще один вариант — создать интерфейсные классы для всех опасных функций, которые должны использоваться в многопоточном приложении, т.е. опасные функции инкапсулируются в одном интерфейсном классе. Такой интерфейсный класс может быть скомбинирован с соответствующими объектами синхронизации с помощью наследования или композиции и использован специализированным классом. Такой подход устраняет возможность возникновения условий «гонок».

Выше в этой главе мы рассматривали делегирование работы в соответствии с конкретной стратегией или потоковой моделью. Итак, используются следующие распространенные модели:

• делегирование («управляющий-рабочий»");

• сеть с равноправными узлами;

• конвейер;

• «изготовитель-потребитель».

Каждая модель характеризуется собственной декомпозицией работ (Work Breakdown Structure — WBS), которая определяет, кто отвечает за создание потоков и при каких условиях они создаются. В этом разделе мы рассмотрим пример программы для каж дой модели, использующей функции библиотеки Pthread.

Мы рассмотрели два подхода к реализации модели делегирования при разделении мы на потоки. Вспомним: в модели делегирования один поток (управляющий) создает другие потоки (рабочие) и назначает каждому из них задачу. Управляющий поток делегирует каждому рабочему потоку задачу, которую он должен выполнить, путем задания некоторой функции. При одном подходе управляющий поток создает рабочие потоки как результат запросов, обращенных к системе. Управляющий поток обрабатывает запрос каждого типа в цикле событий. Как только событие произойдет, будет создан рабочий поток и ему будет назначена задача. Функционирование цикла событий в управляющем потоке и создание рабочих потоков продемонстрировано в листинге 4 .5.

// Листинг 4.5. Подход 1: скелет программы реализации II модели управляющего и рабочих потоков

//...

pthread_mutex_t Mutex = PTHREAD_MUTEX_INITIALIZER

int AvailableThreads

pthread_t Thread[Max_Threads]

void decrementThreadAvailability(void)

void incrementThreadAvailability(void)

int threadAvailability(void);

// boss thread

{

//...

if(sysconf(_SC_THREAD_THREADS_MAX) > 0){

AvailableThreads = sysconf(_SC_THREAD_THREADS_MAX)

}

else{

AvailableThreads = Default

}

int Count = 1;

loop while(Request Queue is not empty)

if(threadAvailability()){

Count++

decrementThreadAvailability()

classify request

switch(request type)

{

case X : pthread_create(&(Thread[Count])...taskX...)

case Y : pthread_create(&(Thread[Count])...taskY...)

case Z : pthread_create(&(Thread[Count])...taskZ...)

//...

}

}

else{

//free up thread resources

}

end loop

}

void *taskX(void *X)

{

// process X type request

incrementThreadAvailability()

return(NULL)

}

void *taskY(void *Y)

{

// process Y type request

incrementThreadAvailability()