Системное программирование в среде Windows

 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);
 producer_arg[ithread].thread_number = ithread;
 producer_arg[ithread].work_goal = goal;
 producer_arg[ithread].work_done = 0;
 producer_th[ithread] = (HANDLE)_beginthreadex(NULL, 0, producer, (PVOID)&producer_arg[ithread], 0, &thid);
 }
 transraitter_th = (HANDLE)_beginthreadex(NULL, 0, transmitter, NULL, 0, &thid);
 receiver_th = (HANDLE)_beginthreadex (NULL, 0, receiver, NULL, 0, &thid);
_tprintf(_T("ХОЗЯИН: Выполняются все потоки\n"));
 /* Ждать завершения потоков производителя. */
 for (ithread = 0; ithread < nthread; ithread++) {
 WaitForSingleObject(producer_th[ithread], INFINITE);
 _tprintf(_T("ХОЗЯИН: производитель %d выработал %d единичных сообщений\n"), ithread, producer_arg[ithread].work_done);
 }
 /* Производители завершили работу. */
 _tprintf(_T("ХОЗЯИН: Все потоки производителя выполнили свою работу.\n"));
 /* Ждать завершения потоков потребителя. */
 for (ithread = 0; ithread < nthread; ithread++) {
 WaitForSingleObject(consumer_th[ithread], INFINITE);
 _tprintf(_T("ХОЗЯИН: потребитель %d принял %d одиночных сообщений\n"), ithread, consumer_arg[ithread].work_done);
 }
 _tprintf(_T("ХОЗЯИН: Все потоки потребителя выполнили свою работу.\n"));
 ShutDown = 1; /* Установить флаг завершения работы. */
 /* Завершить выполнение и перейти в состояние ожидания передающих и принимающих потоков. */
 /* Эта процедура завершения работает нормально, поскольку и передающий,*/
 /* и принимающий потоки не владеют иными ресурсами, кроме мьютекса, */
 /* которые они могли бы покинуть по завершении выполнения, не уступив прав владения ими. Можете ли вы улучшить эту процедуру? */
 TerminateThread(transmitter_th, 0);
 TerminateThread(receiver_th, 0);
 WaitForSingleObject(transmitter_th, INFINITE);
 WaitForSingleObject(receiver_th, INFINITE);
 q_destroy(&p2tq);
 q_destroy(&t2rq);
 for (ithread = 0; ithread < nthread; ithread++) q_destroy(&r2cq_array [ithread]);
 free(r2cq_array);
 free(producer_th);
 free(consumer_th);
 free(producer_arg);
 free(consumer_arg);
 _tprintf(_T("Система завершила работу. Останов системы\n"));
 return 0;
}

DWORD WINAPI producer(PVOID arg) {
 THARG * parg;
 DWORD ithread, tstatus;
 msg_block_t msg;
 parg = (THARG *)arg;
 ithread = parg->thread_number;
 while (parg->work_done < parg->work_goal) {
 /* Вырабатывать единичные сообщения, пока их общее количество */
 /* не станет равным "goal". */