QNX/UNIX: Анатомия параллелизма - Олег Цилюрик

hi_water — максимальное число потоков, которые допустимо иметь в блокированном состоянии. Если после завершения обработки некоторым потоком число заблокированных потоков становится больше hi_water, то этот поток уничтожается.

maximum — общая верхняя граница числа потоков пула (активизированных и заблокированных). Даже если число заблокированных потоков (в пике активности) станет ниже lo_water, но общее число потоков уже достигнет maximum, то новые потоки для пула создаваться не будут.

Функциональные параметры пула определяют:

context_alloc() и context_free() — функции создания и уничтожения контекста потока, которые вызываются при создании и уничтожении каждого потока пула. Функция создания контекста потока ответственна за индивидуальные настройки создаваемого потока. Она возвращает «указатель на контекст» типа THREAD_POOL_PARAM_T. Однако системе такой тип неизвестен:

#ifndef THREAD_POOL_PARAM_T

 #define THREAD_POOL_PARAM_T void

#endif

В качестве контекста может использоваться любой пользовательский тип, и он будет передаваться последовательно в качестве параметра (ctp) во все последующие функции обслуживания потока.

block_func() — функция блокирования, которая вызывается в потоке сразу же после context_alloc() или после очередного этапа выполнения потоком функции обработчика handler_func(). Функция блокирования получает и возвращает далее обработчику (возможно, после модификации) структуру контекста (в приведенном выше примере контекстом является int — значение присоединенного TCP-сокета).

handler_func() — это, собственно, и есть аналог потоковой функции, в которой выполняется вся полезная работа потока. Функция вызывается библиотекой после выхода потока из блокирующей функции block_func(), при этом функция-обработчик handler_func() получит параметр контекста, возвращенный block_func().

В текущей реализации handler_func() должна возвращать 0; все другие значения зарезервированы для дальнейших расширений. Аналогично определенная в атрибутной записи функция unblock_func() зарезервирована для дальнейших расширений, и вместо ее адреса следует устанавливать NULL.

2. После создания атрибутной записи пула, определяющей всю функциональность его дальнейшего поведения, можно приступать к непосредственному созданию пула потоков:

thread_pool_t* thread_pool_create(

 thread_pool_attr_t* attr, unsigned flags);

где attr — подробно рассмотренная (и созданная) ранее атрибутная запись пула;

flags — флаг, определяющий поведение вызывающего потока после последующего вызова thread_pool_start(). В документации описано два возможных значения флага:

 • POOL_FLAG_EXIT_SELF — после старта пула поток, вызвавший thread_pool_start() (часто это главный поток приложения), завершается;

 • POOL_FLAG_USE_SELF — после старта пула поток, вызвавший thread_pool_start(), включается в пул в качестве одного из его потоков.

И в том и в другом случае в типовом фрагменте (как и в показанном выше примере):

thread_pool_start(tpp);

exit(EXIT_SUCCESS);

управление никогда не дойдет до выполнения exit(). Но существует еще третье допустимое значение, прямо не указанное в документации, но мельком упоминаемое в других местах документации:

 • 0 — после старта пула поток, вызвавший thread_pool_start(), продолжает свое естественное выполнение.

Например, некоторый фрагмент кода мог бы выглядеть так:

thread_pool_attr_t att; // ...

thread_pool_t *tpp = thread_pool_create(&attr, 0);

thread_pool_start(tpp);

while (true) {

 // выполнять некоторую отличную от пула работу

}

exit(EXIT_SUCCESS);

Как уже понятно из описаний, thread_pool_create() возвращает указатель на управляющую структуру пула потоков, которая позже будет передана thread_pool_start(). Если создание пула завершилось неудачей, то результатом выполнения будет NULL, а в errno будет установлен код ошибки (документацией предусмотрен только один код ошибки: ENOMEM — недостаточно памяти для размещения структур данных).

Управляющая структура пула потоков описана так:

typedef struct _thread_pool thread_pool_t;

struct _thread_pool {

 thread_pool_attr_t pool_attr;

 unsigned created;

 unsigned waiting;

 unsigned flags;

 unsigned reserved[3];

};

