협동형 스케줄러 만들기

이 장에서 배우는 것

앞 장에서 컨베이어의 동작을 상태와 전이로 나누었다. 상태를 잘 나누어도 각 상태를 검사하는 시점이 들쭉날쭉하면 센서 반응과 모터 제어의 간격이 흔들린다. 이제 “무엇을 할 것인가”에 더해 “언제 실행할 것인가”를 코드로 표현한다.

협동형 스케줄러(cooperative scheduler)는 실행 중인 작업이 스스로 반환해야 다음 작업을 실행할 수 있는 구조다. 이 장에서는 센서 읽기, 모터 제어, 상태 보고를 짧은 함수로 만들고 하나의 실행 흐름에서 차례로 호출한다. 운영체제의 시간 흐름에 결과가 좌우되지 않도록 PC에서 가상 시간을 사용하는 시뮬레이터도 함께 만든다.

• 기본 타임 슬롯(time slot)을 정하고 서로 다른 주기의 작업을 배치한다.
• 이전 실행의 종료 시각과 구분되는 주기적 실행 기준을 유지한다.
• 예정 시각과 실제 시작 시각의 차이로 시작 지터(jitter)를 측정한다.
• 긴 작업 하나가 뒤의 작업에 미치는 영향을 실행 기록으로 확인한다.
• 직접 만든 실행 구조와 FreeRTOS의 관련 기능이 대응하는 범위를 구분한다.

문제 상황

공장 컨베이어 제어기에는 광센서 입력을 읽는 함수, 센서 결과로 모터 출력을 결정하는 함수, 운전 상태를 보고하는 함수가 있다. 처음에는 메인 반복문에서 세 함수를 연달아 호출한 뒤 잠시 기다렸다. 센서는 자주 읽혔지만 보고 기능을 추가한 뒤 모터 출력의 갱신 간격이 길어졌다.

원인은 각 기능의 실행 빈도가 반복문 전체의 길이에 묶였기 때문이다. 보고 함수가 오래 걸리면 다음 센서 읽기도 늦어진다. 반복문 끝에 10ms 대기를 넣더라도 반복 간격은 10ms가 아니다. 세 함수의 실행 시간과 대기 시간이 합쳐진 값이다.

이번 요구 사항은 센서 읽기 10ms, 모터 제어 20ms, 상태 보고 50ms다. 센서가 물체를 감지하면 제어 함수가 모터를 끄고, 감지가 해제되면 다시 켠다. 실제 설비의 정지 회로나 안전 기능 전체를 모델링하는 예제는 아니다. 여기서는 이 단순한 동작을 이용해 실행 시각의 흔들림을 관찰한다.

정상적인 보고는 4ms가 걸리지만 두 번째 보고에서는 전송 준비가 길어져 14ms가 걸린다고 가정한다. 이때 뒤따르는 센서와 제어가 얼마나 늦어지는지 숫자로 확인하는 것이 목표다. 평균 실행 간격만 보면 짧은 지연 구간이 가려지므로 작업별 시작 시각을 남긴다.

타임 슬롯으로 실행 기회를 나눈다

기본 슬롯을 10ms로 정하면 센서는 매 슬롯, 제어는 두 슬롯마다, 보고는 다섯 슬롯마다 실행할 수 있다. 세 주기가 모두 10ms의 정수 배수이므로 별도의 소수 시간 계산이 필요 없다. 시작 기준은 모두 0ms로 두며, 같은 슬롯에 여러 작업이 있으면 센서, 제어, 보고 순서로 실행한다.

슬롯은 작업에 CPU 시간을 강제로 나누어 주는 장치가 아니다. 이 구현에서는 작업을 호출할 실행 기회를 나타낸다. 작업이 슬롯 경계를 넘겨도 스케줄러가 함수를 중단하지 않는다. 실행 시간을 제한하려면 작업 자체를 짧게 만들고 반환하도록 설계해야 한다.

컨베이어 작업의 주기와 슬롯 내 실행 순서

작업	주기	모의 실행 시간	같은 슬롯의 순서
센서 읽기	10ms	2ms	첫 번째
모터 제어	20ms	3ms	두 번째
상태 보고	50ms	4ms, 두 번째는 14ms	세 번째

