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

Текст добавлен: 8 октября 2016, 23:59

Текст книги "QNX/UNIX: Анатомия параллелизма"

Автор книги: Олег Цилюрик

Соавторы: Владимир Зайцев,Егор Горошко

Жанры:

ОС и Сети

сообщить о нарушении

Текущая страница: 6 (всего у книги 23 страниц) [доступный отрывок для чтения: 9 страниц]

Назад к карточке книги

Безопасность вызовов в потоковой среде

Рассмотрев «в первом приближении» технику собственных данных потоков, мы теперь готовы ответить на вопрос: «В чем же главное предназначение такой в общем-то достаточно громоздкой техники? И зачем для ее введения потребовалось специально расширять стандарты POSIX?» Самое прямое ее предназначение, помимо других «попутных» применений, которые были обсуждены ранее, – это общий механизм превращения существующей функции для однопотокового исполнения в функцию, безопасную (thread safe) в многопоточном окружении. Этот механизм предлагает единую (в смысле «единообразную», а не «единственно возможную») технологию для разработчиков библиотечных модулей.

Примечание
ОС QNX, заимствующая инструментарий GNU-технологии (gcc, make, …), предусматривает возможность построения как статически связываемых библиотек (имена файлов вида xxx.a), так и разделяемых или динамически связываемых (имена файлов вида xxx.so). Целесообразность последних при построении автономных и встраиваемых систем (на что главным образом и нацелена ОС QNX) достаточно сомнительна. Однако высказанное выше положение о построении реентерабельных программных единиц относится не только к библиотечным модулям (как статическим, так и динамическим) в традиционном понимании термина «библиотека», но и охватывает куда более широкий спектр возможных объектов и в той же мере относится и просто к любым наборам утилитных объектных модулей (вида xxx.о), разрабатываемых в ходе реализации под целевой программный проект.

Если мы обратимся к технической документации API QNX (аналогичная картина будет и в API любого UNIX), то заметим, что только небольшая часть функций отмечена как thread safe. К «небезопасным» отнесены такие общеизвестные вызовы, как select(), rand()и readln(), а многим «небезопасным» в потоковой среде вызовам сопутствуют их безопасные дубликаты с суффиксом *_rв написании имени функции, например MsgSend() – MsgSend_r().

В чем же состоит небезопасность в потоковой среде? В нереентерабельности функций, подготовленных для выполнения в однопоточной среде, в первую очередь связанной с потребностью в статических данных, хранящих значение от одного вызова к другому. Рассмотрим классическую функцию rand(), традиционно реализуемую в самых разнообразных ОС примерно так (при «удачном» выборе констант А, В, С):

int rand(void) {

static int x = rand_init();

return x = (A*x + B)%C;

}

Такая реализация, совершенно корректная в последовательной (однопотоковой) модели, становится небезопасной в многопоточной: а) вычисление xможет быть прервано событием диспетчеризации, и не исключено, что вновь получивший управление поток в свою очередь обратится к rand()и исказит ход текущего вычисления; б) каждый поток «хотел бы» иметь свою автономную последовательность вычислений x, не зависящую от поведения параллельных потоков. Желаемый результат будет достигнут, если каждый поток будет иметь свой автономный экземпляр переменной x, что может быть получено двумя путями:

1. Изменить прототип объявления функции:

int rand_r(int *x) {

return x = (А * (*x) + В) % С;

};

При этом проблема «клонирования» переменной x в каждом из потоков (да и начальной ее инициализации) не снимается, она только переносится на плечи пользователя, что, однако, достаточно просто решается при создании потоковой функции за счет ее стека локальных переменных:

void* thrfunc(void*) {

int x = rand_init();

... = rand_r(&x);

};

Именно такова форма и многопоточного эквивалента в API QNX – rand_r().

2. В этом варианте мы сохраняем прототип описания функции без изменений за счет использования различных экземпляров собственных данных потока. (Весь приведенный ниже код размещен в отдельной единице компиляции; все имена, за исключением rand(), невидимы и недоступны из точки вызова, что подчеркнуто явным использованием квалификатора static.)

static pthread_key_t key;

static pthread_once_t once = PTHREAD_ONCE_INIT;

static void destr(void* db) { delete x; }

static void once_creator(void) { pthread_key_create(&key, destr); }

int rand(void) {

pthread_once(&once, once_creator);

int *x = pthread_getspecific(key);

if (x == NULL) {

pthread_setspecific(key, x = new int);

*x = rand_init();

}

return x = (A * (*x) + B) % C;

}

