POSIX.1-202x: CPU-clocks, Monotonic clock, и таймауты

Мы затронули интересную тему, когда занимались реализацией CPU-clock и очередей событий в планировщике.
Планировщик живет одновременно в нескольких местах:
1. обработка прерываний от таймера и
2. процесс переключения задач.

В качестве системного таймера мы традиционно используем системный таймер SysTik, но можем использовать еще множество других таймеров. В частности, мы стали использовать счетчик циклов DWT для измерения CPU-clock, что позволяет выразить время работы треда в тактах процессора. В нашей системе SysTik работает на частоте 25кГц. Когда мы рассматриваем многоядерные архитектуры, может быть несколько системных таймеров, каждый со своим разрешением. Идентификатор системного таймера, на котором работает процесс, можно запросить функцией pthread_getcpuclockid() (TCT - POSIX Thread CPU-clock option). В новом стандарте POSIX.1-202x предлагают поддержать ожидание с явным указанием идентификатора системного таймера. Стандарт предлагают дополнить функциями синхронизации тредов с суффиксом *_clockwait, по аналогии с суффиксом *_timedwait.

sem_clockwait
pthread_mutex_clocklock
pthread_cond_clockwait
pthread_rwlock_clockrdlock
pthread_rwlock_clockwrlock
mq_clockreceive
mq_clocksend
. . .

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

В настоящее время мы используем одну очередь событий и один системный таймер. Процессы, ожидающие событие никак не обрабатываются в таймерном прерывании, обрабатываютя в режиме FIFO в планировщике. Использование функций с явным указанием на таймер позволяет ставить событие в очередь для более точной обработки таймаута из прерывания. Работа таймера: если произошло таймерное событие, то инициируется последующая передача управления планировщику с указанием, ради какого треда это переключение выполняется. Таймерные события организуются с использованием асинхронной очереди с ограниченной глубиной и сортируются по дедлайну (времени срабатывания). Один из вариантов таймерного события - запуск треда с нарушением очередности запуска процессов в планировщике.

POSIX предполагает использовать функции *_clockwait ради отказа от CLOCK_REALTIME в пользу CLOCK_MONOTONIC. Функции *_timedwait используют CLOCK_REALTIME, который должен соответствовать TIME_UTC в стандарте языка Си. CLOCK_REALTIME не является монотонным, может подстраиваться при необходимости. В обоих случаях виден некоторый изъян с неоднозначностью вычисления интервала времени, на который настраивается срабатывание таймера. Приходится дважды пересчитывать: сначала в прикладном коде к интервалу ожидания прибавлять текущее время, затем в системном коде вычитать абсолютное время и высчитывать задержку на срабатывание. Это черевато сбоями при подстройке времени. Если мы обеспечим монотонность CLOCK_REALTIME или станем выражать абсолютное время через монотонное время CLOCK_MONOTONIC, то таких проблем не возникает.

// пример настройки интерфейса POSIX
pthread_getcpuclockid(self, &clock_id);
clock_gettime(clock_id, &abstime);
timespec_add (&abstime, interval);
sem_clockwait(sem, clock_id, abstime);

Нужно стандартизировать функцию timespec_add(), чтобы убрать неоднозначность вычисления временных интервалов. Одновременно надо стандартизировать функцию сравнения timespec_cmp() для использования в со стороны RTOS.

С2х определяет таймеры TIME_THREAD_ACTIVE, которые в POSIX соответсвуют выбору CLOCK_THREAD_CPUTIME_ID. В нашей интерпретации эти времена могут высчитываться с большей точностью, чем CLOCK_MONOTONIC, которое обновляется в системном таймере и имеет сравнительно низкое разрешение.

В С1x всего три функции, которые завязаны на время ожидания: mtx_timedlock() cnd_timedwait() и thrd_sleep(). Ожидание задается дедлайном относительно TIME_UTC, что соответствует CLOCK_REALTIME в POSIX.

// пример настройки интерфейса C1x
struct timespec ts;
timespec_get(&ts, TIME_UTC);
timespec_add(&abstime, interval);
mtx_timedlock(mtx, &ts);

Внутри фукнции mtx_timedlock приходится еще раз запрашивать timespec_get и вычитать из абсолютного времени величину TIME_UTC, чтобы получить интервал ожидания - таймаут. Я думаю, как бы определить эти три функции, чтобы при компиляции не надо было два раза запрашивать системное время и не надо было бы прибавлять и вычитать переменную величину, чтобы с некоторой вероятностью опять получить значение интервала. Это не удобно. Это плохо, хочу использовать относительные времена ожидания.

