RTOS 동기화 - 세마포어·뮤텍스·큐

이 장에서 배우는 것

앞 장에서 태스크의 우선순위와 실행 순서를 살펴보았다. 이제 컨베이어의 센서 수집과 모터 제어를 서로 다른 태스크에 맡긴다. 센서가 물체를 감지하면 제어 태스크로 측정값을 전달해야 하고, 두 태스크가 같은 통신 장치를 사용한다면 접근 순서도 정해야 한다. 실행할 일을 나누는 것만으로는 이러한 관계가 정리되지 않는다.

이 장에서는 세마포어(semaphore), 뮤텍스(mutex), 큐(queue)가 어떤 관계를 표현하는지 구분한다. 이어서 낮은 우선순위의 태스크가 공유 장치를 잡았을 때 생기는 우선순위 역전(priority inversion)을 살펴보고, 우선순위 상속(priority inheritance)이 실행 순서를 어떻게 바꾸는지 확인한다. 마지막에는 이벤트 통지와 측정값 전달을 연결한 프로그램을 실행한다.

• 이벤트의 개수, 자원의 소유권, 데이터의 전달에 맞는 동기화 수단을 선택한다.
• 우선순위 역전이 생기는 실행 순서를 읽고 상속의 효과와 한계를 설명한다.
• 용량이 제한된 큐에서 생산자와 소비자가 기다리는 조건을 구현한다.
• PC 시뮬레이터의 동작을 실제 FreeRTOS API와 대응시킨다.

문제 상황

컨베이어 제어기에는 센서 보정값을 읽는 유지보수 태스크 L, 생산량을 집계하는 태스크 M, 모터 설정을 갱신하는 제어 태스크 H가 있다. 우선순위는 H, M, L 순서다. L과 H는 같은 SPI 장치를 사용하며, 장치 설정과 전송이 서로 섞이지 않도록 하나의 잠금으로 보호한다. M은 이 장치를 사용하지 않는다.

L이 잠금을 얻은 다음 H가 실행되었다고 하자. H는 장치가 비워질 때까지 기다려야 한다. 이때 M이 실행 가능해지면 스케줄러는 L보다 우선순위가 높은 M을 선택한다. H가 기다리는 직접적인 이유는 L의 잠금이지만, 실제로 잠금 해제를 늦추는 태스크는 장치와 관계없는 M이다.

다른 문제도 있다. 센서가 연속으로 물체를 감지할 때 제어 태스크가 잠시 바쁘면, 공유 변수 하나에 저장한 측정값은 다음 값으로 덮인다. 플래그를 하나 더 두더라도 두 번의 감지를 한 번으로 합칠 수 있다. 잠금 문제와 전달 문제를 하나의 수단으로 해결하려 하면 소유권, 이벤트 개수, 데이터 보존 중 하나를 놓치기 쉽다.

예제는 이 두 상황을 분리해서 관찰한다. 먼저 같은 작업을 상속 없이 실행하고 상속을 켜서 다시 실행한다. 그다음 센서 이벤트를 받은 생산자가 두 칸짜리 큐에 값을 넣고, 늦게 시작한 제어 태스크가 값을 순서대로 꺼내도록 한다. 숫자는 센서 측정 패킷의 순번을 대신한다.

세 가지 동기화 수단의 역할

세마포어는 사용할 수 있는 허가나 발생한 이벤트의 개수를 나타낸다. 카운팅 세마포어는 여러 개를 누적하고, 이진 세마포어는 상태를 0 또는 1로 제한한다. 이벤트를 준 실행 문맥과 받은 태스크가 같을 필요는 없다. 반면 세마포어에 개수를 쌓았다고 해서 각 이벤트의 측정값까지 보관되는 것은 아니다.

뮤텍스는 공유 자원을 누가 사용 중인지 나타낸다. 획득한 태스크가 소유자가 되고, 사용을 마친 소유자가 해제한다. 이 소유권이 있어야 기다리는 태스크와 잠금을 풀어야 할 태스크를 연결할 수 있다. 우선순위 상속도 이 관계를 바탕으로 동작한다. 이진 세마포어와 뮤텍스를 둘 다 값 하나로 표현할 수 있다고 해서 의미까지 같지는 않다.

큐는 항목을 정해진 크기로 보관하고 순서대로 전달한다. 수신 태스크는 큐가 비었을 때 기다릴 수 있고, 송신 태스크는 큐가 가득 찼을 때 기다릴 수 있다. 값의 복사와 대기 상태 변경을 큐 연산이 함께 처리하므로, 별도 플래그와 배열의 상태를 서로 맞추는 부담이 줄어든다.