3. Последний шаг в процедуре запуска пула потоков:

int thread_pool_start(void* pool);

где pool — это указатель, возвращаемый thread_pool_create().[40]

При успешном завершении (которого почти никогда не происходит, за исключением значения флага 0; об этом см. выше) функция возвращает EOK, в противном случае (что происходит гораздо чаще) — значение -1.

4. Другие, относящиеся к библиотеке динамического пула потоков функции, которые целесообразно посмотреть в документации QNX (но которые в силу различных обстоятельств используются гораздо реже):

int thread_pool_destroy(thread_pool_t* pool);

int thread_pool_control(thread_pool_t* pool, thread_pool_attr_t* attr,

 _Uint16t lower, _Uint16t upper, unsigned flags);

int thread_pool_limits(thread_pool_t* pool,

 int lowater, int hiwater, int maximum, int increment, unsigned flags);

Менеджеры ресурсов

QNX вводит технику программирования, которая единообразно проходит сквозь всю систему.[41] Идея техники менеджеров ресурсов столь же проста, сколь и остроумна:

• Вся система построена на тщательно проработанной в теории (и редко используемой при построении реальных ОС) концепции - коммутации сообщений. Ядро (точнее «микроядро») операционной системы при таком подходе выступает в качестве компактного коммутатора сообщений между взаимодействующими программными компонентами. При этом взаимодействующие компоненты выступают в качестве клиента, запрашивающего услугу (ресурс), и сервера, обеспечивающего эту услугу (обслуживающего ресурс).

• Большинство системных вызовов API (в том числе все «привычные» POSIX-вызовы: open(), read(), write(), seek(), close()…) реально посылаются обслуживающему данный ресурс сервису (например, в файловую систему типа FAT32 — fs-dos) в виде сообщений уровня микроядра. Код сообщения при этом определяет тип операции (например, open()), а последующее тело сообщения — конкретные параметры запроса, зависящие от типа операции (параметры запроса пакуются в тело сообщения).

• Раз эта схема столь универсальна, единообразна и не зависит от конкретной природы ресурса, на котором обеспечивается обслуживание, то разработчики QNX предоставляют некоторый шаблон сервера, в котором на месте обработчиков стандартных POSIX-запросов находятся пустые программные заглушки. Этот шаблон и служит базовым элементом построения разнообразных серверов услуг, называемых при выполнении в такой технике «менеджерами ресурса».

• При запуске программа менеджера ресурса регистрирует свое имя (точнее имя управляемого ею ресурса) в пространстве имен файловой системы QNX (обычно в каталоге /dev, но это может быть любое место файловой системы). Теперь можно обращаться с запросами к данному менеджеру так же, как и к любому реальному файлу в файловой системе.

• Программисту, пишущему свой драйвер услуги, ресурса, устройства или псевдоустройства, остается только переопределить программное наполнение тех программных заглушек, которые ответственны за интересующие его вызовы (например, open(), read(), close()), никак не затрагивая вызовы, не обеспечиваемые этим ресурсом (например, write(), seek() и др.).

В наши цели не входит детальное обсуждение техники написания менеджеров ресурсов (этому посвящено специальное исчерпывающее руководство в составе технической документации QNX объемом более 80 страниц[42]). Поэтому, как и ранее с динамическим пулом потоков, начнем с примера. Приведем простейший код менеджера ресурса, который использовался нами для тестирования наследования приоритетов в QNX (файл prior.cc):

#include <errno.h>

#include <stdio.h>

#include <stddef.h>

#include <stdlib.h>

#include <unistd.h>

#include <string.h>

#include <pthread.h>

#include <sys/iofunc.h>

#include <sys/dispatch.h>

// обработчик запроса от клиента read(),

// возвращающий текущий приоритет обслуживания

static int prior_read(resmgr_context_t *ctp, io_read_t *msg,

 RESMGR_OCB_T *ocb) {

 static bool odd = true;

 int status = iofunc_read_verify(ctp, msg, ocb, NULL);

 if (status != EOK) return status;

 if (msg->i.xtype & _IO_XTYPE_MASK != _ID_XTYPE_NONE)

  return ENOSYS;

 if (odd) {