Читаем без скачивания Системное программирование в среде Windows - Джонсон Харт

• Использование для защиты объекта очереди объекта CRITICAL_SECTION вместо мьютекса также может привести к повышению производительности. Однако в этом случае вместо функции SignalObjectAndWait следует использовать функцию EnterCriticalSection с последующим ожиданием события. Этот альтернативный подход иллюстрируется двумя файлами — QueueObjCS.с и QueueObjCS_Sig.с, находящимися на Web-сайте книги.

• На Web-сайте находятся два других файла с исходными кодами — QueueObj_noSOAW.с и QueueObjSig_noSOAW.с, в которых функция SignalObjectAndWait не используется и которые обеспечивают выполнение программы под управлением Windows 9x.

• Результаты, приведенные в приложении В, свидетельствуют о нелинейном поведении производительности при большом количестве потоков, состязающихся за доступ к очереди. Проекты для каждой из альтернативных стратегий содержатся на Web-сайте книги; эти проекты соответствуют различным вариантам конвейерной системы ThreeStage, описанной в следующих разделах.

•Резюмируя, следует подчеркнуть, что свойства очередей могут быть расширены таким образом, чтобы очередь могла совместно использоваться несколькими процессами и обеспечивать отправку или получение сразу нескольких сообщений за одну операцию. В то же время, некоторого выигрыша в производительности можно добиться за счет использования сигнальной модели, объектов CRITICAL_SECTIONS или функции SignalObjectAndWait. Соответствующие результаты представлены в приложении В. 

Пример: использование очередей в многоступенчатом конвейере

Модель "хозяин/рабочий", во всех ее вариациях, является одной из наиболее популярных моделей многопоточного программирования, а программа 8.2 представляет простую модель "производитель/потребитель", являющуюся частным случаем более общей конвейерной модели (pipeline model).

В другом важном частном случае имеется один главный поток, который производит единичные рабочие задания (work units) для ограниченного количества рабочих потоков и помещает их в очередь. Такая методика может оказаться полезной при создании масштабируемого сервера с большим количеством клиентов (число которых может достигать тысячи и более), когда возможность выделения независимого рабочего потока для каждого клиента весьма сомнительна. В главе 14 задача создания масштабируемого сервера обсуждается в контексте портов завершения ввода/вывода.

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

В программе 10.5 (ThreeStage.c) предусмотрено создание нескольких этапов производства и потребления, на каждой из которых поддерживается очередь рабочих заданий, подлежащих обработке. Каждая очередь имеет ограниченную, конечную длину. Всего существует три конвейерных ступени, соединяющих четыре этапа обработки. Программа имеет следующую структуру:

• Производители (producers) периодически создают единичные сообщения, дополненные контрольными суммами, используя для этого ту же функцию, что и в программе 8.2, если не считать того, что в каждом сообщении содержится дополнительное поле адресата, указывающее поток потребителя (consumer), для которой предназначено это сообщение, причем каждый производитель связывается только с одним потребителем. Количество пар "производитель/потребитель" задается в виде параметра командной строки. Далее производитель посылает одиночное сообщение передающему потоку (transmitter), помещая его в очередь передачи сообщений. Если очередь заполнена, производитель ждет, пока ее состояние не изменится.

• Передающий поток объединяет имеющиеся единичные сообщения (но не более пяти за один раз) и создает одно передаваемое сообщение, которое содержит заголовок и ряд единичных сообщений. Затем передающий поток помещает каждое передаваемое сообщение в очередь приема сообщений (receiver), блокируясь, если очередь заполнена. В общем случае передатчик и приемник могут связываться между собой через сетевое соединение. Произвольно выбранное здесь значение коэффициента блокирования (blocking factor), равное 5:1, легко поддается регулировке. 

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

• Каждый поток потребителя получает одиночные сообщения по мере их поступления и записывает сообщение в файл журнала регистрации.

Блок-схема системы представлена на рис. 10.1. Обратите внимание, что эта система моделирует сетевое соединение, в котором сообщения, относящиеся к различным парам "отправитель/получатель" объединяются и передаются по общему каналу связи. 

Рис. 10.1. Многоступенчатый конвейер