전달하려는 의미에 따라 동기화 수단을 고른다

수단	보관하는 정보	컨베이어에서의 용도	주의할 점
카운팅 세마포어	허가 또는 이벤트 개수	처리할 감지 횟수 통지	측정 데이터는 별도로 필요하다
이진 세마포어	통지가 있는지 여부	장치 작업 완료 통지	연속 통지가 합쳐질 수 있다
뮤텍스	잠금 상태와 소유자	공유 SPI 장치 보호	소유자가 해제해야 한다
큐	순서가 있는 데이터 항목	측정 패킷 전달	가득 찼을 때의 정책이 필요하다

어떤 태스크를 다시 실행 가능하게 만드는 것과 그 태스크가 자원을 얻는 것은 구분해야 한다. 예제의 세마포어 통지와 잠금 해제는 대기 태스크를 준비 상태로 바꿀 뿐이다. 실제 획득은 해당 태스크가 다시 선택되어 연산을 재시도할 때 이루어진다. 따라서 깨어난 태스크도 조건을 다시 확인한다.

실제 RTOS의 대기 가능한 API는 이 과정을 호출 내부에서 처리한다. 반면 이번 협동형 시뮬레이터는 C 함수의 호출 지점을 보존하지 않는다. 대기 조건을 발견하면 태스크 함수가 반환하고, 다음 실행에서 같은 연산을 다시 시도한다. 이어서 해야 할 일은 구조체의 단계나 보류 여부로 기억한다.

우선순위 역전과 상속

우선순위가 높아도 이미 다른 태스크가 소유한 자원을 사용할 수는 없다. 따라서 H가 L의 짧은 작업을 기다리는 것 자체는 공유 자원이 만드는 대기다. 문제는 M이 끼어들어 L의 진행을 늦추는 데 있다. M이 계속 실행 가능하다면 H의 대기는 L이 자원을 사용하는 데 필요한 시간보다 훨씬 길어질 수 있다.

우선순위 상속을 적용하면 H가 잠금을 기다리는 동안 L의 유효 우선순위를 H 수준으로 올린다. L이 M보다 먼저 실행되어 잠금을 해제하도록 하는 것이다. 기본 우선순위는 그대로 두고 스케줄러가 선택에 쓰는 유효 우선순위만 변경한다. 예제에서는 관련 잠금이 하나뿐이므로 해제 시 기본 우선순위로 되돌린다.

그림의 각 칸은 태스크 함수를 한 번 호출하는 시뮬레이션 단위다. H가 기다리기로 결정한 호출도 한 칸을 차지한다. 실제 RTOS에서는 H가 차단되자마자 같은 타이머 틱 안에서 다른 태스크가 실행될 수 있으므로, 그림의 칸 수를 실제 응답 시간으로 읽으면 안 된다. 여기서는 선택 순서의 차이를 관찰한다.

상속이 L의 장치 작업을 대신 끝내 주는 것은 아니다. L이 잠금을 가진 채 다른 이벤트를 기다리거나 오래 걸리는 처리를 수행하면 H도 그 영향을 받는다. 잠금 구간에는 공유 자원의 일관성을 지키는 데 필요한 작업만 넣고, 큐에서 다음 작업을 기다리는 동작은 보통 잠금 밖에 둔다.

중첩 잠금까지 들어가면 상속 관계는 더 복잡해진다. 한 태스크가 여러 뮤텍스를 보유하거나, 소유자 자신이 다른 소유자를 기다릴 수 있다. 이번 코드는 단일 뮤텍스와 단일 경쟁 관계만 구현한다. 여러 잠금의 해제 순서와 우선순위 복원 규칙은 사용하는 RTOS의 구현을 확인해야 한다. FreeRTOS의 사실 관계는 뮤텍스 공식 설명에서 확인할 수 있다.

큐로 태스크를 연결하기

두 번째 실험에서 시뮬레이션 계층인 hal_sim은 세 번의 센서 이벤트를 만든다. 생산자는 이벤트 하나를 받아 순번 하나를 생성한다. 큐에는 두 항목만 들어가므로 소비자가 시작하기 전에 세 번째 항목을 넣으려 하면 기다려야 한다. 소비자가 한 항목을 꺼내면 공간이 생기고 생산자가 다시 실행 가능해진다.

