Олег Цилюрик - QNX/UNIX: Анатомия параллелизма
6: [0000000]->[0001505] ... [0002007]->[0002007]
7: [0000000]->[0002007] ... [0002508]->[0002508]
8: [0000000]->[0002508] ... [0003010]->[0003010]
9: [0000000]->[0003010] ... [0003512]->[0003512]
10: [0000000]->[0003512] ... [0004014]->[0004014]
11: [0000000]->[0004014] ... [0004516]->[0004516]
12: [0000000]->[0004516] ... [0005017]->[0005018]
3: [0005501]->[0005501] ... [0006003]->[0006003]
5: [0008024]->[0008024] ... [0008526]->[0008526]
7: [0008038]->[0008526] ... [0009028]->[0009028]
4: [0009273]->[0009273] ... [0009775]->[0009775]
6: [0010377]->[0010377] ... [0010878]->[0010878]
8: [0010590]->[0010878] ... [0011380]->[0011380]
9: [0010952]->[0011380] ... [0011882]->[0011882]
12: [0011297]->[0011882] ... [0012384]->[0012384]
11: [0011356]->[0012384] ... [0012886]->[0012886]
10: [0012024]->[0012886] ... [0013387]->[0013388]
3: [0012874]->[0013388] ... [0013889]->[0013889]
7: [0014888]->[0014888] ... [0015390]->[0015390]
4: [0016254]->[0016254] ... [0016756]->[0016756]
5: [0017646]->[0017646] ... [0018148]->[0018148]
6: [0019088]->[0019088] ... [0019590]->[0019590]
11: [0020206]->[0020206] ... [0020708]->[0020708]
8: [0020320]->[0020708] ... [0021210]->[0021210]
10: [0021078]->[0021210] ... [0021712]->[0021712]
12: [0021384]->[0021712] ... [0022213]->[0022213]
7: [0021630]->[0022213] ... [0022715]->[0022715]
9: [0021811]->[0022715] ... [0023217]->[0023217]
3: [0022009]->[0023217] ... [0023719]->[0023719]
Динамический пул потоков
Динамический пул потоков не является каким-то специфическим механизмом, продиктованным именно микроядерной архитектурой QNX. Это удачная искусственная конструкция, все определения которой размещены в файле <sys/dispatch.h>. Удивительно не то, что в составе API QNX имеется такой механизм, а то, что подобные инструменты отсутствуют в других ОС.
В предыдущих примерах кода мы неоднократно создавали наборы потоков для тех или иных целей, но всем им было присуще одно: общее количество потоков в них было фиксированным на момент создания. Это и были статические пулы потоков, разделяющих между собой работу приложения. Архитекторы QNX идут чуть дальше: они предоставляют инструментарий для создания пулов однотипных (с общей функцией потока) потоков, в которых конкретное число потоков может увеличиваться или уменьшаться синхронно с изменением нагрузки на приложение. Именно своим динамическим составом эта конструкция и отличается.
Динамический пул потоков нужен разработчикам QNX в первую очередь как инструмент построения многопоточных менеджеров ресурсов - основы построения сервисов ОС QNX. Но и помимо этой цели динамический пул потоков представляет собой мощнейшее средство для конструирования параллельных механизмов обработки.
Проиллюстрируем применение динамического пула потоков примером программного кода, который был нами описан в книге [4] в главе «Сервер TCP/IP... много серверов хороших и разных». По сути, это ретранслирующий TCP/IP-сервер, но сейчас это для нас неважно:
Сервер на базе динамического пула потоков#include <pthread.h>
#include <sys/dispatch.h>
static int ls; // прослушивающий TCP-сокет
THREAD_POOL_PARAM_T* alloc(THREAD_POOL_HANDLE_T* h) {
return (THREAD_POOL_PARAM_T*)h;
}
// функция блокирования пула потоков
THREAD_POOL_PARAM_T* block(THREAD_POOL_PARAM_T* p) {
int rs = accept(ls, NULL, NULL);
if (rs < 0) errx("accept error");
return(THREAD_POOL_PARAM_T*)rs;
}
int handler(THREAD_POOL_PARAM_T* p) {
retrans((int)p);
close((int)p);
delay(250);
cout << pthread_self() << flush;
return 0;
}
int main(int argc, char* argv[]) {
// создать TCP-сокет на порт
ls = getsocket(THREAD_POOL_PORT);
// создание атрибутной записи пула потоков:
thread_pool_attr_t attr;
memset(&attr, 0, sizeof(thread_pool_attr_t));
// заполнение блока атрибутов пула
/* - mm число блокированных потоков в пуле */
attr.lo_water = 3;
/* - max число блокированных потоков в пуле */
attr.hi_water = 7;
/* - инкремент шага создания потоков */
attr.increment = 2;
attr.maximum = 9;
/* - общий предел числа потоков в пуле */
attr.handle = dispatch_create();
attr.context_alloc = alloc;
attr.block_func = block;
attr.handler_func = handler;
// фактическое создание пула потоков:
void* tpp = thread_pool_create(&attr, POOL_FLAG_USE_SELF);
if (tpp == NULL) errx("create pool");
// начало функционирования пула потоков:
thread_pool_start(tpp);
// ... выполнение никогда не дойдет до этой точки!
exit(EXIT_SUCCESS);
}
ПримечаниеВ примере используются, но не определены две функции, которые не столь существенны для понимания примера сточки зрения функционирования пула:
• errx() — реакция на ошибку выполнения с выводом сообщения и последующим аварийным завершением;
• retrans() — прием сообщения с присоединенного TCP-сокета с последующей ретрансляцией полученного содержимого в него же.
Итак, первая особенность пула потоков в том, что мы построили многопоточный сервер, почти не прописывая собственного кода, — большую часть рутинной работы за нас сделала библиотека пула.
Приведем описание логики работы пула потоков и показанного примера на самом качественном, простейшем уровне:
• Первоначально (при запуске пула потоков в работу вызовом thread_pool_start()) создается attr.lo_water потоков («нижняя ватерлиния» числа блокированных потоков).
• При создании любого потока (как в процессе начального, так и в процессе последующего создания) вызывается функция attr.соntext_alloc() (в контексте созданного потока).
• По завершении функция вызывает блокирующую функцию потока attr.block_func(), на которой созданный поток ожидает события активизации (в показанном примере событие активизации — это установление соединения новым клиентом по возврату из accept()).
• Блокирующая функция после наступления события активизации переведет поток в состояние READY и вызовет в контексте этого потока функцию обработчика attr.handler_func().
• Если после предыдущего шага число оставшихся заблокированных потоков станет ниже attr.lo_water, механизм пула создаст дополнительно attr.increment потоков и «доведет» их до блокирующей функции.
• Активизированный поток производит всю обработку, предписанную функцией потока, и после выполнения потоковой функции будет опять переведен в блокированное состояние в функции блокирования…
• …но перед переводом потока вновь в блокированное состояние проверяется, не будет ли при этом превышено число блокированных потоков attr.hi_water («верхняя ватерлиния»), и если это имеет место, то поток вместо перевода в блокированное состояние самоуничтожается.
• Все проверки числа потоков производятся для того, чтобы общее число потоков пула (т. e. число активизированных потоков вместе с блокированными) не превышало общее ограничение attr.maximum.
Разобрав общую логику функционирования пула потоков, можно теперь детальнее рассмотреть отдельные шаги всего процесса:
1. Прежде чем создавать пул потоков, мы должны создать атрибутную запись, определяющую все поведение пула. Атрибутная запись описана так (<sys/dispatch.h>):
typedef struct _thread_pool_attr {
THREAD_POOL_HANDLE_T* handle;
THREAD_POOL_PARAM_T*
(*block_func)(THREAD_POOL_PARAM_T* ctp);
void (*unblock_func)(THREAD_POOL_PARAM_T* ctp);
int (*handler_func)(THREAD_POOL_PARAM_T* ctp);
THREAD_POOL_PARAM_T*
(*context_alloc)(THREAD_POOL_HANDLE_T* handle);
void (*context_free)(THREAD_POOL_PARAM_T* ctp);
pthread_attr_t* attr;
unsigned short lo_water;
unsigned short increment;
unsigned short hi_water;
unsigned short maximum;
unsigned reserved[8];
} thread_pool_attr_t;
Дескриптор создаваемого пула потоков handle, посредством которого мы будем ссылаться на пул, является просто синонимом типа dispatch_t:
#ifndef THREAD_POOL_HANDLE_T
#define THREAD_POOL_HANDLE_T dispatch_t
#endif
Атрибуты потоков, которые будут работать в составе пула, определяются полем attr типа pthread_attr_t (эту структуру мы детально рассматривали ранее при обсуждении создания единичных потоков).
Численные параметры пула определяют:
lo_water — «нижняя ватерлиния», минимальное число потоков пула, находящихся в блокированном состоянии (в ожидании активизации). Если в результате некоторого события один из ожидающих потоков переходит в состояние активной обработки и число оставшихся блокированных потоков становится меньше lo_water, создается дополнительно increment потоков, которые переводятся в блокированное состояние.
hi_water — максимальное число потоков, которые допустимо иметь в блокированном состоянии. Если после завершения обработки некоторым потоком число заблокированных потоков становится больше hi_water, то этот поток уничтожается.