В этом варианте, в отличие от предыдущего, весь код вызывающего фрагмента при переходе к многопоточной реализации остается текстуально неизменным:

void* thrfunc(void*) {

// ...

while (true) {

... = rand(x);

}

Перевод всего программного проекта на использование потоковой среды состоит в замене объектной единицы (объектного файла, библиотеки), содержащей реализацию rand(), и новой сборке приложения с этой объектной единицей.

При таком способе изменяются под потоковую безопасность и стандартные общеизвестные библиотечные функции API, написанные в своем первозданном виде 25 лет назад… (по крайней мере, так предлагает это делать стандарт POSIX, вводящий в обиход собственные данные потоков).

Диспетчеризация потоков

Каждому потоку, участвующему в процессе диспетчеризации, соответствует экземпляр структуры, определенной в файле , в котором находятся все фундаментальные для ОС QNX определения:

struct sched_param {

_INT32 sched_priority;

_INT32 sched_curpriority;

union {

_INT32 reserved[8];

struct {

_INT32 __ss_low_priority;

_INT32 __ss_max_repl;

struct timespec __ss_repl_period;

struct timespec __ss_init_budget;

} __ss;

} __ss_un;

};

#define sched_ss_low_priority __ss_un.__ss.__ss_low_priority

#define sched_ss_max_repl __ss_un.__ss.__ss_max_repl

#define sched_ss_repl_period __ss_un.__ss.__ss_repl_period

#define sched_ss_init_budget __ss_un.__ss.__ss_init_budget

Все, что определяется внутри union __ss_un, имеет отношение только к спорадической диспетчеризации (спорадическая диспетчеризация была введена значительно позже других, и ей будет уделено достаточно много внимания). Для всех остальных типов диспетчеризации потока это поле заполняется фиктивным полем reserved, и именно так в укороченном виде) определялась структура диспетчеризации в версии QNX 6.1.

Сейчас нас интересуют начальные поля этой структуры, не зависящие от типа диспетчеризации потока:

sched_priority– статический приоритет, который присваивается потоку при его создании и который может быть программно изменен по ходу выполнения потока;

sched_curpriority– текущий приоритет, с которым выполняется (и согласно которому диспетчеризируется) данный поток в текущий момент времени. Это значение приоритета образуется системой на основе заданного статического приоритета, но оно может динамически изменяться системой, например при отработке дисциплин наследования приоритетов или граничных приоритетов для потока. Программа не имеет средств воздействия на это значение [20]20
Пользователь может изменять это поле, однако это лишено смысла и не влечет за собой никаких последствий, ведь значением текущего приоритета «управляет» ОС, например для осуществления наследования приоритетов. С другой стороны, иногда целесообразно считать значение именно этого поля, чтобы определить значение динамического приоритета потока, установленного, например, в результате того же наследования.

[Закрыть], но может его считывать.

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

pid tid name prio STATE Blocked

1 1 6/boot/sys/procnto 0f READY

1 2 6/boot/sys/procnto 10r RUNNING

...

1 5 6/boot/sys/procnto 63r RECEIVE 1

...

1 9 6/boot/sys/procnto 6r NANOSLEEP

...

6 1 roc/boot/devb-eide 10o SIGWAITINFO

В поле prioуказывается приоритет (текущий; возможно, последнее из унаследованных значений!) каждого потока с установленной для него дисциплиной диспетчеризации: f– FIFO, r– RR, o – OTHER, s– SPORADIC.

В системе на сегодняшний день реализованы три [21]21
В документации неоднократно упоминается еще одна дисциплина – «адаптивная» ( SCHED_ADAPTIVE), и даже детально описывается, «как она работает». Видимо, это можно отнести только к тому, что корректировка обширной документации отстает от развития самой системы. На конференции «QNX-Россия 2003» на вопрос по поводу ADAPTIVE-диспетчеризации представители QSSL отвечали так: «Этот вид диспетчеризации был в QNX 4.xx, а в QNX 6.x вместо него введена более продвинутая техника SPORADIC-диспетчеризации». Тем не менее более продвинутая спорадическая диспетчеризация не позволяет абсолютно точно выразить логику адаптивной.

[Закрыть]дисциплины диспетчеризации: очередь потоков равных приоритетов (FIFO – first in first out; еще в ходу термин «невытесняющая»), карусельная (RR – round-robin) и спорадическая. Рассмотрим фрагмент их определения в файле :

#if defined(__EXT_QNX)

#define SCHED_NOCHANGE 0

#endif

#define SCHED_FIFO 1