여기서 보류 중인 값이 중요하다. 생산자는 이미 세 번째 이벤트를 소비하고 값 3을 만들었다. 큐가 가득 찼다고 태스크 함수가 반환한 뒤, 다음 실행에서 이벤트를 또 받으면 값과 이벤트의 대응이 어긋난다. 따라서 pending이 참이면 이벤트 수신을 건너뛰고 보관한 값을 다시 전송한다.

큐의 용량은 처리 속도 차이를 잠시 흡수하는 공간이다. 생산 속도가 소비 속도보다 계속 빠르면 유한한 큐는 결국 찬다. 그때 기다릴지, 새 데이터를 버릴지, 오래된 데이터를 대체할지는 데이터의 의미에 따라 정한다. 이 예제는 모든 순번을 전달하기 위해 생산자가 기다리는 정책을 사용한다.

다만 소프트웨어 생산자를 멈추어도 실제 컨베이어나 센서 이벤트까지 멈추는 것은 아니다. 대기 중에도 입력이 계속 들어온다면 이벤트 계수와 측정 저장 공간에 별도 한계가 있다. 예제는 입력을 세 번으로 제한하고 범위 위반을 검사용 조건으로 잡는다. 실제 장치에서는 포화가 발생했음을 기록하거나 상위 제어에 알리는 정책이 필요하다.

시뮬레이터의 연산과 FreeRTOS API의 대응

예제 연산	FreeRTOS 대응	주요 차이
Event의 count와 limit 설정	xSemaphoreCreateCountingStatic()	실제 API에는 관리용 저장 공간을 제공한다
event_take(), event_give()	xSemaphoreTake(), xSemaphoreGive()	대기 시간과 반환 결과를 지정하고 확인한다
Mutex의 owner 초기화	xSemaphoreCreateMutexStatic()	커널이 소유권과 상속 상태를 관리한다
mutex_take(), mutex_give()	xSemaphoreTake(), xSemaphoreGive()	뮤텍스는 태스크 문맥에서 사용한다
Queue의 배열과 인덱스	xQueueCreateStatic()	항목 크기와 항목 수를 명시한다
queue_put(), queue_get()	xQueueSend(), xQueueReceive()	성공하거나 대기 조건이 끝날 때 호출이 반환한다
hal_sim_tick()의 이벤트 통지	xSemaphoreGiveFromISR()	실제 인터럽트에서는 전용 API와 포트의 전환 요청 규칙을 사용한다

표의 정적 생성 API를 사용하려면 해당 FreeRTOS 설정과 저장 공간이 준비되어야 한다. 또한 대기 시간 0은 즉시 결과를 확인하는 요청이다. 이 예제의 대기는 해제 조건이 생길 때까지 유지되므로 실제 API에 옮길 때는 허용할 대기 시간을 따로 결정해야 한다. API의 세부 계약은 큐 공식 참조와 세마포어와 뮤텍스 공식 참조에서 확인할 수 있다.

완성 코드

다음 전체 프로그램을 rtos_sync.c로 저장한다. 숫자가 클수록 우선순위가 높다. 한 번 선택된 태스크는 함수가 반환할 때까지 실행되며, 준비된 태스크 중 유효 우선순위가 가장 높은 태스크를 다음에 선택한다. 별도 스레드를 만들지 않으므로 호스트 운영체제의 스케줄링에 따라 출력 순서가 달라지지 않는다.

뮤텍스 실험에는 세 태스크를, 큐 실험에는 두 태스크를 등록한다. 모든 동기화 연산이 같은 실행 흐름에서 수행된다는 것이 이 시뮬레이터의 전제다. 코드를 그대로 여러 pthread에서 호출하면 공유 상태를 보호할 수 없다. 컴파일 명령의 -pthread는 지정된 빌드 조건을 맞추기 위한 옵션이며, 이 코드를 병렬 실행에 안전하게 바꾸는 옵션은 아니다.

#include <assert.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>

/* [1] Task state and synchronization objects. */
enum {
    TASK_CAP = 3,
    QUEUE_CAP = 2,
    EVENT_LIMIT = 3,
    STEP_LIMIT = 20
};

typedef enum {
    WAIT_NONE,
    WAIT_EVENT,
    WAIT_MUTEX,
    WAIT_READ,
    WAIT_WRITE
} Wait;

typedef struct {
    int base;
    int effective;
    unsigned release;
    unsigned phase;
    int remaining;
    Wait wait;
    bool done;
    void (*step)(int id);
} Task;