0ms 슬롯에는 세 작업이 모두 들어간다. 정상 실행 시간의 합은 9ms이므로 다음 슬롯까지 1ms가 남는다. 10ms 슬롯에는 센서만 들어가고 8ms가 남는다. 이 남은 시간은 다음 예정 시각까지 기다리는 구간이다. PC 예제에서는 가상 시각을 앞으로 옮기며, 실제 보드에서는 타이머를 확인하거나 적절한 대기 기능을 사용할 수 있다.

그림의 칸은 실행 대상의 목록이며 작업 길이를 나타내는 막대가 아니다. 같은 칸의 작업들은 위에서 아래로 실행된다. 센서와 제어가 같은 슬롯에 있으면 센서를 먼저 읽으므로 제어는 그 슬롯에서 갱신한 값을 사용할 수 있다. 배열의 순서를 바꾸는 일도 동작을 바꾸는 설계 변경이다.

모든 작업을 0ms에서 시작해야 하는 것은 아니다. 보고의 첫 실행을 다른 슬롯으로 옮기면 실행이 한곳에 몰리는 정도가 달라진다. 이러한 시작 오프셋(offset)은 작업 사이의 데이터 관계와 함께 정해야 한다. 이번 코드는 순서와 지연의 관계를 쉽게 확인하도록 시작 기준을 동일하게 둔다.

주기 기준을 유지하고 늦은 슬롯을 처리한다

주기 작업(periodic task)의 예정 시각은 기준 시각에 주기의 정수 배를 더한 값이다. 센서의 예정 시각은 0, 10, 20ms로 이어진다. 실제 시작이 66ms까지 늦어졌더라도 해당 실행의 예정 시각은 60ms다. 이 둘을 별도로 보관해야 지연을 측정할 수 있다.

작업이 끝난 뒤 현재 시각에 주기를 더하면 실행 시간이 주기에 섞인다. 2ms짜리 센서 작업이 0ms에 시작하고 종료 시각에 10ms를 더하면 다음 예정 시각은 12ms가 된다. 같은 계산을 반복하면 24, 36ms로 기준이 밀린다. 이번 구현은 이전 예정 시각에 주기를 더해 10ms 간격의 기준을 유지한다.

현재 시각이 다음 슬롯을 지났을 때 무엇을 할지도 정해야 한다. 이 장에서는 지나간 슬롯을 순서대로 처리하는 정책을 사용한다. 보고가 66ms에 끝나면 60ms 슬롯을 즉시 처리하고, 그 처리가 71ms에 끝나면 70ms 슬롯도 즉시 처리한다. 80ms 슬롯을 처리하기 전에는 다시 여유가 생긴다.

이 정책은 늦은 실행을 기록하기 쉽지만 과거 입력을 복원하지는 않는다. 60ms 예정 센서를 66ms에 호출하면 66ms의 입력을 읽는다. 통신 요청처럼 실행 횟수 자체가 의미 있는 작업과, 최신 상태만 필요한 작업은 밀린 실행을 처리하는 방식이 다를 수 있다. 실제 설계에서는 누적 실행, 건너뛰기, 최신 한 번만 실행하기 가운데 요구 사항에 맞는 정책을 선택해야 한다.

이 예제의 반복문은 논리적인 슬롯을 하나씩 방문한다. 그래서 작업의 다음 예정 시각과 슬롯 시각이 같은지 비교해도 실행 기회를 잃지 않는다. 실제 시계를 직접 읽으면서 우연히 정확한 시각이 될 때까지 기다리는 코드와는 다르다. 실제 시계는 읽는 사이에 여러 시간 단위를 건너뛸 수 있다.

협동형 구조에서는 기다리는 방식도 작업 길이에 포함된다. 보고 함수가 전송 완료를 반복해서 검사하며 머무르면 그동안 센서를 실행할 수 없다. 앞 장에서 만든 상태 기계를 활용하면 전송 시작, 진행 확인, 완료 처리를 짧은 호출로 나눌 수 있다. 각 호출이 빠르게 반환해야 스케줄러에도 다음 실행을 결정할 기회가 생긴다.