#define SCHED_RR 2

#define SCHED_OTHER 3

#if defined(__EXT_QNX)

#define SCHED_SPORADIC 4 /* Approved 1003.1d D14 */

#define SCHED_ADJTOHEAD 5 /* Move to head of ready queue */

#define SCHED_ADJTOTAIL 6 /* Move to tail of ready queue */

#define SCHED_MAXPOLICY 6 /* Maximum valid policy entry */

#endif

Все дисциплины диспетчеризации, кроме спорадической, достаточно полно описаны в литературе [1], поэтому мы лишь перечислим их отличительные особенности:

1. FIFO – это та дисциплина диспетчеризации, которая в литературе по Windows 3.1/3.11 называлась «невытесняющей многозадачностью» (или «кооперативной»). Здесь выполнение потока не прерывается потоками равного приоритета до тех пор, пока сам поток «добровольно» не передаст управление, например вызовом sched_yield()(часто для этой цели используется косвенный эффект вызовов delay(), sleep()и им подобных). В других источниках такой способ диспетчеризации называют очередями потоков равных приоритетов.

2. RR – это та дисциплина диспетчеризации, которая в Windows 98/NT/XP именуется «вытесняющей многозадачностью»; еще в литературе для нее используется термин «режим квантования времени».

Поток работает непрерывно только в течение предопределенного кванта времени. (В нескольких местах документации утверждается, что значение этого кванта времени составляет 4 системных тика (time-slice), что в QNX 6.2.1 по умолчанию составляет 4 миллисекунды, и только в одном месте документации говорится, что квант диспетчеризации составляет 50 миллисекунд; это определенное разночтение. Справедливым является именно первое утверждение.)

После истечения отведенного ему кванта времени поток вытесняется потоком равного приоритета (при отсутствии других потоков этим новым потоком может быть и только что вытесненный, то есть его выполнение будет продолжено, но передиспетчеризация тем не менее происходит). Установленный квант времени диспетчеризации может быть получен вызовом (стандарт POSIX 1003.1):

#include

int sched_rr_get_interval(pid_t pid, struct timespec* interval);

где pid– это PID процесса, для которого определяется квант времени, как и для многих других подобных функций. Если PID = 0, вызов относится к текущему процессу;

interval– указатель на структуру timespec(стандарт POSIX 1003.1):

#include

struct timespec {

time_t tv_sec; // значение секунд

long tv_nsec; // значение наносекунд

}

При успешном выполнении функция sched_rr_get_interval()возвращает 0, в противном случае -1.

Примечание
Две другие функции, часто удобные для работы со структурой timespec:

#include

void nsec2timespec(struct timespec *timespec_p, _uint64 nsec);

– это преобразование интервала, выраженного в наносекундах (nsec), в структуру timespec(«выходной» параметр вызова timespec_p);

#include

_uint64 timespec2nsec(const struct timespec* ts);

– это преобразование структуры timespec в значение, выраженное в наносекундах (это функция из native API QNX).

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

Часто задают вопрос: «А как много потоков целесообразно делать? Не сколько снижается эффективность многопоточной программы за счет диспетчеризации потоков?» С другой стороны, в литературе часто встречаются (достаточно голословные, на качественном уровне) утверждения, что многопоточная программа будет заметно уступать в фиктивности своему последовательному (в одном потоке) эквиваленту. Проверим это на реальной задаче:

Множественные потоки в едином приложении

#include

// преобразование процессорных циклов в миллисекунды:

static double cycle2milisec(uint64_t ccl) {

const static double s2m = 1.E+3;

// это скорость процессора

const static uint64_t

cps = SYSPAGE_ENTRY(qtime)->cycles_per_sec;

return (double)ccl * s2m / (double)cps;

}

static int nsingl = 1;

// рабочая функция, которая имитирует вычисления:

void workproc(int how) {

const int msingl = 30000;

for (int j = 0; j < how; j++)

for (uint64_t i=0; i < msingl * nsingl; i++)

i = (i + 1) – 1;

}

static pthread_barrier_t bstart, bfinish;

struct interv { uint64_t s, f; };

interv *trtime;

void* threadfunc(void* data) {

// все потоки после создания должны «застрять» на входном

// барьере, чтобы потом одновременно «сорваться» в исполнение

pthread_barrier_wait(&bstart);

int id = pthread_self() – 2;

trtime[id].s = ClockCycles();

workproc((int)data);

trtime[id].f = ClockCycles();

pthread_barrier_wait(&bfinish);

return NULL;

}

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