typedef struct {
    unsigned count;
    unsigned limit;
} Event;

typedef struct {
    int owner;
    bool inherit;
} Mutex;

typedef struct {
    int data[QUEUE_CAP];
    unsigned head;
    unsigned count;
} Queue;

static Task tasks[TASK_CAP];
static int task_count;
static unsigned tick;
static bool pipeline;
static Event event;
static Mutex bus;
static Queue queue;
static bool pending;
static int next_value;

/* [2] Wakeup changes readiness, not ownership. */
static void wake(Wait reason)
{
    for (int i = 0; i < task_count; ++i) {
        if (!tasks[i].done && tasks[i].wait == reason) {
            tasks[i].wait = WAIT_NONE;
        }
    }
}

static bool event_take(int id)
{
    if (event.count == 0U) {
        tasks[id].wait = WAIT_EVENT;
        return false;
    }
    --event.count;
    return true;
}

static void event_give(void)
{
    assert(event.count < event.limit);
    ++event.count;
    wake(WAIT_EVENT);
}

/* [3] One mutex; recursive locking is not supported. */
static bool mutex_take(int id)
{
    if (bus.owner == -1) {
        bus.owner = id;
        return true;
    }

    assert(bus.owner != id);
    tasks[id].wait = WAIT_MUTEX;
    if (bus.inherit &&
        tasks[bus.owner].effective < tasks[id].effective) {
        tasks[bus.owner].effective = tasks[id].effective;
    }
    return false;
}

static void mutex_give(int id)
{
    assert(bus.owner == id);
    bus.owner = -1;
    tasks[id].effective = tasks[id].base;
    wake(WAIT_MUTEX);
}

/* [4] A bounded FIFO that copies integer values. */
static bool queue_put(int id, int value)
{
    if (queue.count == QUEUE_CAP) {
        tasks[id].wait = WAIT_WRITE;
        return false;
    }

    unsigned tail = (queue.head + queue.count) % QUEUE_CAP;
    queue.data[tail] = value;
    ++queue.count;
    wake(WAIT_READ);
    return true;
}

static bool queue_get(int id, int *value)
{
    if (queue.count == 0U) {
        tasks[id].wait = WAIT_READ;
        return false;
    }

    *value = queue.data[queue.head];
    queue.head = (queue.head + 1U) % QUEUE_CAP;
    --queue.count;
    wake(WAIT_WRITE);
    return true;
}

/* [5] Inversion experiment: low, medium, high. */
static void low_step(int id)
{
    Task *self = &tasks[id];

    if (self->phase == 0U) {
        if (!mutex_take(id)) {
            return;
        }
        printf("[%u] L lock\n", tick);
        self->phase = 1U;
        return;
    }

    printf("[%u] L work %d\n", tick, self->remaining);
    --self->remaining;
    if (self->remaining == 0) {
        mutex_give(id);
        printf("[%u] L unlock\n", tick);
        self->done = true;
    }
}

static void medium_step(int id)
{
    printf("[%u] M work %d\n", tick, tasks[id].remaining);
    --tasks[id].remaining;
    if (tasks[id].remaining == 0) {
        tasks[id].done = true;
    }
}

static void high_step(int id)
{
    if (!mutex_take(id)) {
        printf("[%u] H wait, L priority=%d\n",
               tick, tasks[bus.owner].effective);
        return;
    }

    printf("[%u] H acquired\n", tick);
    mutex_give(id);
    tasks[id].done = true;
}

/* [6] Pipeline: retain an item while the queue is full. */
static void producer_step(int id)
{
    if (!pending) {
        if (!event_take(id)) {
            printf("[%u] sensor wait event\n", tick);
            return;
        }
        ++next_value;
        pending = true;
    }

    if (!queue_put(id, next_value)) {
        printf("[%u] sensor wait space, keep %d\n",
               tick, next_value);
        return;
    }

    printf("[%u] sensor send %d\n", tick, next_value);
    pending = false;
    if (next_value == EVENT_LIMIT) {
        tasks[id].done = true;
    }
}

static void consumer_step(int id)
{
    int value;

    if (!queue_get(id, &value)) {
        printf("[%u] control wait data\n", tick);
        return;
    }

    printf("[%u] control receive %d\n", tick, value);
    if (value == EVENT_LIMIT) {
        tasks[id].done = true;
    }
}

/* [7] hal_sim and a cooperative scheduling loop. */
static void hal_sim_tick(void)
{
    if (pipeline && tick >= 1U && tick <= EVENT_LIMIT) {
        event_give();
    }
}