В программе 10.5 предложен вариант реализации, в котором используются функции очереди из программы 10.4. Функции генерации и отображения сообщений здесь не представлены, но они взяты из программы 8.1. При этом, наряду с контрольными суммами и данными, в блоки сообщений введены поля производителя и адресата.

/* Глава 10. ThreeStage.с */

/* Трехступенчатая система производитель/потребитель. */

/* Использование: ThreeStage npc goal. */

/* Запустить "npc" пар потоков производителя и потребителя. */

/* Каждый производитель должен сгенерировать в общей сложности */

/* "goal" сообщений, каждое из которых снабжается меткой, указывающей */

/* потребителя, для которого оно предназначено. */

/* Сообщения отправляются "передающему потоку", который, прежде чем */

/* отправить группу сообщений "принимающему потоку", выполняет некоторую*/

/* дополнительную обработку. Наконец, принимающий поток отправляет сообщения потокам потребителя. */

#include "EvryThng.h"

#include "SynchObj.h"

#include "messages.h"

#include <time.h>

#define DELAY_COUNT 1000

#define MAX_THREADS 1024

/* Размеры и коэффициенты блокирования очередей. Эти величины являются */

/* произвольными и могут регулироваться для обеспечения оптимальной */

/* производительности. Текущие значения не являются сбалансированными. */

#define TBLOCK_SIZE 5 /*Передающий поток формирует группы из 5 сообщений.*/

#define TBLOCK_TIMEOUT 50 /*Интервал ожидания сообщений передающим потоком.*/

#define P2T_QLEN 10 /* Размер очереди "производитель/передающий поток". */

#define T2R_QLEN 4 /*Размер очереди "передающий поток/принимающий поток".*/

#define R2C_QLEN 4 /* Размер очереди "принимающий поток/потребитель" -- */

/* для каждого потребителя существует только одна очередь.*/

DWORD WINAPI producer(PVOID);

DWORD WINAPI consumer(PVOID);

DWORD WINAPI transmitter(PVOID);

DWORD WINAPI receiver(PVOID);

typedef struct _THARG {

 volatile DWORD thread_number;

 volatile DWORD work_goal; /* Используется потоками производителей. */

 volatile DWORD work_done; /* Используется потоками производителей и потребителей. */ '

 char future[8]; 

} THARG;

/* Сгруппированные сообщения, посылаемые передающим потоком потребителю.*/

typedef struct t2r_msg_tag {

 volatile DWORD num_msgs; /* Количество содержащихся сообщений. */

 msg_block_t messages[TBLOCK_SIZE];

} t2r_msg_t;

queue_t p2tq, t2rq, *r2cq_array;

static volatile DWORD ShutDown = 0;

static DWORD EventTimeout = 50;

DWORD _tmain(DWORD argc, LPTSTR * argv[]) {

 DWORD tstatus, nthread, ithread, goal, thid;

 HANDLE *producer_th, *consumer_th, transmitter_th, receiver_th;

 THARG *producer_arg, *consumer_arg;

 nthread = atoi(argv[1]);

 goal = atoi(argv[2]);

 producer_th = malloc(nthread * sizeof(HANDLE));

 producer_arg = calloc(nthread, sizeof(THARG));

 consumer_th = malloc(nthread * sizeof(HANDLE));

 consumer_arg = calloc(nthread, sizeof(THARG));

 q_initialize(&p2tq, sizeof(msg_block_t), P2T_QLEN); 

 q_initialize(&t2rq, sizeof(t2r_msg_t), T2R_QLEN);

 /* Распределить ресурсы, инициализировать очереди "принимающий поток/потребитель" для каждого потребителя. */

 r2cq_array = calloc(nthread, sizeof(queue_t));

 for (ithread = 0; ithread < nthread; ithread++) {

  /* Инициализировать очередь r2с для потока данного потребителя. */

  q_initialize(&r2cq_array[ithread], sizeof(msg_block_t), R2C_QLEN);

  /* Заполнить аргументы потока. */

  consumer_arg[ithread].thread_number = ithread;

  consumer_arg[ithread].work_goal = goal;

  consumer_arg[ithread].work_done = 0;

  consumer_th[ithread] = (HANDLE)_beginthreadex(NULL, 0, consumer, (PVOID)&consumer_arg[ithread], 0, &thid);