src/processor/rt_processor.c - main - dbazhenov/rt-isolation-study

src/processor/rt_processor.c
405 строк21 KB
dbazhenov
Публикация диссертации
14 июн 2026, 16:28
14 июн 2026, 16:28fc3d9ea
/* ================================================================
 *  rt_processor.c — измеритель джиттера планировщика реального времени
 *
 *  НАЗНАЧЕНИЕ В СТЕНДЕ:
 *  Основной компонент Эксперимента 1: количественная оценка джиттера
 *  пробуждения и доли пропущенных дедлайнов при периодической
 *  обработке бортовых данных с частотой SENSOR_HZ (100 Гц).
 *  На каждом такте программа:
 *    1) засыпает по абсолютной метке времени до следующего дедлайна;
 *    2) считывает последний кадр канала 0 из разделяемой памяти,
 *       публикуемой sensor_sim;
 *    3) опционально выполняет синтетическую активную загрузку
 *       (--work-us=N) для моделирования полезной нагрузки;
 *    4) фиксирует метрики такта:
 *         • jitter_ns — отклонение момента фактического пробуждения
 *           от назначенного дедлайна (ключевая метрика);
 *         • work_ns — длительность полезной работы такта;
 *         • missed — признак пропуска дедлайна (work > period).
 *  Результаты выводятся в кольцевой буфер MetricsRing (читается
 *  dashboard.py) и в CSV-файл (читается analyze.py).
 *
 *  ИСПОЛЬЗУЕМЫЕ МЕХАНИЗМЫ LINUX:
 *    - sched_setaffinity (pin_to_core) — изоляция процесса на одном ядре;
 *    - sched_setscheduler(SCHED_FIFO) (set_realtime) — назначение
 *      политики жёсткого реального времени с приоритетом 1..99;
 *    - clock_nanosleep(CLOCK_MONOTONIC, TIMER_ABSTIME) — пробуждение
 *      по абсолютной временной метке без накопления дрейфа;
 *    - POSIX SHM (shm_open + mmap, MAP_SHARED) — взаимодействие
 *      с процессами sensor_sim (чтение) и dashboard.py (запись);
 *    - чтение /proc/stat для оценки фактической загрузки целевого ядра.
 *
 *  ПАРАМЕТРЫ ЗАПУСКА:
 *    --core=N            — ядро, к которому привязывается процесс;
 *    --rate=Hz           — частота такта (по умолчанию 100 Гц);
 *    --priority=P        — приоритет SCHED_FIFO (1..99, по умолчанию 80);
 *    --no-rt             — отказ от планировщика реального времени;
 *    --work-us=N         — длительность синтетической нагрузки, мкс;
 *    --output=file.csv   — путь к файлу метрик такта;
 *    --sensor-output=... — путь к файлу журнала сенсорных данных;
 *    --duration=s        — длительность работы в секундах.
 *
 *  ОСОБЕННОСТИ ПЛАТФОРМ:
 *    Linux: запуск из-под root или с CAP_SYS_NICE для SCHED_FIFO.
 *    macOS: требуется частичное отключение SIP; привязка к ядру
 *           носит рекомендательный характер.
 * ================================================================ */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <time.h>
#include <signal.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
#include <errno.h>

#include "../common/ipc_types.h"
#include "../common/platform.h"

/* ── Глобальные переменные ──────────────────────────────────────── */
/* Флаг продолжения работы, сбрасываемый по SIGINT/SIGTERM. */
static volatile sig_atomic_t g_running = 1;
static void on_signal(int s) { (void)s; g_running = 0; }

/* ── Оценка загрузки конкретного ядра через /proc/stat ──────────── */
/* Содержит «сырые» накопленные счётчики простоя и общего времени ядра. */
typedef struct { unsigned long long idle, total; } CpuStat;

/* Чтение строки cpu<core> из /proc/stat и заполнение структуры CpuStat.
 * Возвращает 0 при успехе, -1 если строка для запрошенного ядра не найдена. */
static int read_cpu_stat(int core, CpuStat *s)
{
    FILE *f = fopen("/proc/stat", "r");
    if (!f) return -1;
    char line[256], name[16];
    unsigned long long u, n, sy, id, iow, irq, sirq, st;
    char target[16];
    snprintf(target, sizeof(target), "cpu%d", core);
    while (fgets(line, sizeof(line), f)) {
        if (sscanf(line, "%15s %llu %llu %llu %llu %llu %llu %llu %llu",
                   name, &u, &n, &sy, &id, &iow, &irq, &sirq, &st) == 9
            && strcmp(name, target) == 0) {
            s->idle  = id + iow;
            s->total = u + n + sy + id + iow + irq + sirq + st;
            fclose(f);
            return 0;
        }
    }
    fclose(f);
    return -1;
}