static bool all_done(void)
{
    for (int i = 0; i < task_count; ++i) {
        if (!tasks[i].done) {
            return false;
        }
    }
    return true;
}

static void run(void)
{
    for (tick = 0U; tick < STEP_LIMIT; ++tick) {
        if (all_done()) {
            break;
        }
        hal_sim_tick();

        int best = -1;
        for (int i = 0; i < task_count; ++i) {
            if (tasks[i].done || tasks[i].wait != WAIT_NONE ||
                tasks[i].release > tick) {
                continue;
            }
            if (best == -1 ||
                tasks[i].effective > tasks[best].effective) {
                best = i;
            }
        }

        if (best == -1) {
            printf("[%u] idle\n", tick);
        } else {
            tasks[best].step(best);
        }
    }
    assert(all_done());
}

/* [8] Reset and register the tasks for each experiment. */
static void reset(bool use_pipeline, bool inherit)
{
    memset(tasks, 0, sizeof tasks);
    task_count = 0;
    pipeline = use_pipeline;
    event = (Event){ .count = 0U, .limit = EVENT_LIMIT };
    bus = (Mutex){ .owner = -1, .inherit = inherit };
    queue = (Queue){ .head = 0U, .count = 0U };
    pending = false;
    next_value = 0;
}

static void add_task(int priority, unsigned release,
                     int remaining, void (*step)(int))
{
    assert(task_count < TASK_CAP);
    tasks[task_count] = (Task){
        .base = priority,
        .effective = priority,
        .release = release,
        .phase = 0U,
        .remaining = remaining,
        .wait = WAIT_NONE,
        .done = false,
        .step = step
    };
    ++task_count;
}

static void run_inversion(bool inherit)
{
    reset(false, inherit);
    puts(inherit ? "== inheritance on ==" :
                   "== inheritance off ==");
    add_task(1, 0U, 2, low_step);
    add_task(2, 2U, 2, medium_step);
    add_task(3, 1U, 0, high_step);
    run();
}

int main(void)
{
    run_inversion(false);
    run_inversion(true);

    reset(true, false);
    puts("== queue pipeline ==");
    add_task(2, 0U, 0, producer_step);
    add_task(1, 4U, 0, consumer_step);
    run();

    assert(event.count == 0U);
    assert(queue.count == 0U);
    assert(!pending);
    puts("all checks passed");
    return 0;
}

줄별 해설

주석의 [1]에서 Task의 base는 설정한 우선순위이고 effective는 선택에 사용하는 우선순위다. release는 최초로 실행할 수 있는 시점이다. wait가 WAIT_NONE이면서 release를 지난 태스크만 준비 상태로 취급한다. done은 실험용 작업이 끝났음을 나타낸다. 실제 제어기의 상시 태스크와 달리 이번 작업은 유한하게 종료한다.

Event는 count와 limit만 가진다. 어떤 태스크가 이벤트를 주었는지는 기록하지 않는다. Mutex는 owner에 태스크 배열의 인덱스를 저장하고, -1을 비어 있는 상태로 쓴다. Queue는 head와 count로 유효한 항목 범위를 표현한다. 세 자료형의 필드 차이가 각 동기화 수단이 보존하는 의미의 차이다.

[2]의 wake는 대기 이유가 일치하는 태스크를 준비 상태로 바꾼다. 이 프로그램에는 종류별 동기화 객체가 하나씩만 있으므로 대기 이유만 비교해도 된다. 객체를 두 개 이상 만들면 태스크가 어느 객체를 기다리는지도 저장해야 한다. 그렇지 않으면 다른 큐에서 기다리던 태스크까지 잘못 깨우게 된다.

event_take는 count가 0일 때 대기 이유를 기록하고 false를 반환한다. 호출자는 그 반환값을 보고 이번 실행을 끝낸다. event_give는 먼저 개수를 늘린 다음 대기자를 깨운다. assert는 이번 입력 범위에서 포화가 없어야 한다는 검사용 조건이다. 제품의 포화 처리 정책을 대신하는 코드는 아니다.

[3]의 mutex_take는 소유자가 없을 때만 owner를 바꾼다. 이미 소유자가 있으면 호출자를 대기 상태로 만들고, 상속이 켜진 경우 소유자의 유효 우선순위를 올린다. mutex_give는 소유자가 맞는지 확인한 다음 잠금을 비우고 우선순위를 복원한다. 재귀 획득, 시간 제한, 여러 잠금에 걸친 상속은 구현 범위에 포함하지 않는다.