// здесь только обработка многочисленных ключей...

int opt, val, nthr = 1, nall = SHRT_MAX;

while ((opt = getopt(argc, argv, «t:n:p:a:»)) != -1) {

switch(opt) {

case 't':

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

perror(«parse command line failed»), exit(EXIT_FAILURE);

if (val > 0 && val <= SHRT_MAX) nthr = val;

break;

case 'p':

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

perror(«parse command line failed»), exit(EXIT_FAILURE);

if (val != getprio(0))

if (setprio(0, val) == -1)

perror(«priority isn't a valid»), exit(EXIT_FAILURE);

break;

case 'n':

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

perror(«parse command line failed»), exit(EXIT_FAILURE);

if (val > 0) nsingl *= val;

break;

case 'a':

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

perror(«parse command line failed»), exit(EXIT_FAILURE);

if (val > 0) nall = val;

break;

default:

exit(EXIT_FAILURE);

}

// ... вот здесь начинается собственно сама программа.

if (nthr > 1)

cout << "Multi-thread evaluation, thread number = " << nthr;

else cout << «Single-thread evaluation»;

cout << " , priority level: " << getprio(0) << endl;

__clockperiod clcout;

ClockPeriod(CLOCK_REALTIME, NULL, &clcout, 0);

// интервал диспетчеризации – 4 периода tickslice

// (системного тика):

cout << «rescheduling = t»

<< clcout.nsec * 4 / 1000000. << endl;

// калибровка времени выполнения в одном потоке

const int NCALIBR = 512;

uint64_t tmin = 0, tmax = 0;

tmin = ClockCycles();

workproc(NCALIBR);

tmax = ClockCycles();

cout << «calculating = t»

<< cycle2milisec(tmax – tmin) / NCALIBR << endl;

// а теперь контроль времени многих потоков

if (pthread_barrier_init(&bstart, NULL, nthr) != EOK)

perror(«barrier init»), exit(EXIT_FAILURE);

if (pthread_barrier_init(&bfinish, NULL, nthr + 1) != EOK)

perror(«barrier init»), exit(EXIT_FAILURE);

trtime = new interv[nthr];

int cur = 0, prev = 0;

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

// границы участков работы для каждого потока.

cur = (int)floor((double)nall / (double)nthr * (i + 1) + .5);

prev = (int)floor((double)nall / (double)nthr * i + 5);

if (pthread_create(NULL, NULL, threadfunc, (void*)(cur – prev)) != EOK)

perror(«thread create»), exit(EXIT_FAILURE);

}

pthread_barrier_wait(&bfinish);

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

tmin = (i == 0) ? trtime[0].s : __min(tmin, trtime[i].s);

tmax = ( i == 0 ) ? trtime[0].f : __max(tmax, trtime[i].f);

}

cout << «evaluation = t»

<< cycle2milisec(tmax – tmin) / nall << endl;

pthread_barrier_destroy(&bstart);

pthread_barrier_destroy(&bfinish);

delete trtime;

exit(EXIT_SUCCESS);

}

Логика этого приложения крайне проста:

• Есть некоторая продолжительная по времени рабочая функция ( workproc), выполняющая массированные вычисления.

• Многократно (это число определяется ключом запуска а )выполняется рабочая функция. Хорошо (то есть корректнее), если время ее единичного выполнения, которое задается ключом n, больше интервала диспетчеризации системы (в системе установлена диспетчеризация по умолчанию – круговая, или карусельная).

• Весь объем этой работы делится поровну (или почти поровну) между несколькими (ключ t) потоками.

• Сравниваем усредненное время единичного выполнения рабочей функции для разного числа выполняющих потоков (в выводе «calculating»– это время эталонного вычисления в одном главном потоке, a «evaluation»– время того же вычисления, но во многих потоках).

• Для того чтобы иметь еще большую гибкость, предоставляется возможность переопределять приоритет, под которым в системе все это происходит (ключ p).

Вот самая краткая сводка результатов (1-я строка вывода переносится для удобства чтения):

# t1 -n1 -t1000 -a2000

Multi-thread evaluation, thread number = 1000, priority level: 10

rescheduling = 3.99939

calculating = 1.04144

evaluation = 1.08001

# t1 -n1 -t10000 -a20000

Multi-thread evaluation, thread number = 10000, priority level: 10

rescheduling = 3.99939

calculating = 1.04378

evaluation = 1.61946

# t1 -n5 -a2000 -t1

Single-thread evaluation, priority level: 10

rescheduling = 3.99939

calculating = 5.07326

