Параллельное и распределенное программирование на С++

В листинге 4.9 объявляется N очередей для N этапов. Начальный поток помещает все порции входных потоков в очередь первого этапа, а затем создает все потоки, необходимые для выполнения всех этапов. Каждый этап содержит свой цикл событий. Поток выполнения находится в состоянии ожидания до тех пор, пока в его очереди не появится порция входных данных. Внутренний цикл продолжается до опустения соответствующей очереди. Порция входных данных извлекается из очереди, обрабатывается, а затем помещается в очередь следующего этапа обработки (следующего потока выполнения).

Использование модели «изготовитель-потребитель»

В модели «изготовитель-потребитель» поток- «изготовитель» готовит данные, «потребляемые» потоком-«потребителем» (причем таких потоков-«потребителей" может быть несколько). Данные хранятся в блоке памяти, разделяемом всеми потока, как изготовителем, так и потребителями. В листинге 4.10 представлен скелет программы реализации модели «изготовитель-потребитель» (эта модель также использовалась в программах 4.5, 4.6 и 4.7).

Листинг 4.10. Скелет программы реализации модели «изготовитель-потребитель»

pthread_mutex_t Mutex = PTHREAD_MUTEX_INITIALIZER

pthread_t Thread[2]

Queue

// initial thread

{

pthread_create(&(Thread[1]...producer...);

pthread_create(&(Thread[2]...consumer...);

//...

}

void *producer(void *X)

{

loop

perform work

pthread_mutex_lock(&Mutex)

enqueue data

pthread_mutex_unlock(&Mutex)

signal consumer

//...

until done

}

void *consumer(void *X)

{

loop

suspend until signaled

loop while(Data Queue not empty)

pthread_mutex_lock(&Mutex)

dequeue data

pthread_mutex_unlock(&Mutex)

perform work

end loop

until done

}

//

В листинге 4.9 начальный поток создает оба потока: «изготовителя» и «потребителя». Поток- «изготовитель» содержит цикл, в котором после выполнения некоторых действий блокируется мьютекс для совместно используемой очереди, чтобы поместить в нее подготовленные для потребителя данные. После этого «изготовитель» деблокирует мьютекс и посылает сигнал потоку- «потребителю» о том, что ожидаемые им данные уже находятся в очереди. Поток- «изготовитель» выполняет итерации цикла до тех пор, пока не будет выполне н а вся работа. Поток- «потребитель» также выполняет цикл, в котором он приостанавливается до тех пор, пока не получит сигнал. Во внутреннем цикле поток- «потребитель» обрабатывает все данные до тех пор, пока не опустеет очередь. Он блокирует мьютекс для разде л яемой очереди перед извлечением из нее данных и деблокирует мьютекс после этого. Затем он выполняет обработку извлеченных данных. В программе 4.6 поток-«потребитель» помещает свои результаты в файл. Вместо файла может быть использована другая структура данных. Зачастую потоки-«потребители» играют две роли: как потребителя, так и изготовителя. Сначала возможно «исполнение» роли потребителя необработанных данных, подготовленных потоком-«изготовителем», а затем поток играет роль «изготовителя», когда он обрабатывает данные, сохраняемые в другой совместно используемой очереди, «потребляемой» другим потоком.

Создание многопоточных объектов

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

В любом случае потоки объявляются в рамках объекта и создаются одной из функций-членов (например, конструктором). Потоки могут затем выполнять некоторые независимые функции (функции, определенные вне объекта), которые вызывают функции-члены глобальных объектов. Это — один из способов создания многопоточного объекта. Пример многопоточного объекта представлен в листинге 4.10.

// Листинг 4.11 . Объявление и определение многопоточного

// объекта

#include

#include

#include

void *taskl(void *);

void *task2(void *);

class multithreaded_object {

pthread_t Threadl,

Thread2; public:

multithreaded_object(void);

int cl(void);

int c2(void);

//.. .

);

multithreaded_object::multithreaded_object(void) {

//. . .