/* Вычисление загрузки ядра в процентах между двумя последовательными
 * замерами /proc/stat: 100 % - (Δidle / Δtotal). */
static double cpu_pct_delta(const CpuStat *prev, const CpuStat *curr)
{
    unsigned long long dt = curr->total - prev->total;
    unsigned long long di = curr->idle  - prev->idle;
    return dt == 0 ? 0.0 : 100.0 * (1.0 - (double)di / (double)dt);
}

/* ── Накопитель статистики тактов ───────────────────────────────── */
typedef struct {
    uint64_t n;
    int64_t  sum_jitter;
    int64_t  max_jitter;
    int64_t  min_jitter;
    uint64_t sum_work;
    uint64_t max_work;
    uint64_t misses;
    uint64_t total_frames;
} Stats;

/* Обновление накопителя статистики данными очередного такта:
 * увеличиваются счётчики, обновляются минимум/максимум джиттера и работы. */
static void stats_update(Stats *s, const CycleRecord *r)
{
    s->n++;
    s->sum_jitter   += r->jitter_ns;
    s->sum_work     += r->work_ns;
    s->total_frames += (uint64_t)r->frames_consumed;
    if (r->missed) s->misses++;
    if (r->jitter_ns > s->max_jitter) s->max_jitter = r->jitter_ns;
    if (s->n == 1 || r->jitter_ns < s->min_jitter) s->min_jitter = r->jitter_ns;
    if (r->work_ns > s->max_work) s->max_work = r->work_ns;
}

/* Вывод итоговой сводной статистики в стандартный вывод. */
static void stats_print(const Stats *s)
{
    if (s->n == 0) { printf("Тактов не зарегистрировано.\n"); return; }
    double avg_j = (double)s->sum_jitter / (double)s->n / 1000.0; /* в микросекундах */
    double max_j = (double)s->max_jitter / 1000.0;
    double min_j = (double)s->min_jitter / 1000.0;
    double avg_w = (double)s->sum_work   / (double)s->n / 1000.0;
    double miss_pct = 100.0 * (double)s->misses / (double)s->n;
    printf("\n════════════════════════════════════════\n");
    printf(" Итоги работы RT-обработчика\n");
    printf("════════════════════════════════════════\n");
    printf("  Тактов            : %llu\n",  (unsigned long long)s->n);
    printf("  Джиттер, среднее  : %+.1f мкс\n", avg_j);
    printf("  Джиттер, минимум  : %+.1f мкс\n", min_j);
    printf("  Джиттер, максимум : %+.1f мкс\n", max_j);
    printf("  Работа, среднее   :  %.1f мкс\n", avg_w);
    printf("  Работа, максимум  :  %.1f мкс\n", (double)s->max_work / 1000.0);
    printf("  Пропущено         : %llu  (%.2f%%)\n",
           (unsigned long long)s->misses, miss_pct);
    printf("  Считано кадров    : %llu\n", (unsigned long long)s->total_frames);
    printf("════════════════════════════════════════\n");
}

/* ── Главная функция ────────────────────────────────────────────── */
/* Точка входа RT-обработчика. Выполняет инициализацию (привязка к ядру,
 * назначение SCHED_FIFO, открытие SHM, создание выходных файлов CSV)
 * и запускает строго периодический цикл измерения джиттера до получения
 * сигнала завершения или истечения --duration. */