시작 지터를 같은 기준으로 측정한다

이 장에서 측정하는 시작 지터는 “실제 시작 시각 − 예정 시각”으로 정의한다. 구현이 예정 시각보다 일찍 작업을 호출하지 않으므로 값은 0 이상이다. 문맥에 따라 지터가 실행 간격의 변동 폭을 뜻하기도 하므로, 측정 기록에는 어떤 차이를 구했는지 명시해야 한다.

슬롯 지연과 작업 지터는 다르다. 0ms 슬롯은 제때 시작하지만 제어는 센서가 반환한 2ms에 시작한다. 따라서 제어의 시작 지터는 2ms다. 이는 이번에 정한 공통 예정 시각과 직렬 실행 순서에서 생긴 값이다. 같은 슬롯의 모든 작업이 동시에 시작할 수 있다고 해석해서는 안 된다.

기록에는 예정 시각, 시작 시각, 종료 시각을 함께 남긴다. 시작에서 예정을 빼면 시작 지터이고, 종료에서 시작을 빼면 실행 시간이다. 둘을 섞으면 “작업이 오래 걸렸다”와 “앞 작업 때문에 늦게 시작했다”를 구분하기 어렵다.

시뮬레이터의 하드웨어 추상화 계층(HAL)은 호스트의 실제 시간을 사용하지 않는다. 작업이 2ms를 소비한다고 선언하면 가상 시각을 정확히 2ms 증가시킨다. macOS와 Linux의 스케줄링 차이 없이 같은 결과를 얻는 대신, 함수 호출이나 메모리 접근에 걸리는 실제 CPU 시간은 측정하지 않는다. 따라서 출력은 정해 둔 실행 시간 모델의 결과이며 보드의 성능 측정값은 아니다.

실제 보드로 옮길 때는 단조롭게 증가하는 시간 기준으로 작업 직전과 직후를 측정한다. 측정 단위보다 짧은 변동은 구분할 수 없고, 기록 자체도 실행 시간을 늘린다. 예제처럼 실행 중에는 메모리에 기록하고 측정 구간이 끝난 뒤 출력하면 출력 함수가 관찰 대상을 크게 바꾸는 일을 줄일 수 있다.

직접 만든 스케줄러와 FreeRTOS 기능의 대응 범위

목적	이번 구현	FreeRTOS의 관련 기능	구분할 점
반복할 작업 등록	작업 배열과 함수 포인터	xTaskCreate()	이번 함수에는 독립된 태스크 스택이 없다.
일정한 주기 기준 유지	next_ms에 period_ms 누적	xTaskDelayUntil()	FreeRTOS에서는 틱 단위의 기준 시각을 사용한다.
현재 시점부터 대기	이번 주기 계산에는 사용하지 않음	vTaskDelay()	호출 시점에 대한 상대 대기는 실행 시간 누적을 막지 못한다.
다음 실행에 기회 제공	작업 함수에서 반환	taskYIELD(), 지연으로 대기 상태 진입	단순 반환과 문맥 전환은 다른 동작이다.
시간 기준 읽기	hal_sim_now()	xTaskGetTickCount()	틱 해상도보다 세밀한 측정에는 별도 시간원이 필요하다.

이 표는 목적의 대응표다. 함수 이름을 바꾸는 것만으로 같은 프로그램이 되지는 않는다. 특히 FreeRTOS 태스크의 진입 함수를 이번 작업 함수처럼 매번 반환하도록 옮기면 안 된다. 독립된 실행 흐름과 대기 상태를 갖는 태스크의 작성 방식은 다음 장에서 다룬다. API 사실 확인에는 FreeRTOS의 주기 지연 API 문서를 참고할 수 있다.

완성 코드

다음 전체 내용을 scheduler.c로 저장한다. 외부 라이브러리나 별도 입력 파일은 필요 없다. hal_sim으로 시작하는 함수들이 PC 시뮬레이션 계층이고, run_scheduler()가 간단한 협동형 스케줄러다. 실행 대상은 예정 시각이 0ms 이상 100ms 미만인 작업이며, 100ms에 예정된 작업은 포함하지 않는다.

