QNX/UNIX: Анатомия параллелизма

Цилюрик Олег Иванович

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 <stdio.h>

#include <iostream.h>

#include <signal.h>

#include <unistd.h>

#include <pthread.h>

#include <sys/neutrino.h>

#include <vector>

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<threcord> 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<threcord>::iterator i = tharray.begin;

i != tharray.end; i++)

if (nolast = i->noblock) break;

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

if (nolast) kill(getpid, SIGNUM);

// ... когда пересылать больше некому -

// переинициализация масок

else

for (vector<threcord>::iterator i = tharray.begin;

i != tharray.end; i++)

i->noblock = (SignalProcmask(0, i->tid, SIG_UNBLOCK, &sig, NULL) != -1);

}

}

int main {

// переопределение реакции ^C в старой манере

signal(SIGINT, endhandler);

// маска блокирования-разблокирования

sigemptyset(&sig);

sigaddset(&sig, SIGNUM);

// блокировка в главном потоке приложения

sigprocmask(SIG_BLOCK, &sig, NULL);