int main(int argc, char **argv)
{
    int    core       = 0;
    int    rate_hz    = SENSOR_HZ;
    int    rt_prio    = 80;
    int    duration   = 0;
    int    use_rt     = 1;
    int    work_us    = 0;   /* длительность синтетической нагрузки на такт, мкс */
    char   out_path[512]    = "results/metrics.csv";
    char   sensor_path[512] = "";   /* пустая строка — журналирование сенсоров отключено */

    /* Разбор аргументов командной строки в формате --ключ=значение. */
    for (int i = 1; i < argc; i++) {
        if      (sscanf(argv[i], "--core=%d",            &core)        == 1) {}
        else if (sscanf(argv[i], "--rate=%d",            &rate_hz)     == 1) {}
        else if (sscanf(argv[i], "--priority=%d",        &rt_prio)     == 1) {}
        else if (sscanf(argv[i], "--duration=%d",        &duration)    == 1) {}
        else if (sscanf(argv[i], "--output=%511s",       out_path)     == 1) {}
        else if (sscanf(argv[i], "--sensor-output=%511s",sensor_path)  == 1) {}
        else if (sscanf(argv[i], "--work-us=%d",         &work_us)     == 1) {}
        else if (strcmp(argv[i], "--no-rt") == 0) use_rt = 0;
        else { fprintf(stderr, "Неизвестный аргумент: %s\n", argv[i]); return 1; }
    }

    long period_ns = 1000000000L / rate_hz;
    printf("[rt] core=%d  rate=%d Hz  period=%ld µs  work=%d µs  output=%s\n",
           core, rate_hz, period_ns / 1000, work_us, out_path);

    signal(SIGINT,  on_signal);
    signal(SIGTERM, on_signal);

    /* ── Привязка к выбранному ядру (sched_setaffinity) ── */
    pin_to_core(core);

    /* ── Перевод процесса в режим жёсткого реального времени:
     * политика SCHED_FIFO с приоритетом rt_prio. Это даёт процессу
     * абсолютное преимущество перед обычными задачами SCHED_OTHER. ── */
    if (use_rt) set_realtime(rt_prio);

    /* ── Открытие сегмента SHM с кадрами датчиков (только чтение).
     * Если sensor_sim не запущен — выход с диагностикой. ── */
    int sfd = shm_open(SHM_SENSORS, O_RDONLY, 0);
    if (sfd < 0) {
        fprintf(stderr, "[rt] ОШИБКА: не удалось открыть %s — запущен ли sensor_sim?\n",
                SHM_SENSORS);
        return 1;
    }
    SensorRing *sring = mmap(NULL, sizeof(SensorRing),
                             PROT_READ, MAP_SHARED, sfd, 0);
    close(sfd);
    if (sring == MAP_FAILED) { perror("mmap sensors"); return 1; }

    /* ── Создание сегмента SHM для публикации метрик (читается dashboard.py). ── */
    int mfd = shm_open(SHM_METRICS, O_CREAT | O_RDWR, 0666);
    if (mfd < 0) { perror("shm_open metrics"); return 1; }
    if (ftruncate(mfd, sizeof(MetricsRing)) == -1) { perror("ftruncate metrics"); return 1; }
    MetricsRing *mring = mmap(NULL, sizeof(MetricsRing),
                              PROT_READ | PROT_WRITE, MAP_SHARED, mfd, 0);
    close(mfd);
    if (mring == MAP_FAILED) { perror("mmap metrics"); return 1; }
    memset(mring, 0, sizeof(MetricsRing));
    atomic_store(&mring->write_idx, 0);

    /* ── Открытие CSV-файла метрик и запись заголовка. ── */
    FILE *csv = fopen(out_path, "w");
    if (!csv) {
        fprintf(stderr, "[rt] Предупреждение: не удалось открыть %s: %s\n",
                out_path, strerror(errno));
    } else {
        fprintf(csv, "cycle,deadline_ns,wakeup_ns,jitter_ns,"
                     "work_ns,missed,frames_consumed\n");
    }

    /* ── Открытие журнала сенсорных данных (если задан --sensor-output). ── */
    FILE *scsv = NULL;
    if (sensor_path[0] != '\0') {
        scsv = fopen(sensor_path, "w");
        if (!scsv) {
            fprintf(stderr, "[rt] Предупреждение: не удалось открыть журнал %s: %s\n",
                    sensor_path, strerror(errno));
        } else {
            fprintf(scsv,
                "cycle,ts_ns,"
                "alt_m,cas_kt,"
                "gyro_x,gyro_y,gyro_z,"
                "accel_x,accel_y,accel_z,"
                "cpu_pct\n");
            printf("[rt] sensor log → %s\n", sensor_path);
        }
    }

    /* ── Базовый замер загрузки ядра для последующего расчёта Δ. ── */
    CpuStat cpu_prev = {0, 0};
    read_cpu_stat(core, &cpu_prev);

    /* ── Локальное состояние цикла ── */
    Stats         stats  = { 0, 0, INT64_MIN, INT64_MAX, 0, 0, 0, 0 };
    uint64_t      cycle  = 0;

    /* Привязка начала цикла к границе следующего периода
     * (выравнивание дедлайнов по сетке кратной period_ns). */
    struct timespec next;
    clock_gettime(CLOCK_MONOTONIC, &next);
    next.tv_nsec += period_ns;
    while (next.tv_nsec >= 1000000000L) {
        next.tv_nsec -= 1000000000L;
        next.tv_sec++;
    }

    uint64_t read_pos = atomic_load(&sring->read_idx);

    printf("[rt] работа запущена — нажмите Ctrl-C для остановки\n\n");

    while (g_running) {

        /* ════ Сон до назначенного дедлайна ════
         * abstime_sleep вызывает clock_nanosleep(TIMER_ABSTIME); это
         * единственно корректный способ обеспечить периодичность без
         * накопления дрейфа от шага к шагу. */
        uint64_t deadline_ns = (uint64_t)next.tv_sec * 1000000000ULL
                               + (uint64_t)next.tv_nsec;
        abstime_sleep(&next);
        uint64_t wakeup_ns = mono_ns();

        /* ════ Фаза полезной работы такта ════ */
        uint64_t work_start = wakeup_ns;

        /* Чтение последнего доступного кадра канала 0 из кольцевого буфера.
         * read_pos продвигается до write_pos, по пути отбирается самый
         * свежий кадр с src == 0 (для «сухой» обработки берётся последний). */
        SensorFrame latest;
        memset(&latest, 0, sizeof(latest));
        int got_frame  = 0;
        int frames_read = 0;

        uint64_t write_pos = atomic_load_explicit(&sring->write_idx,
                                                  memory_order_acquire);
        while (read_pos < write_pos) {
            const SensorFrame *f = &sring->buf[read_pos & RING_MASK];
            if (f->src == 0) {
                latest    = *f;
                got_frame = 1;
            }
            read_pos++;
            frames_read++;
        }

        /* ════ Журналирование сенсорных данных в CSV (опционально) ════ */
        if (scsv && got_frame) {
            CpuStat cpu_now = {0, 0};
            read_cpu_stat(core, &cpu_now);
            double cpu_p = cpu_pct_delta(&cpu_prev, &cpu_now);
            cpu_prev = cpu_now;

            fprintf(scsv,
                "%llu,%llu,"
                "%.2f,%.3f,"
                "%.4f,%.4f,%.4f,"
                "%.4f,%.4f,%.4f,"
                "%.1f\n",
                (unsigned long long)cycle,
                (unsigned long long)wakeup_ns,
                latest.alt_m,       latest.cas_kt,
                latest.gyro_dps.x,  latest.gyro_dps.y,  latest.gyro_dps.z,
                latest.accel_mss.x, latest.accel_mss.y, latest.accel_mss.z,
                cpu_p);
        }

        /* ════ Синтетическая активная нагрузка (--work-us=N) ════
         * Имитирует более тяжёлую RT-задачу: позволяет демонстрировать
         * пропуски дедлайнов при конкуренции за ядро. Объявление volatile
         * препятствует устранению цикла оптимизатором. */
        if (work_us > 0) {
            uint64_t spin_end = mono_ns() + (uint64_t)work_us * 1000ULL;
            volatile double x = 1.0;
            while (mono_ns() < spin_end)
                x = x * 1.00001 + 0.00001;   /* предотвращает удаление кода оптимизатором */
            (void)x;
        }

        /* ════ Регистрация метрик завершённого такта ════
         * missed: завершилась ли работа ПОСЛЕ следующего дедлайна,
         * то есть прошло ли от назначенного пробуждения больше периода. */
        uint64_t work_end = mono_ns();
        CycleRecord rec = {
            .cycle           = cycle,
            .deadline_ns     = deadline_ns,
            .wakeup_ns       = wakeup_ns,
            .jitter_ns       = (int64_t)(wakeup_ns - deadline_ns),
            .work_ns         = work_end - work_start,
            .missed          = (work_end - deadline_ns) > (uint64_t)period_ns,
            .frames_consumed = frames_read
        };

        /* Запись метрик в кольцевой буфер SHM без блокировок;
         * memory_order_release гарантирует видимость данных rec до
         * увеличения индекса write_idx со стороны читателя. */
        uint64_t midx = atomic_load_explicit(&mring->write_idx,
                                             memory_order_relaxed);
        mring->buf[midx & METRICS_MASK] = rec;
        atomic_store_explicit(&mring->write_idx, midx + 1,
                              memory_order_release);

        /* Дублирование записи в CSV для последующего анализа в analyze.py. */
        if (csv)
            fprintf(csv, "%llu,%llu,%llu,%lld,%llu,%d,%d\n",
                    (unsigned long long)rec.cycle,
                    (unsigned long long)rec.deadline_ns,
                    (unsigned long long)rec.wakeup_ns,
                    (long long)rec.jitter_ns,
                    (unsigned long long)rec.work_ns,
                    rec.missed,
                    rec.frames_consumed);

        stats_update(&stats, &rec);
        cycle++;

        /* Проверка ограничения по длительности (--duration). */
        if (duration > 0 && (double)cycle / rate_hz >= duration) break;

        /* Смещение дедлайна на следующий период. */
        ts_add_ns(&next, period_ns);

        /* Раз в 5 секунд выводится строка прогресса. */
        if (cycle % (uint64_t)(rate_hz * 5) == 0) {
            printf("[rt] cycle %6llu | jitter=%+7.1f µs | work=%5.1f µs | miss=%llu\r",
                   (unsigned long long)cycle,
                   (double)rec.jitter_ns / 1000.0,
                   (double)rec.work_ns   / 1000.0,
                   (unsigned long long)stats.misses);
            fflush(stdout);
            if (csv) fflush(csv);
        }
    }

    putchar('\n');
    stats_print(&stats);

    if (csv)  fclose(csv);
    if (scsv) fclose(scsv);
    shm_unlink(SHM_METRICS);
    munmap(sring, sizeof(SensorRing));
    munmap(mring, sizeof(MetricsRing));
    return 0;
}