시간은 이 짧은 실험에서 충분한 범위의 unsigned 정수로 표현한다. 장시간 동작할 제품에서는 카운터 범위와 순환 처리를 별도로 설계해야 한다. 여기서는 유한한 실험의 실행 순서를 분명히 보여 주는 데 집중한다.

#include <stdbool.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>

/* [01] Experiment configuration */
enum {
    SLOT_MS = 10,
    WINDOW_MS = 100,
    TRACE_CAPACITY = 32
};

/* [02] Simulated hardware */
typedef struct {
    unsigned now_ms;
    bool sensor_blocked;
    bool motor_on;
} HalSim;

static HalSim hal_sim;

static unsigned hal_sim_now(void)
{
    return hal_sim.now_ms;
}

static void hal_sim_wait_until(unsigned target_ms)
{
    if (hal_sim.now_ms < target_ms) {
        hal_sim.now_ms = target_ms;
    }
}

static void hal_sim_consume(unsigned duration_ms)
{
    hal_sim.now_ms += duration_ms;
}

static bool hal_sim_read_sensor(void)
{
    return hal_sim.now_ms >= 40u && hal_sim.now_ms < 80u;
}

/* [03] Short, returning work functions */
static void sensor_step(unsigned release_ms)
{
    (void)release_ms;
    hal_sim.sensor_blocked = hal_sim_read_sensor();
    hal_sim_consume(2u);
}

static void control_step(unsigned release_ms)
{
    (void)release_ms;
    hal_sim.motor_on = !hal_sim.sensor_blocked;
    hal_sim_consume(3u);
}

static void report_step(unsigned release_ms)
{
    /* Model preparation and transmission time only. */
    hal_sim_consume(release_ms == 50u ? 14u : 4u);
}

/* [04] Task descriptions and measurements */
typedef void (*StepFunction)(unsigned release_ms);

typedef struct {
    const char *name;
    unsigned period_ms;
    unsigned next_ms;
    StepFunction step;
    unsigned runs;
    unsigned max_jitter_ms;
    unsigned sum_jitter_ms;
} Task;

typedef struct {
    const char *name;
    unsigned release_ms;
    unsigned start_ms;
    unsigned end_ms;
} Trace;

typedef struct {
    unsigned late_slots;
    unsigned max_late_ms;
} SlotStats;

static Task tasks[] = {
    { "sensor",  10u, 0u, sensor_step,  0u, 0u, 0u },
    { "control", 20u, 0u, control_step, 0u, 0u, 0u },
    { "report",  50u, 0u, report_step,  0u, 0u, 0u }
};

static Trace traces[TRACE_CAPACITY];
static size_t trace_count;

/* [05] Execute one release and record its timing */
static bool dispatch(Task *task)
{
    if (trace_count >= TRACE_CAPACITY) {
        return false;
    }

    unsigned release_ms = task->next_ms;
    unsigned start_ms = hal_sim_now();
    unsigned jitter_ms = start_ms - release_ms;

    task->step(release_ms);

    Trace *entry = &traces[trace_count++];
    entry->name = task->name;
    entry->release_ms = release_ms;
    entry->start_ms = start_ms;
    entry->end_ms = hal_sim_now();

    task->runs++;
    task->sum_jitter_ms += jitter_ms;
    if (jitter_ms > task->max_jitter_ms) {
        task->max_jitter_ms = jitter_ms;
    }
    task->next_ms += task->period_ms;
    return true;
}

/* [06] Visit every logical slot, including late slots */
static bool run_scheduler(SlotStats *stats)
{
    size_t task_count = sizeof tasks / sizeof tasks[0];

    for (unsigned slot_ms = 0u;
         slot_ms < WINDOW_MS;
         slot_ms += SLOT_MS) {
        hal_sim_wait_until(slot_ms);

        unsigned late_ms = hal_sim_now() - slot_ms;
        if (late_ms != 0u) {
            stats->late_slots++;
        }
        if (late_ms > stats->max_late_ms) {
            stats->max_late_ms = late_ms;
        }

        for (size_t i = 0u; i < task_count; ++i) {
            if (tasks[i].next_ms == slot_ms) {
                if (!dispatch(&tasks[i])) {
                    return false;
                }
            }
        }
    }
    return true;
}