evaluation = 5.04726

# t1 -n5 -a2000 -t2

Multi-thread evaluation, thread number = 2, priority level: 10

rescheduling = 3.99939

calculating = 5.06309

evaluation = 5.04649

# t1 -n5 -a2000 -t20

Multi-thread evaluation, thread number = 20, priority level: 10

rescheduling = 3.99939

calculating = 5.06343

evaluation = 4.96956

# t1 -n5 -p51 -a512 -t1

Single-thread evaluation, priority level: 51

rescheduling = 3.99939

calculating = 4.94502

evaluation = 4.94511

# t1 -n5 -р51 -a512 -t11

Multi-thread evaluation, thread number = 11, priority level: 51

rescheduling = 3.99939

calculating = 4.94554

evaluation = 4.94549

# t1 -n5 -p51 -a512 -t111

Multi-thread evaluation, thread number = 111, priority level: 51

rescheduling = 3.99939

calculating = 5.02755

evaluation = 4.94487

# t1 -n5 -p51 -a30000 -t10000

Multi-thread evaluation, thread number = 10000, priority level: 51

rescheduling = 3.99939

calculating = 4.94575

evaluation = 5.31224

Краткий и, возможно, несколько парадоксальный итог этого теста может звучать так: при достаточно высоком уровне приоритета (выше 12–13, когда на его выполнение не влияют процессы обслуживания клавиатуры, мыши и др.) время выполнения в «классическом» последовательном коде и в многопоточном коде (где несколько тысяч потоков!) практически не различаются. Различия не более 8%, причем в обе стороны, что мы склонны считать «статистикой эксперимента». К обсуждению этого якобы противоречащего здравому смыслу феномена мы еще вернемся.

А пока посмотрим на текст примера, что и является нашей главной дачей. Обсуждаемое приложение вполне работоспособно в QNX с большой вероятностью в большинстве других UNIX-систем, но в Linux оно завершится аварийно. Причина этого кроется в операторах

int id = pthread_self() – 2;

trtime[id].s = ...

Это дает повод лишний раз обратиться к вопросу «POSIX-совместимости». POSIX описывает, что TID потока присваивается: а) в рамках процесса, которому принадлежит поток; б) начиная со значения 1, соответствующего главному потоку приложения. В Linux, выполняющем и pthread_create(), и fork()через единый системный вызов _clone()сделано небольшое «упрощение», навязанное в том числе и гонкой за повышением производительности: TID присваиваются из единого ряда PID. И сразу же «вылезает» несовместимость, ведущая к аварийному завершению показанного выше приложения. В последних редакциях ядра Linux делаются изменения по приведению механизмов параллельности к общей POSIX-модели.

Этот момент сам по себе достаточно интересен, поэтому остановимся на нем подробнее, для чего создадим простейший программный тест [22]22
Этот тест и его результаты для Linux подсказаны одним из участников (имя нам неизвестно) обсуждений на http://qnxclub.net.forum.

[Закрыть]:

#define TCNT 10

void * test(void *in) {

printf(«pid %ld, tid %ldn», getpid(), pthread_self());

return NULL;

}

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

pthread_t tld[TCNT];

int i, status;

for (i=0; i < TCNT; i++) {

status = pthread_create(&tid[i], NULL, test, NULL);

if (status != 0)

err(EXIT_FAILURE, «pthread_create()»);

}

return(EXIT_SUCCESS);

}

Результаты выполнения этого теста в нескольких POSIX-совместимых ОС различны и весьма красноречивы:

$ uname -sr Linux 2.4.21-0.13mdk

$ ./test_pthread

pid 2008, tid 16386

pid 2009, tid 32771

pid 2010, tid 49156

pid 2011, tid 65541

pid 2012, tid 81926

pid 2013, tid 98311

pid 2014, tid 114696

pid 2015, tid 131081

pid 2016, tid 147466

pid 2017, tid 163851

А вот результат эволюции в направлении POSIX при переходе от ядра Linux 2.4.x к 2.6.x (алгоритм формирования TID все еще остается загадочным, но уже выполняются требования POSIX о выделении TID в рамках единого PID):

$ uname -sr Linux 2.6.3-7mdk

$ ./test_pthread

pid 13929, tid 1083759536

pid 13929, tid 1092156336

pid 13929, tid 1100549040

pid 13929, tid 1108941744

pid 13929, tid 1117334448

pid 13929, tid 1125727152

pid 13929, tid 1134119856

pid 13929, tid 1142512560

pid 13929, tid 1150905264

pid 13929, tid 1159297968