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

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

unsigned int T = 2;

static sem_t* sem;

static bool debug = false;

static char* str; // строка диагностики

static volatile int ind = 0;

uint64_t *t;

void* threadfunc(void* data) {

unsigned long i = 0;

char tid[8];

sprintf(tid, "%X", pthread_self);

//

временная метка начала во всех потоках устанавливается

// на время достижения этой точки в последнем (активном) потоке

if ((int)data == T - 1) {

uint64_t с = ClockCycles;

for (int i = 0; i < T; i++ ) t[i] = c;

}

// рабочий цикл переключений за счет синхронизации

while (i++ < N) {

sem_wait(sem + (int)data);

if (debug) str[ind++] = *tid;

sem_post(sem + ((int)data +1) % T);

}

t[(int)data] = ClockCycles - t[(int)data];

return NULL;

}

int main(int argc, char *argv[]) {

int opt, val;

while ((opt = getopt(argc, argv, "n:t:v")) != -1) {

switch(opt) {

case 'n':

if (sscanf(optarg, "%i", &val) != 1)

cout << "parse command line error" << endl, exit(EXIT_FAILURE);

if (val > 0) N — val;

break;

case 't':

if (sscanf(optarg, "%i", &val) != 1)

cout << "parse command line error" << endl, exit(EXIT_FAILURE);

if (val > 0) T = val;

break;

case 'v':

debug = true;

break;

default:

exit(EXIT_FAILURE);

}

}

if (debug) str = new char[T * N + 1];

pthread_t* tid = new pthread_t[T];

sem = new sem_t[T];

t = new uint64_t[T];

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

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

// на своих семафорах сразу же после старта

if (sem_init(sem + i, 0, (i == (T - 1)) ? 1 : 0))

perror("semaphore init"), exit(EXIT_FAILURE);

if (pthread_create(tid + i, NULL, threadfunc, (void*)i

! = EOK)

perror( "thread create error"), exit(EXIT_FAILURE);

}

for (int i=0; i < T; i++)

pthread_join(tid[i], NULL);

for (int i = 0; i < T; i++) sem_destroy(sem + i);

delete [] sem;

for (int i = 0; i < T; i++)

cout << tid[i] << "\t: cycles - " << t[i] << ";\ton semaphore - " <<

t[i] / T / N << endl;

delete [] tid;

delete [] t;

if (debug) {

str[ind] = "\0"; cout << str << endl;

delete [] str;

}

exit(EXIT_SUCCESS);

}

Логически приложение изменилось следующим образом:

• Теперь у нас может быть не 2 идентичных (симметричных) потока, а произвольное их количество (ключ

– t

при запуске приложения).

• Потоки синхронизируются не на одном семафоре — введен массив семафоров по числу потоков: каждый поток блокируется на «своем» семафоре, но разблокирует его (после очередного выполнения своего фрагмента) семафор заблокированного «соседа».

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

• Из кода исключены какие бы то ни было средства принудительной передачи управления (

sched_yield

) — все управление логикой ветвления осуществляется только состояниями семафоров.

Посмотрим, что у нас получилось. Запускаем приложение с диагностическим выводом идентификаторов потоков (ключ

– v

; он у нас был в тестах и ранее, только мы о нем не упоминали):

# nice -n-19 sy21 -n20 -t12 -v

2 : cycles - 664874; on semaphore - 2770