/* [07] Print after the measured simulation */
static void print_results(const SlotStats *stats)
{
    size_t task_count = sizeof tasks / sizeof tasks[0];

    puts("time unit: ms");
    puts("task release start end jitter");
    for (size_t i = 0u; i < trace_count; ++i) {
        const Trace *entry = &traces[i];
        printf("%s %u %u %u %u\n",
               entry->name, entry->release_ms,
               entry->start_ms, entry->end_ms,
               entry->start_ms - entry->release_ms);
    }

    puts("summary: task runs max_jitter sum_jitter");
    for (size_t i = 0u; i < task_count; ++i) {
        const Task *task = &tasks[i];
        printf("%s %u %u %u\n",
               task->name, task->runs,
               task->max_jitter_ms, task->sum_jitter_ms);
    }

    printf("slots late=%u max_late=%u\n",
           stats->late_slots, stats->max_late_ms);
    printf("finish=%u motor=%s\n",
           hal_sim_now(), hal_sim.motor_on ? "on" : "off");
}

/* [08] One finite experiment */
int main(void)
{
    SlotStats stats = { 0u, 0u };

    if (!run_scheduler(&stats)) {
        fputs("trace capacity exceeded\n", stderr);
        return EXIT_FAILURE;
    }

    print_results(&stats);
    return EXIT_SUCCESS;
}

줄별 해설

[01] SLOT_MS는 슬롯 사이의 간격이고 WINDOW_MS는 예정 시각을 선택하는 구간의 끝이다. 두 값이 전체 프로그램의 실행 시간을 직접 강제하지는 않는다. 마지막 대상 작업이 늦게 끝나면 종료 시각이 구간의 끝보다 뒤로 갈 수도 있다. 기록 배열은 이번 실행에 필요한 17개보다 넉넉한 32개로 고정한다.

[02] HalSim에는 현재 가상 시각, 마지막으로 읽은 센서 값, 모터 출력이 들어간다. 정적 저장 기간을 가지므로 시작할 때 시각과 불리언 값이 모두 0으로 초기화된다. hal_sim_wait_until()은 목표 시각이 미래일 때만 시간을 옮긴다. 이미 늦었다면 시간을 되돌리지 않는다.

hal_sim_consume()은 작업이 차지한 시간을 모델에 반영한다. 실제로 CPU를 그 시간만큼 바쁘게 돌리거나 잠들게 하지는 않는다. hal_sim_read_sensor()는 현재 시각이 40ms 이상 80ms 미만이면 물체가 있다고 반환한다. 따라서 늦게 읽은 입력은 늦어진 시점의 상태다.

[03] 세 작업은 모두 한 번의 일을 수행하고 반환한다. 공통 함수 형식에 맞추기 위해 예정 시각을 인수로 받으며, 사용하지 않는 두 함수는 명시적으로 사용하지 않음을 표시한다. 센서는 입력을 읽은 뒤 2ms를 소비하고, 제어는 저장된 센서 값으로 모터를 설정한 뒤 3ms를 소비한다.

report_step()은 보고 내용의 직렬화나 실제 통신 대신 소요 시간만 모델링한다. 예정 시각이 50ms인 실행에만 14ms를 부여한다. 실제 시작 시각으로 조건을 판단하면 지연 때문에 조건 자체가 달라질 수 있으므로 어떤 실행에 지연을 주는지 예정 시각으로 지정한다.

[04] StepFunction은 예정 시각을 받고 반환값이 없는 작업 함수의 형식이다. Task는 고정적인 설정과 실행 중 갱신되는 통계를 함께 가진다. next_ms는 아직 실행하지 않은 다음 호출의 예정 시각이다. 배열에 센서, 제어, 보고를 이 순서로 넣었으므로 같은 슬롯에서도 순서가 유지된다.

Trace는 한 번의 호출에 관한 시각 세 개를 보관한다. SlotStats는 작업별 통계와 별도로 슬롯 진입 지연을 기록한다. 슬롯이 제때 시작해도 뒤쪽 작업은 늦게 시작할 수 있으므로 하나의 통계로 합치지 않는다.

