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

• главный поток (TID = 1) создает 3 новых потока (TID = 2, 3, 4);

• главный поток переходит в пассивное ожидание сигналов, но в его маске доставка посылаемого сигнала (41) заблокирована;

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

• … 3 потока (TID = 2, 3, 4) ожидают поступления сигнала;

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

Таким образом, сигнал доставляется одному и только одному потоку, который не блокирует этот сигнал. Обработчик сигнала вызывается в контексте (стек, области собственных данных) этого потока. После выполнения обработчика сигнал поглощается. Какому из потоков, находящихся в состоянии блокирования в ожидании сигналов (в масках которых разблокирован данный сигнал), будет доставлен экземпляр сигнала, предсказать невозможно; это так и должно быть исходя из общих принципов диспетчеризации потоков. Но реально этим потоком является поток, последнимперешедший в состояние ожидания. Для того чтобы убедиться в этом, заменим предпоследнюю строку программы ( pause();) на:

threadfunc(NULL);

Теперь у нас 4 равнозначных потока, ожидающих прихода сигнала, переходящих в состояние ожидания в последовательности: TID = 2, 3, 4, 1. Реакция процесса на приход сигнала изменится на:

SIG = 41, TID = 1

Изменим текст функции потока на ( файл s7.cc ):

void* threadfunc(void* data) {

while (true) {

SignalProcmask(0, 0, SIG_UNBLOCK, &sig, NULL);

delay(1);

SignalProcmask(0, 0, SIG_BLOCK, &sig, NULL);

delay(10);

}

}

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

SIG = 41, TID = 1

SIG = 41; TID = 4

SIG = 41; TID = 4

SIG = 41; TID = 1

SIG = 41; TID = 1

SIG = 41; TID = 4

SIG = 41; TID = 4

SIG = 41; TID = 1

SIG = 41; TID = 2

SIG = 41; TID = 2

SIG = 41; TID = 3

SIG = 41; TID = 4

Такая модель вряд ли может быть названа в полной мере «сигналами в потоках», так как сигнал в ней в конечном итоге направляется процессу как контейнеру, содержащему потоки (можно сказать и так: в оболочку адресного пространства процесса). И только после этого в контексте одного из потоков(и в случае множественных потоков, разблокированных на обработку единого сигнала, невозможно предсказать, в контексте какого из них) выполняется обработчик сигнала. Главный поток процесса (TID = 1) в этой схеме участвует в равнозначном качестве (здесь хорошо видно, что устоявшееся понятие «реакция процесса на сигнал» в строгом смысле некорректно).

Перейдем к более конкретным вопросам: как можно продуктивно использовать эту схему в многопоточных приложениях? Рассмотрим сначала случай, когда каждый из рабочих потоков разблокирован на получение одного, свойственного только ему сигнала ( файл s9.cc ):

#include

#include

#include

#include

#include

#include

#include

static void handler(int signo, siginfo_t* info, void* context) {

cout << "SIG = " << signo << "; TID = " << pthread_self() << endl;

}

void* threadfunc(void* data) {

// блокировать реакцию на все сигналы

sigset_t sig;

sigfillset(&sig);

SignalProcmask(0, 0, SIG_BLOCK, &sig, NULL);

// разблокировать реакцию на свой сигнал

sigemptyset(&sig);

sigaddset(&sig, (int)data);

SignalProcmask(0, 0, SIG_UNBLOCK, &sig, NULL);

// цикл ожидания приходящих сигналов

while (true) pause();

}

int main() {

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

// единую функцию реакции, иначе все было бы гораздо проще.

struct sigaction act;

sigemptyset(&act.sa_mask);

act.sa_sigaction = handler;

act.sa_flags = SA_SIGINFO;

// создаем группу однотипных потоков

const int thrnum = 3;

for (int i = SIGRTMIN; i - SIGRTMIN < thrnum; i++) {

sigset_t sig;

sigemptyset(&sig);

sigaddset(&sig, 1);

// нам нужно, чтобы главный поток не реагировал:

sigprocmask(SIG_BLOCK, &sig, NULL);

if (sigaction(i, &act, NULL) < 0) perror("set signal handler: ");

// для передачи номера сигнала используется

// трюк с подменой типа параметра:

pthread_create(NULL, NULL, threadfunc, (void*)(i));

}

// начинаем циклическую синхронизацию потоков.

for (int i = 0; ; i++) {

sleep(1);

// посылку сигнала можно (так даже будет корректнее)

// сделать так:

// union sigval val;

// val.sival_int = i;

// sigqueue(getpid(), SIGRTMIN + i % thrnum, val);

// но мы сознательно демонстрируем и приемлемость kill:

kill(getpid(), SIGRTMIN + i % thrnum);

}

}

В этой программе главный поток циклически по таймеру активизирует поочередно каждый поток. Вот фрагмент вывода работающей программы:

SIG = 41; TID = 2

SIG = 42; TID = 3

SIG = 43; TID = 4

SIG = 41; TID = 2

SIG = 42; TID = 3

SIG = 43; TID = 4

SIG = 41; TID = 2

SIG = 42; TID = 3

SIG = 43; TID = 4

Часто приходится слышать: «…хотелось бы доставить сигнал всем потокам, уведомить всех потребителей и выполнить функцию реакции в каждом потоке…», и именно в такой последовательности действий понимается модель сигналов в потоках при поверхностном с ней ознакомлении. Иногда это представляется очень интересной возможностью, и мы реализуем такую схему взаимодействия в следующем фрагменте ( файл s10.cc ):

#include

#include

#include

#include

#include

#include

#include

static void handler(int signo, siginfo_t* info, void* context) {

cout << "SIG = " << signo << ", TID = " << pthread_self() << endl;

}

static void endhandler(int signo) {}

// сигнал, на который реагируют потоки:

const int SIGNUM = SIGRTMIN;

sigset_t sig;

struct threcord {

int tid;

bool noblock;

};

static vector tharray; // вектор состояний потоков

void* threadfunc(void* data) {

// блокирование всех прочих сигналов:

sigset_t sigall;

sigfillset(&sigall);

SignalProcmask(0, 0, SIG_BLOCK, &sigall, NULL);

// передеспетчеризация для завершения формирования вектора

sched_yield();

tharray[(int)data].noblock =

(SignalProcmask(0, 0, SIG_UNBLOCK, &sig, NULL) != -1);

while (true) {

pause();

tharray[(int)data].noblock =

!(SignalProcmask(0, 0, SIG_BLOCK, &sig, NULL) != 1);

bool nolast = false;

for (vector::iterator i = tharray.begin();

i != tharray.end(); i++)