Роб Кёртен - Введение в QNX/Neutrino 2. Руководство по программированию приложений реального времени в QNX Realtime Platform Страница 14

В этот момент возникает законный вопрос: «А что если потоки завершат работу в обратном порядке?» Другими словами, если имеются 4 процессора, и по какой-либо причине поток, выполняющийся на последнем процессоре (с номером 3), завершит работу первым, затем завершится поток, выполняющийся на процессоре с номером 2, и так далее? Вся прелесть приведенной схемы заключается в том, что ничего плохого не произойдет.

Первое, что произойдет — это то, что pthread_join() блокируется на thread_ids[0]. Тем временем пусть завершится поток thread_ids[3]. Это не окажет абсолютно никакого воздействия на поток main(), который будет по-прежнему ждать завершения первого потока. Затем, пусть завершит работу поток thread_ids[2]. По-прежнему, никаких последствий. И так далее — пока не завершит работу поток thread_ids[0].

В этот момент pthread_join() разблокируется, и мы немедленно переходим к следующей итерации цикла for. Вторая итерация цикла for применит pthread_join() к потоку thread_ids[1], который не будет блокирован, и итерация завершится немедленно. Почему? Потому что поток, идентифицированный как thread_ids[1], уже завершился. Поэтому наш цикл for просто «проскочит» остальные потоки и завершится. В этот момент мы будем знать, что вычислительные потоки синхронизированы, и теперь мы можем выводить результаты отображение.

Когда мы говорили о синхронизации функции main() по моменту завершения рабочих потоков (в параграфе «Синхронизация по отношению к моменту завершения потока», см. выше), мы упомянули два метода синхронизации: один метод с применением функции pthread_join(), который мы только что рассмотрели, и метод с применением барьера.

Возвращаясь к нашей аналогии с процессами в жилом доме, предположим, что семья пожелала где-нибудь отдохнуть на природе. Водитель садится в микроавтобус и запускает двигатель. И ждет. Водитель будет ждать до тех пор, пока все члены семьи не сядут в машину, и только затем можно будет ехать — не можем же мы кого-нибудь оставить!

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

Однако, отметьте для себя одну важную отличительную особенность. С применением функции pthread_join() мы ожидаем завершения потоков. Это означает, что на момент ее разблокирования потоков нет больше с нами; они закончили работу и завершились.

В случае с барьером, мы ждем «встречи» определенного числа потоков у барьера. Затем, когда заданное число потоков достигнуто, мы их всех разблокируем (заметьте, что потоки при этом продолжат выполнять свою работу).

Сначала барьер следует создать при помощи функции barrier_init():

#include <sync.h>

int barrier_init(barrier_t *barrier, const barrier_attr_t *attr, int count);

Эта функция создает объект типа «барьер» по переданному ей адресу (указатель на барьер хранится в параметре barrier) и назначает ему атрибуты, которые определены в attr (мы будем использовать NULL, чтобы установить значения по умолчанию). Число потоков, которые должны вызывать функцию barrier_wait(), передается в параметре count.

После того как барьер создан, каждый из потоков должен будет вызвать функцию barrier_wait(), чтобы сообщить, что он отработал:

#include <sync.h>

int barrier_wait(barrier_t *barrier);

После того как поток вызвал barrier_wait(), он будет блокирован до тех пор, пока число потоков, указанное первоначально в параметре count функции barrier_init(), не вызовет функцию barrier_wait() (они также будут блокированы). После того как нужное число потоков выполнит вызов функции barrier_wait(), все эти потоки будут разблокированы «одновременно».

Вот пример:

/*

* barrier1.c

*/

#include <stdio.h>

#include <time.h>

#include <sync.h>

#include <sys/neutrino.h>

barrier_t barrier; // Объект типа «барьер»

void* thread1(void *not_used) {

 time_t now;

 char buf[27];

 time(&now);

 printf("Поток 1, время старта %s", ctime_r(&now, buf));

 // Выполнить вычисления

 // (вместо этого просто сделаем sleep)

 sleep(20);

 barrier_wait(&barrier);

 // После этого момента все потоки уже завершатся

 time(&now);

 printf("Барьер в потоке 1, время срабатывания %s",

  ctime_r(&now, buf));

}

void* thread2(void *not_used) {

 time_t now;

 char buf[27];

 time(&now);

 printf("Поток 2, время старта %s", ctime_r(&now, buf));

 // Выполнить вычисления

 // (вместо этого просто сделаем sleep)

 sleep(40);

 barrier_wait(&barrier);

 // После этого момента все потоки уже завершатся

 time(&now);

 printf("Барьер в потоке 2, время срабатывания %s",

  ctime_r(&now, buf));

}

main() // Игнорировать аргументы

{

 time_t now;

 char buf[27];

 // Создать барьер со значением счетчика 3

 barrier_init(&barrier, NULL, 3);

 // Создать два потока, thread1 и thread2

 pthread_create(NULL, NULL, thread1, NULL);

 pthread_create(NULL, NULL, thread2, NULL);

 // Сейчас выполняются оба потока

 // Ждать завершения

 time(&now);

 printf("main(): ожидание у барьера, время %s",

  ctime_r(&now, buf));

 barrier_wait(&barrier);

 // После этого момента все потоки уже завершатся

 time(&now);

 printf("Барьер в main(), время срабатывания %s",

  ctime_r(&now, buf));

}

Основной поток создал объект типа «барьер» и инициализировал его значением счетчика, равным числу потоков (включая себя!), которые должны «встретиться» у барьера, прежде чем он «прорвется». В нашем примере этот индекс был равен 3 — один для потока main(), один для потока thread1() и один для потока thread2(). Затем, как и прежде, стартуют потоки вычисления графики (в нашем случае это потоки thread1() и thread2()). Для примера вместо приведения реальных алгоритмов графических вычислений мы просто временно «усыпили» потоки, указав в них sleep(20) и sleep(40), чтобы имитировать вычисления. Для осуществления синхронизации основной поток (main()) просто блокирует сам себя на барьере, зная, что барьер будет разблокирован только после того, как рабочие потоки аналогично присоединятся к нему.

Как упоминалось ранее, с функцией pthread_join() рабочие потоки для синхронизации главного потока с ними должны умереть. В случае же с барьером потоки живут и чувствуют себя вполне хорошо. Фактически, отработав, они просто разблокируются по функции barrier_wait(). Тонкость здесь в том, что вы обязаны предусмотреть, что эти потоки должны делать дальше! В нашем примере с графикой мы не дали им никакого задания для них — просто потому что мы так придумали алгоритм. В реальной жизни вы могли бы захотеть, например, продолжить вычисления.

Предположим, что мы слегка изменили наш пример так, чтобы можно было проиллюстрировать, почему иногда хорошо иметь несколько потоков даже в системе с одиночным процессором.

В таком модифицированном примере один узел на сети ответственен за вычисление строк растра (как и в примере с графикой, рассмотренном выше). Однако, когда строка рассчитана, ее данные должны быть отправлены по сети другому узлу, который выполняет функцию отображения. Ниже приведена соответствующая модифицированная функция main() (на основе первоначального примера без потоков):

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

 int x1;

 ... // выполнить инициализации

 for (x1 = 0; x1 < num_x_lines; x1++) {

  do _one_line(x1); // Область «С» на схеме

  tx_one_line_wait_ack(x1); // Области «X» и «W» на схеме

 }

}

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