[05] dispatch()는 기록 공간을 먼저 확인한다. 공간이 부족한데 작업부터 실행하면 실행 횟수와 남아 있는 기록이 어긋나므로 이번 실험은 실행 전에 실패를 반환한다. 정상 경로에서는 호출 직전 시각을 저장하고, 작업이 반환한 직후 시각을 종료 값으로 기록한다.

시작 시각에서 예정 시각을 빼는 계산은 시작이 예정 이상이라는 조건에 기대고 있다. 스케줄러가 해당 슬롯까지 기다리고 시간이 뒤로 가지 않으므로 이 조건이 성립한다. 작업 실행 후에는 호출 횟수, 지터 합계, 최댓값을 갱신한다. 마지막의 next_ms 누적이 실행 종료 시각과 독립적인 주기 기준을 유지한다.

[06] 바깥 반복문은 0부터 90ms까지 열 개의 논리 슬롯을 방문한다. 슬롯에 도착한 뒤의 시각과 슬롯의 예정 시각을 비교해 지연을 센다. 안쪽 반복문은 작업 배열을 한 번 훑고 이번 슬롯에 해당하는 함수만 호출한다.

이 비교 방식에는 모든 주기가 슬롯의 정수 배수이고 첫 예정 시각이 슬롯 경계에 있다는 전제가 있다. 여기에 15ms 주기를 그대로 추가하면 next_ms가 15가 된 뒤 일치하는 슬롯을 만나지 못한다. 주기를 바꿀 때는 슬롯 크기도 다시 정하거나 별도의 예정 시각 기반 선택 구조로 바꿔야 한다.

[07] 출력은 시뮬레이션이 끝난 뒤 수행한다. 문자열 포인터는 수명이 프로그램 전체인 문자열 리터럴을 가리키므로 기록 배열에 보관해도 유효하다. 시간과 횟수는 unsigned이므로 printf()에는 %u를 사용한다. 호스트에서 출력이 느려도 이미 확정된 가상 시각에는 영향을 주지 않는다.

[08] main()은 통계를 초기화하고 한 번의 유한한 실험을 실행한다. 실패할 때는 오류 메시지와 실패 종료 상태를 반환한다. 현재 설정에서는 기록이 17개이므로 정상 경로로 끝난다. -pthread 옵션으로 컴파일하지만 스레드는 만들지 않으며, 작업들이 하나의 호출 흐름에서 실행된다는 점도 협동형 동작의 일부다.

실행 결과

macOS 또는 Linux에서 다음 명령을 사용한다. 아래 출력은 코드의 가상 시간 규칙에 따른 예상 결과다. 실제 경과 시간을 재지 않으므로 PC의 부하에 따라 숫자가 달라지지 않는다.

cc -std=c11 -Wall -Wextra -pthread scheduler.c -o scheduler
./scheduler

time unit: ms
task release start end jitter
sensor 0 0 2 0
control 0 2 5 2
report 0 5 9 5
sensor 10 10 12 0
sensor 20 20 22 0
control 20 22 25 2
sensor 30 30 32 0
sensor 40 40 42 0
control 40 42 45 2
sensor 50 50 52 0
report 50 52 66 2
sensor 60 66 68 6
control 60 68 71 8
sensor 70 71 73 1
sensor 80 80 82 0
control 80 82 85 2
sensor 90 90 92 0
summary: task runs max_jitter sum_jitter
sensor 10 6 7
control 5 8 16
report 2 5 7
slots late=2 max_late=6
finish=92 motor=on

50ms 슬롯의 센서는 52ms에 반환한다. 이어진 보고가 14ms를 소비해 현재 시각이 66ms가 된다. 따라서 60ms 예정 센서는 6ms 늦고, 그 센서 뒤의 제어는 8ms 늦다. 제어가 71ms에 반환하므로 다음 센서도 1ms 늦어진다.

보고 자신의 최대 시작 지터는 5ms다. 가장 오래 실행한 두 번째 보고의 시작 지터는 2ms에 불과하다. 긴 실행이 자기 시작 지터를 키우는 것이 아니라 뒤따르는 작업의 시작을 늦춘다는 차이가 기록에 드러난다.