[4]에서 삽입 위치는 head와 count의 합을 용량으로 나눈 나머지다. 꺼낼 때는 head를 한 칸 옮기고 count를 줄인다. 삽입 성공은 데이터 대기자를 깨우고, 수신 성공은 공간 대기자를 깨운다. 가득 차거나 비어 있으면 인덱스를 바꾸지 않으므로 실패한 연산이 큐의 내용을 손상시키지 않는다.

[5]의 L은 처음 선택되었을 때 잠금만 얻고 반환한다. 이후 두 번 선택되어 작업하고 마지막 호출에서 해제한다. M도 두 번의 작업이 필요하다. H는 잠금을 얻지 못하면 반환하고, 다시 선택되었을 때 획득을 재시도한다. 출력의 L work 숫자는 남은 작업 횟수이며 시간 단위가 아니다.

[6]의 pending은 이벤트를 이미 소비했지만 큐에 아직 전달하지 못한 값이 있음을 뜻한다. true인 동안 next_value를 증가시키지 않는 것이 핵심이다. 소비자는 성공적으로 꺼낸 경우에만 value를 읽는다. 큐가 비었을 때 value는 초기화되지 않지만, 실패 경로에서 바로 반환하므로 그 값을 사용하지 않는다.

[7]에서는 hal_sim_tick이 먼저 이벤트를 만들고 준비 태스크를 선택한다. 이는 인터럽트 중첩을 재현하지 않고 이벤트 도착 시점만 제어하는 장치다. 유효 우선순위가 같으면 먼저 등록한 태스크를 고른다. 같은 우선순위 사이의 순환 실행은 구현하지 않았으며, 이번 세 실험에서는 그 차이가 결과에 영향을 주지 않는다.

[8]의 reset은 실험 사이에 남은 상태를 지운다. 각 태스크는 add_task에서 모든 의미 있는 필드를 다시 초기화한다. main의 마지막 조건은 이벤트, 큐 항목, 보류 값이 남지 않았는지 확인한다. run의 반복 횟수 제한은 시뮬레이터가 끝없이 도는 것을 막는 검사이며 실제 태스크의 대기 시간 제한이 아니다.

실행 결과

macOS 또는 Linux에서 다음 명령으로 빌드하고 실행한다. 표준 C11 헤더와 기능만 사용하며 아래 명령의 경고 옵션을 기준으로 작성했다. 예상 출력은 다음과 같다.

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

== inheritance off ==
[0] L lock
[1] H wait, L priority=1
[2] M work 2
[3] M work 1
[4] L work 2
[5] L work 1
[5] L unlock
[6] H acquired
== inheritance on ==
[0] L lock
[1] H wait, L priority=3
[2] L work 2
[3] L work 1
[3] L unlock
[4] H acquired
[5] M work 2
[6] M work 1
== queue pipeline ==
[0] sensor wait event
[1] sensor send 1
[2] sensor send 2
[3] sensor wait space, keep 3
[4] control receive 1
[5] sensor send 3
[6] control receive 2
[7] control receive 3
all checks passed

상속이 없으면 H는 시점 1에 기다리기 시작해 시점 6에 획득한다. 상속이 있으면 L이 M보다 먼저 실행되어 H의 획득이 시점 4로 앞당겨진다. M의 작업이 사라진 것은 아니다. 잠금을 기다리는 H의 진행을 위해 실행 순서가 달라졌을 뿐이다.

큐 실험의 시점 3에서는 이벤트 개수가 이미 감소했고 값 3이 보류 상태다. 시점 4에서 소비자가 값 1을 꺼내자 공간이 생긴다. 시점 5에서 생산자는 추가 이벤트 없이 값 3을 넣는다. 마지막 수신 순서가 1, 2, 3인 것은 빈 공간이 생겨도 기존 항목의 순서를 유지하기 때문이다.

마지막 문구는 코드에 넣은 조건들이 통과했음을 뜻한다. 다양한 입력과 병렬 실행까지 검증했다는 뜻은 아니다. 이 실험에서 확인한 것은 제한된 모델 안의 실행 순서와 데이터 보존이다. 실제 시간 단위의 지연과 마감 시간 충족 여부는 다음 장에서 별도로 분석한다.

실무에서 자주 틀리는 것