// расшифровка, псевдокод
struct timespec ts;
ts = system_time; // копируем время
// добавляем смещение UTC
ts.tv_sec += UTC.tv_sec, ts.tv_nsec += UTC.tv_nsec;
if(ts.tv_nsec>=1000000000) {// коррекция при сложении
   ts.tv_nsec-=1000000000;
   ts.tv_sec++;
}
// добавляем Интервал времени ожидания
ts.tv_sec += interval.tv_sec, ts.tv_nsec += interval.tv_nsec;
if(ts.tv_nsec>=1000000000) {// коррекция
   ts.tv_nsec-=1000000000;
   ts.tv_sec++;
}
ts2 = system_time; // копируем время еще раз
ts2.tv_sec = ts.tv_sec - (ts2.tv_sec + UTC.tv_sec);
ts2.tv_nsec= ts.tv_nsec - (ts2.tv_sec + UTC.tv_nsec);
// вычисляем таймаут в микросекундах
timeout = (ts2.tv_sec*1000000 + ts2.tv_nsec/1000);
// передаем управление системе
syscall(mtx_timedlock, mtx, &timeout);

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

Давайте представим, что людьми предложившими использовать абсолютное время для ожидания мьютекса двигали возвышенные цели. Например, они предполагали, что обработка системного вызова происходит в другом часовом поясе и на другом компьютере, где TIME_MONOTONIC другой и их нельзя сравнивать. Тогда что мешает сравнивать интервалы?!

Следующая проблема, в ядре. Счетчик микросекунд - переменная, которая считает время с большим разрешением, переодически переполняется. Чтобы переполнение не сказывалось на проверке и разнца монотонно возрастала, мы приводим тип к беззнаковому:
(unsigned int)(clock() - activation_time) >= timeout

Как сделать тоже самое с абсолютным временем я не знаю. Все что приходит в голову - считать разницу в 64-битных числах. Т.е. в этом случае нам придется пересчитывать 64 битное монотонно возрастающее время в struct timespec и обратно бесчисленное количество раз. Хотим избавиться от бессмысленных операций. Вспоминаются числа с фиксированной точкой, можно хранить время в виде числа double.
Следующее приближение: как вычисление времени выглядит с использованием 64 битного времени:

// расшифровка
struct timespec ts;
ts = system_time; // копируем время
ts+= UTC;
ts+= interval;
ts2 = system_time; // копируем время еще раз
ts2 = ts - (ts2 + UTC);
timeout = ts2;
syscall(mtx_timedlock, mtx, &timeout);

Казалось бы такую последовательность возможно упростить до одного присваивания timeout = interval. Чтобы это стало возможным, нужно чтобы system_time не менялся. Т.е. при оптимизации компилятор должен взять значение system_time из регистра, иначе не упростит.

Следующее приближение. Время ожидания - аблолютное, выражено через монотонное системное время.

// расшифровка
struct timespec abstime;
abstime = system_time + UTC + interval;
abstime-= UTC;
syscall(mtx_timedlock, mtx, &abstime);

Со стороны системы обработка выглядит так:

// расшифровка
activation_time = system_time;
timeout  = abstime - activation_time;
. . .
if ((system_time - activation_time) >= timeout) . . .

Вот в таком виде я готов смириться, в ядре лишних действий не совершается.
Для реализации такого способа нужно, чтобы время складывалось и вычиталось, как 64 битное число:
t64 = ((uint64_t)ts.tv_sec<<32) | (uint32_t)ts.tv_nsec;
Коррекция 64 битного времени при сложении:
if ((uint32_t)t64 >= 1000000000) t64 += (1<<32)-1000000000;
Коррекция 64 битного времени при вычитании:
if ((int32_t)t64 < 0) t64 -= (1<<32)-1000000000;

Нужно дать свободу пользователю и хотелось бы придержиываться стандарта. При работе с 64 битным временем упакованным в ту же структуру struct timespec, вычисление интервала может выглядеть так:

// расшифровка
timespec_get(struct timespec *ts, int base){
    ts->u64 = system_monotonic_time + CLOCK_OFFSET(base);
}
timespec_add(struct timespec *ts, struct timespec *interval){
    ts->u64 += interval->u64;
}
mtx_timedlock(mtx_t *mtx, struct timespec *ts){
    ts->u64 -= CLOCK_OFFSET(TIME_UTC);
// выполнить коррекцию
	if(ts->tv_nsec<0) t64 -= (1<<32)-1000000000;
	else
	if(ts->tv_nsec>=1000000000) ts->u64 += (1<<32)-1000000000;
     . . .
}

CLOCK_OFFSET(TIME_MONOTONIC) равен нулю. Если все три функции доступны для оптимизации, то возможно компилятор упростит выражение до
ts->u64 = system_monotonic_time + interval->u64;

Заключение
Весь этот поток сознания вызыван стремелением рабочей группы по стандартизации POSIX привнести ряд функций в стандрт ради привязки времени ожидания к монотонному времени. Я в свою очередь стал анализировать, к чему это приводит в коде RTOS и что требуется реализовать в ядре для поддержки этого функционала. Меня очень беспокоит "кривизна" вычисления задержек и связанные с этим сбои. Хотелось бы избежать этих проблем. Проблему я вижу в том, что изначально принято использовать абсолютное время для ожидания, побочные явления от такого подхода могут привести к сбоям задач работающих в условиях жестких требований к времени исполнения (hard real-time).
Источник:
[1] The Open Group Standard, Additional APIs for the Base Specifications Issue 8, Part 1, 2021
[2] P1003.1™-202x, Draft 2.1+, September 2022 – Portable Operating System Interface (POSIX®)