센서의 지터 합계는 7ms이고 실행 횟수는 10회이므로 평균은 0.7ms다. 이 작은 평균만 보면 최대 6ms 지연이 충분히 드러나지 않는다. 평균은 전체 경향을 보는 값이고, 최댓값과 개별 기록은 특정 구간의 지연을 찾는 데 쓰인다.

모터는 40ms에 읽은 감지 결과를 제어가 반영하는 42ms에 꺼지고, 80ms에 읽은 해제 결과를 반영하는 82ms에 다시 켜진다. 최종 출력은 마지막 상태만 보여 준다. finish가 92인 이유는 마지막 센서가 끝난 즉시 실험을 종료하기 때문이다. 마지막에 100ms까지 기다리는 동작은 넣지 않았다.

실무에서 자주 틀리는 것

종료 시각에 주기를 더한다

다음은 dispatch() 내부의 주기 갱신을 잘못 바꾼 조각이다. 실행 시간이 다음 예정 시각에 포함되며, 이번 슬롯 비교 구조에서는 슬롯 경계를 벗어난 예정 시각 때문에 이후 호출까지 사라질 수 있다.

/* Wrong: derive the next release from completion. */
task->next_ms = hal_sim_now() + task->period_ms;

이전 예정 시각에 주기를 더한다. 이 계산은 작업이 늦었을 때도 기준 격자를 유지한다. 늦은 실행을 어떻게 처리할지는 이 계산과 별도로 정한다.

/* Correct: preserve the release sequence. */
task->next_ms += task->period_ms;

늦어진 시각을 슬롯 시작으로 덮어쓴다

가상 시계를 무조건 슬롯 시각으로 설정하면 긴 작업이 만든 지연을 지운다. 66ms에 끝난 보고 다음에 시계를 60ms로 되돌리는 셈이다. 그 결과 지터가 작게 보이고 시간 순서도 틀어진다.

/* Wrong: this can move time backward. */
hal_sim.now_ms = target_ms;

미래 시각으로 기다릴 때만 시계를 갱신한다. 늦었으면 현재 값을 유지하고 바로 처리한다. 실제 시간원으로 바꾸더라도 이 대기의 의미는 같아야 한다.

/* Correct: wait only when the target is in the future. */
if (hal_sim.now_ms < target_ms) {
    hal_sim.now_ms = target_ms;
}

반환하지 않는 반복문을 작업 안에 넣는다

작업 함수 안에 독립적인 무한 반복문을 넣으면 스케줄러가 다음 함수를 호출하지 못한다. 다음 조각으로 센서 함수를 바꾸면 첫 센서 호출에서 제어와 보고가 멈춘다.

/* Wrong: the scheduler never regains control. */
static void sensor_step(unsigned release_ms)
{
    (void)release_ms;
    for (;;) {
        hal_sim.sensor_blocked = hal_sim_read_sensor();
        hal_sim_consume(2u);
    }
}

한 번 읽고 반환한다. 반복 호출은 스케줄러가 담당한다. 긴 처리 과정이 필요하면 여러 번의 짧은 호출 사이에 진행 상태를 보존한다.

/* Correct: perform one bounded step and return. */
static void sensor_step(unsigned release_ms)
{
    (void)release_ms;
    hal_sim.sensor_blocked = hal_sim_read_sensor();
    hal_sim_consume(2u);
}

종료 지연을 시작 지터라고 기록한다

종료 시각에서 예정 시각을 빼면 대기 지연과 실행 시간이 더해진다. 이 값도 목적에 따라 유용하지만 시작 지터라는 이름을 붙이면 원인 해석이 달라진다.

/* Wrong for a start-jitter measurement. */
unsigned jitter_ms = entry->end_ms - entry->release_ms;

시작 지터와 실행 시간을 각각 계산한다. 이 둘을 나란히 보아야 앞 작업의 영향과 자기 실행 시간의 영향을 구분할 수 있다.

/* Correct: calculate and label the two quantities separately. */
unsigned jitter_ms = entry->start_ms - entry->release_ms;
unsigned execution_ms = entry->end_ms - entry->start_ms;
printf("start_jitter=%u execution=%u\n", jitter_ms, execution_ms);