다음 코드는 잘못된 부분과 수정 부분을 비교하는 태스크 내부 발췌다. 독립 실행 프로그램은 아니며, 이름과 연산은 위 완성 코드를 기준으로 한다.

대기 연산을 반복문으로 감싼다

잠금을 얻을 때까지 호출을 반복하면 기다리는 동안 다른 태스크가 실행될 것이라고 생각하기 쉽다. 하지만 협동형 실행에서는 함수가 반환해야 소유자가 실행된다. 아래 반복문은 대기 상태를 기록하고도 CPU를 내놓지 않는다.

/* Wrong: the owner cannot run. */
while (!mutex_take(id)) {
}

획득하지 못하면 즉시 반환한다. 실제 RTOS에서는 대기 가능한 API를 호출하고 반환 결과를 검사한다. 대기 시간 0의 호출을 높은 우선순위에서 계속 반복하면 소유자의 진행을 방해할 수 있으므로 같은 문제를 다시 만들지 않아야 한다.

/* Correct: let the scheduler run another task. */
if (!mutex_take(id)) {
    return;
}
/* Use the protected device. */
mutex_give(id);

큐 전송에 실패한 항목을 잊는다

이벤트를 받은 뒤 곧바로 순번을 증가시키고 전송하면 코드가 짧아진다. 그러나 전송 실패 후 다음 실행에서 이벤트를 다시 기다리면 이미 받은 이벤트에 해당하는 값은 보관되지 않는다. 세 번째 이벤트가 마지막이라면 더 이상 진행할 계기도 없다.

/* Wrong: the received event has no retained item. */
if (!event_take(id)) {
    return;
}
if (!queue_put(id, ++next_value)) {
    return;
}

이벤트 수신과 데이터 전달 사이에 보류 상태를 둔다. 같은 값의 전송이 성공한 뒤에만 보류 상태를 지운다. 실제 큐 API에서도 시간 제한으로 전송에 실패했다면 재시도할지 버릴지 호출자가 결정해야 한다.

/* Correct: retry the same item. */
if (!pending) {
    if (!event_take(id)) {
        return;
    }
    ++next_value;
    pending = true;
}
if (!queue_put(id, next_value)) {
    return;
}
pending = false;

획득 결과를 무시하고 자원을 사용한다

잠금 함수를 호출했다는 사실과 잠금을 얻었다는 사실은 다르다. 아래 코드는 획득에 실패해도 장치를 사용하고, 다른 태스크의 잠금을 해제하려 한다. 예제에서는 소유권 검사가 이를 잡지만 실제 장치 접근은 그 전에 이미 잘못 수행될 수 있다.

/* Wrong: failure is ignored. */
(void)mutex_take(id);
/* Access the shared device here. */
mutex_give(id);

성공한 경로에서만 자원에 접근하고 해제한다. 오류 처리나 조기 반환을 추가할 때도 소유권이 남는 경로와 이중 해제 경로가 생기지 않도록 확인한다.

/* Correct: ownership gates both access and release. */
if (!mutex_take(id)) {
    return;
}
/* Access the shared device here. */
mutex_give(id);

상속이 모든 지연을 해결한다고 생각한다

큐가 비었을 때 값을 기다리는 동안에는 공유 SPI 장치가 필요하지 않다. 그런데 먼저 장치를 잠그면 생산자가 그 장치를 사용해야 값을 만들 수 있는 구조에서 서로의 진행을 막을 수 있다. 상속은 필요한 데이터나 큐의 빈 공간을 만들어 주지 않는다.

/* Wrong: an empty queue leaves the mutex held. */
if (!mutex_take(id)) {
    return;
}
int value;
if (!queue_get(id, &value)) {
    return;
}
/* Apply value to the device. */
mutex_give(id);

먼저 값을 받고 보관한 뒤 장치를 획득한다. 다음 수정은 소비자 하나가 실행하는 태스크 내부 코드다. 값을 받은 뒤 잠금 대기가 생길 수 있으므로 여기에서도 보류 상태가 필요하다. 여러 소비자가 있다면 이 상태를 각 태스크의 자료구조에 둔다.

/* Correct: retain data, then acquire the device. */
static bool have_value;
static int saved_value;

if (!have_value) {
    if (!queue_get(id, &saved_value)) {
        return;
    }
    have_value = true;
}
if (!mutex_take(id)) {
    return;
}
/* Apply saved_value to the device. */
mutex_give(id);
have_value = false;

한눈에 보기

동기화 설계에서 확인할 조건

설계 지점	확인할 조건	예제의 선택
이벤트 통지	개수 보존이 필요한가	최대 세 개의 카운팅 세마포어
공유 장치 접근	누가 획득하고 해제하는가	소유자만 해제하는 뮤텍스
우선순위 역전	관계없는 태스크가 소유자의 진행을 늦추는가	소유자의 유효 우선순위 상속
큐 포화	기다릴지 버릴지 정했는가	생산자가 값을 보류하고 대기
깨어난 태스크	조건을 다시 확인하는가	획득 또는 전송을 재시도
모델의 범위	실제 실행 시간과 병렬성을 재현하는가	단일 실행 흐름에서 순서만 관찰

동기화 수단을 고를 때는 무엇을 기다리는지부터 문장으로 써 보면 도움이 된다. 감지 횟수를 기다린다면 세마포어, 장치 소유권을 기다린다면 뮤텍스, 다음 측정값을 기다린다면 큐다. 이어서 기다리는 동안 이미 확보한 자원과 데이터가 무엇인지 확인한다. 그 정보가 해제 위치와 보류 상태의 필요성을 결정한다.

연습 문제

1. 상속을 끈 실험에서 M의 remaining을 2에서 4로 바꾸었다. L의 잠금 해제와 H의 획득은 각각 어느 시점에 일어나는가. 상속을 켠 경우와 비교하라.
2. QUEUE_CAP을 2에서 1로 바꾸고 나머지는 그대로 둔다. 큐 실험에서 생산자가 공간을 기다리는 시점과 소비자가 값을 받는 시점을 예측하라.
3. 센서 이벤트를 이진 세마포어로 바꾸어도 세 물체의 감지를 항상 구분할 수 있는가. 생산자가 실행되지 못하는 동안 세 번 감지되는 경우를 설명하라.
4. 측정 항목을 구조체로 확장하고 그 안에 작업용 버퍼를 가리키는 포인터를 넣었다. 큐에 구조체를 복사한 직후 생산자가 버퍼를 덮어써도 되는가. 안전한 전달 방법 두 가지를 제시하라.

정답과 해설

1. 상속이 없으면 M이 시점 2, 3, 4, 5를 사용한다. L은 시점 6과 7에서 작업하고 시점 7에 잠금을 해제한다. H는 시점 8에 획득한다. 상속이 있으면 L의 유효 우선순위가 3이므로 L은 여전히 시점 2와 3에 작업하고, H는 시점 4에 획득한다. M의 늘어난 작업은 그 뒤에 실행된다. 이 모델에서는 상속이 장치와 관계없는 M의 작업량 증가로부터 H의 획득 순서를 보호한다.

2. 시점 1에 값 1을 넣고 시점 2에 값 2를 넣으려다 기다린다. 시점 3에는 생산자가 계속 공간을 기다리고 소비자도 아직 시작하지 않아 idle이 출력된다. 이때 세 번째 이벤트는 세마포어에 남는다. 소비자는 시점 4에 값 1을 받고, 생산자는 시점 5에 값 2를 넣는다. 시점 6에는 남은 이벤트를 받아 값 3을 만들지만 큐가 차 있어 다시 기다린다. 소비자는 시점 7에 값 2를 받고, 생산자는 시점 8에 값 3을 넣으며, 소비자는 시점 9에 값 3을 받는다.

3. 항상 구분할 수는 없다. 이진 세마포어가 이미 1인 동안 추가 통지가 들어오면 개수가 2나 3으로 늘어나지 않는다. 이후 한 번 획득한 태스크는 세 번 발생했다는 정보를 복원할 수 없다. 모든 감지를 세어야 한다면 충분한 범위의 카운팅 세마포어나 명시적인 이벤트 저장이 필요하다. 다만 카운팅 세마포어도 측정값 자체를 보관하지 않으므로 각 물체의 데이터 보존은 별도로 설계해야 한다.

4. 바로 덮어쓰면 안 된다. 구조체를 복사해도 포인터가 가리키는 버퍼 내용까지 복사되지는 않는다. 첫 번째 방법은 크기가 제한된 측정 바이트 배열을 큐 항목 안에 직접 넣어 항목과 함께 복사하는 것이다. 두 번째 방법은 미리 준비한 버퍼의 사용 권한을 소비자에게 넘기고, 소비자가 반환할 때까지 생산자가 해당 버퍼를 재사용하지 않는 것이다. 어느 경우든 큐 전송 실패 시 데이터와 버퍼의 사용 권한이 누구에게 남는지 정해야 한다.