한눈에 보기

협동형 주기 실행을 읽고 점검하는 기준

항목	이번 구현의 기준	확인할 점
슬롯	10ms마다 실행 대상 선택	작업을 중간에 끊는 경계가 아니다.
주기	10, 20, 50ms	모두 슬롯 크기의 정수 배수다.
예정 시각	이전 예정 시각에 주기 누적	종료 시각을 기준으로 다시 잡지 않는다.
실행 순서	센서, 제어, 보고	같은 슬롯에서도 시작 시각은 다르다.
늦은 슬롯	빠뜨리지 않고 순서대로 실행	늦게 읽는 입력은 현재 입력이다.
시작 지터	시작 시각 − 예정 시각	실행 시간과 별도로 기록한다.
시뮬레이션 시간	작업이 선언한 시간만 증가	보드에서 측정한 실제 실행 시간이 아니다.
작업의 책임	짧게 처리하고 반환	다음 작업의 시작 가능 시점을 결정한다.

연습 문제

1. report_step()을 모든 호출에서 4ms만 소비하도록 바꾼다. 센서와 제어의 최대 시작 지터, 늦은 슬롯 수, 마지막 작업의 종료 시각을 구한다.
2. 원래 코드에서 50ms 예정 보고가 14ms 대신 24ms를 소비하도록 바꾼다. 60ms와 70ms 예정 센서의 시작 시각을 구하고, 두 실행이 모두 실제 60ms와 70ms의 입력을 읽는지 설명한다.
3. 원래 코드에 주기가 15ms인 작업을 추가하려 한다. 현재의 next_ms 비교가 어떤 문제를 만드는지 설명하고, 기존 주기를 유지하면서 15ms도 표현할 수 있는 슬롯 크기 하나를 제시한다.
4. 원래 보고 작업의 next_ms 초기값만 0에서 10으로 옮긴다. 나머지 코드가 같을 때 보고의 예정 시각 두 개와 실행 횟수, 늦은 슬롯 수를 구한다. 14ms 실행이 남아 있는지도 설명한다.

정답과 해설

1. 센서의 최대 시작 지터는 0ms이고 제어는 2ms다. 50ms 슬롯에서 센서가 52ms에 끝나고 보고가 56ms에 끝나므로 다음 슬롯을 침범하지 않는다. 늦은 슬롯은 0개이고 마지막 센서는 여전히 90ms에 시작해 92ms에 끝난다. 제어의 2ms는 같은 슬롯에서 센서를 먼저 실행한 결과다.
2. 보고는 52ms에 시작해 76ms에 끝난다. 60ms 예정 센서는 76ms에 시작해 78ms에 끝나고, 제어가 81ms까지 실행된다. 따라서 70ms 예정 센서는 81ms에 시작한다. 두 센서는 각각 76ms와 81ms의 입력을 읽는다. 특히 두 번째 읽기에서는 80ms에 감지가 해제된 뒤이므로 거짓을 얻는다. 예정 시각을 보존해도 과거 센서 값이 보존되는 것은 아니다.
3. 첫 실행 뒤 next_ms가 15ms가 되지만 스케줄러는 10ms 단위의 슬롯만 방문한다. 일치하는 슬롯이 없어 이후 실행이 누락된다. 슬롯을 5ms로 바꾸면 10, 15, 20, 50ms를 모두 정수 개의 슬롯으로 표현할 수 있다. 슬롯을 줄였다고 작업의 실행 시간이 줄거나 긴 작업이 중단되는 것은 아니다.
4. 보고의 예정 시각은 10ms와 60ms이며 실행 횟수는 2회다. 코드의 긴 실행 조건은 예정 시각이 50ms인지 검사하므로 두 호출 모두 4ms를 소비한다. 10ms 슬롯은 16ms에 끝나고, 60ms 슬롯은 센서와 제어와 보고를 마친 69ms에 끝난다. 늦은 슬롯은 0개다. 두 번째 호출에 항상 긴 시간을 주려면 예정 시각 비교 대신 호출 횟수 등으로 실험 조건을 표현해야 한다.