ISR 과 메인 루프 데이터 공유

이 장에서 배우는 것

앞 장에서 컨베이어의 물체 감지 신호를 인터럽트로 받도록 바꾸었다. 이제 인터럽트 서비스 루틴(ISR)은 물체 번호를 기록하고, 메인 루프는 그 번호를 꺼내 검사 대상으로 등록해야 한다. 두 실행 흐름이 같은 데이터를 사용하면서부터 문제는 신호를 받는 방법보다 데이터를 넘기는 방법에 있다. 값을 읽는 도중 인터럽트가 발생하거나, 기록이 끝나기 전에 새 데이터가 있다고 알리면 물체 수와 검사 순서가 어긋날 수 있다.

이 장에서는 단일 코어 MCU에서 하나의 ISR이 생산하고 메인 루프가 소비하는 구조를 다룬다. PC에서는 스레드와 잠금으로 실행 경계를 표현하되, MCU의 인터럽트 제어와 PC의 스레드 잠금이 같은 명령은 아니라는 점을 구분한다. 실행 예제는 정해진 순서로 센서 신호를 발생시켜 결과를 재현한다.

• 원자성(atomicity)을 개별 읽기·쓰기와 읽고 수정하는 연산으로 나누어 설명한다.
• 크리티컬 섹션(critical section)으로 공유 카운터를 일관되게 가져온다.
• 단일 생산자·단일 소비자(SPSC) 링 버퍼에서 인덱스의 소유권과 공개 순서를 정한다.
• C11 원자 연산을 사용한 PC 예제와 실제 MCU에 적용할 때의 조건을 구분한다.

문제 상황

컨베이어 입구의 광센서가 물체 하나를 감지할 때마다 ISR이 실행된다고 하자. 초기 구현에서는 ISR이 공유 변수 pending을 증가시키고, 메인 루프가 그 값을 읽은 뒤 0으로 지운다. 평소에는 정상처럼 보이지만 물체 간격이 짧아지면 메인 루프가 처리한 물체 수가 센서에서 받은 신호 수보다 작아진다.

예를 들어 메인 루프가 pending에서 3을 읽은 직후 ISR이 값을 4로 증가시킬 수 있다. 이어서 메인 루프가 0을 저장하면 새로 들어온 신호 하나가 사라진다. 각각의 읽기와 쓰기가 한 번에 수행되는 MCU에서도 이런 손실은 발생한다. 보호해야 할 대상은 변수 하나만이 아니라 “현재 값을 가져오고 초기화한다”는 연속된 작업이다.

물체 수만으로는 개별 검사 작업을 관리하기도 어렵다. 어떤 물체를 먼저 받았는지 알아야 하므로 ISR이 순번을 큐에 넣고 메인 루프가 순서대로 꺼내도록 바꾼다. 그러나 배열에 데이터를 쓰기 전에 쓰기 위치부터 옮기면, 메인 루프가 아직 작성되지 않은 칸을 읽을 수 있다. 빈 칸이 없을 때 기존 데이터를 덮어쓰면 오래 기다린 물체의 기록이 사라진다.

이 장의 컨베이어 예제는 이 두 문제를 나누어 해결한다. 감지 횟수와 손실 횟수는 짧은 크리티컬 섹션으로 보호한다. 물체 번호는 용량이 고정된 링 버퍼로 넘긴다. 버퍼가 가득 차면 새 항목을 버리고 손실 횟수를 증가시키는 정책을 사용한다. 이 정책은 예제를 위한 선택이며, 실제 설비에서 기록 손실을 허용할지는 별도로 결정해야 한다.

원자성과 가시성은 다른 문제다

원자적인 접근은 다른 실행 흐름이 그 접근의 중간 상태를 관찰하지 못하는 접근이다. 예를 들어 8비트 MCU에서 16비트 카운터를 읽으려면 여러 번의 메모리 접근이 필요할 수 있다. 아래 바이트를 읽은 뒤 ISR이 카운터를 바꾸고 위 바이트를 읽으면, 실제로 저장된 적 없는 조합을 얻을 수 있다. 반대로 어떤 MCU가 정렬된 32비트 읽기를 한 번에 수행하더라도 32비트 변수에 대한 모든 식이 원자적인 것은 아니다.

count++는 논리적으로 읽기, 덧셈, 쓰기로 이루어진다. 원자적인 읽기와 원자적인 쓰기를 차례로 사용했다고 해서 전체 증가 연산이 원자적으로 바뀌지는 않는다. 마찬가지로 원자 변수에 대한 읽기 다음에 0을 저장하는 두 연산도 하나의 작업이 아니다. 그 사이에 들어온 증가를 보존하려면 교환 연산 하나를 사용하거나 전체 구간을 보호해야 한다.

volatile은 이런 보호 장치가 아니다. 하드웨어 레지스터처럼 프로그램 바깥의 요인으로 값이 달라질 수 있는 객체를 접근할 때 필요하지만, 여러 실행 흐름 사이의 상호 배제를 제공하지 않는다. 일반 데이터의 기록을 다른 실행 흐름에 어떤 순서로 공개할지도 정해 주지 않는다. 따라서 공유 변수를 volatile로 선언하는 것만으로 카운터 손실이나 링 버퍼의 공개 순서가 해결되지는 않는다.

PC에서 두 스레드가 같은 일반 객체에 동기화 없이 접근하고 그중 하나가 쓰기를 수행하면 C의 데이터 경쟁이 발생한다. 이는 단순히 가끔 오래된 값을 읽는 현상으로 제한되지 않는다. 프로그램의 동작 자체를 언어 규칙으로 설명할 수 없게 된다. 눈에 보이는 출력이 맞았다는 이유만으로 공유 방식이 올바르다고 판단해서는 안 된다.

C11의 <stdatomic.h>는 원자 객체와 메모리 순서를 제공한다. 원자 객체는 그 객체 자체의 접근을 보호하고, 적절한 메모리 순서는 그 주변의 일반 데이터 접근까지 연결한다. 링 버퍼에서는 슬롯의 물체 번호를 먼저 쓴 뒤 인덱스를 공개하고, 소비자는 공개된 인덱스를 확인한 뒤 슬롯을 읽도록 연결해야 한다.

공유 데이터 문제에 따라 필요한 수단이 달라진다

수단	해결하는 문제	그 자체로 해결하지 않는 문제
volatile	구현 규칙에 따른 해당 객체의 접근 유지	상호 배제, 복합 연산의 원자성, 스레드 동기화
C11 원자 연산	원자 객체 접근과 지정한 메모리 순서	임의의 여러 연산을 하나의 거래로 묶기
크리티컬 섹션	충돌하는 실행 흐름 사이에서 구간 보호	긴 보호 구간으로 인한 응답 지연
SPSC 소유권 규칙	각 인덱스의 쓰기 주체를 하나로 제한	여러 생산자나 여러 소비자의 동시 사용

크리티컬 섹션은 읽기와 초기화를 함께 보호한다

감지 횟수를 가져오는 동작에는 세 단계가 있다. 충돌하는 ISR의 실행을 막고, 현재 횟수를 지역 변수에 복사한 뒤 공유 횟수를 0으로 바꾸고, 이전 실행 상태를 복원한다. 복사한 값의 출력이나 검사 작업은 구간을 벗어난 뒤 수행한다. 이렇게 하면 신호 하나는 이번에 가져오는 묶음이나 다음 묶음 중 한쪽에 포함된다.

MCU에서는 보통 관련 인터럽트를 마스킹하여 이 구간을 만든다. 진입하기 전의 마스크 상태를 저장하고 종료할 때 복원해야 한다. 단순히 끝에서 인터럽트를 켜면 바깥 호출자가 이미 막아 둔 인터럽트까지 켤 수 있다. 중첩을 허용하는 프로젝트라면 중첩 깊이와 복원 책임도 정해져 있어야 한다.

어떤 인터럽트가 실제로 막히는지도 확인해야 한다. 우선순위 경계로 마스킹하는 방식에서는 더 높은 우선순위의 ISR이 계속 실행될 수 있다. 그 ISR도 같은 카운터를 수정한다면 보호가 성립하지 않는다. 또한 단일 코어에서의 인터럽트 마스킹은 다른 코어의 접근을 막지 않는다. 보호 범위는 “인터럽트를 껐다”는 문장보다 구체적이어야 한다.

완성 코드의 hal_sim_irq_gate는 PC용 뮤텍스다. 센서 스레드와 메인 루프가 같은 잠금을 사용하여 동시에 보호 구간에 들어가지 못하게 한다. 이는 MCU의 인터럽트 마스킹이 만드는 상호 배제를 모사한다. 실제 ISR 안에서 이 뮤텍스처럼 대기 가능한 잠금을 사용하라는 뜻은 아니다. 실제 ISR에서는 선택한 MCU와 실행 환경의 인터럽트 제어 수단을 사용한다.

보호 구간의 길이도 설계 대상이다. 카운터 두 개를 복사하는 일과 문자열을 출력하는 일은 걸리는 시간이 다르다. 출력, 외부 통신, 완료 대기를 보호 구간 안에 넣으면 센서 처리 지연이 커질 수 있다. 필요한 데이터만 지역 변수로 가져오고 나머지는 밖에서 수행하는 습관이 중요하다.

실제 FreeRTOS에 옮길 때는 아래 표를 출발점으로 삼는다. 이 장에서는 태스크 간 동기화 기능의 사용법까지 확장하지 않는다. 특히 커널 함수를 호출할 수 있는 ISR 우선순위와 인터럽트 마스크의 범위는 사용하는 포트의 규칙을 함께 확인해야 한다.

PC 공유 방식과 FreeRTOS에서 검토할 수단의 대응

이 장의 동작	FreeRTOS 대응	적용 조건
메인 쪽 짧은 공유 구간	taskENTER_CRITICAL(), taskEXIT_CRITICAL()	태스크 문맥에서 사용하며 해당 ISR이 마스킹되는지 확인한다.
ISR 쪽 마스크 저장·복원	taskENTER_CRITICAL_FROM_ISR(), taskEXIT_CRITICAL_FROM_ISR(saved)	지원 포트에서 저장값을 짝지어 복원한다.
ISR에서 항목 전달	xQueueSendFromISR()	직접 만든 링 버퍼를 대신하는 선택지이며 가득 참을 처리해야 한다.
소비자가 항목 수신	xQueueReceive()	태스크 문맥에서 사용하는 수신 함수다.

FreeRTOS의 크리티컬 섹션은 C11 원자 연산과 같은 추상화가 아니다. 전자는 실행 가능한 문맥의 범위를 제한하고, 후자는 언어 수준에서 객체 접근과 순서를 정의한다. PC 코드의 잠금을 이름만 바꾸어 이식하지 말고, 누가 공유 데이터를 수정하는지부터 다시 확인해야 한다. API 사실 확인에는 FreeRTOS 크리티컬 섹션 문서와 ISR용 큐 전송 문서를 참고할 수 있다.

링 버퍼는 소유권과 공개 순서로 연결한다

링 버퍼(ring buffer)는 배열 끝에 도달하면 처음으로 돌아가서 사용하는 저장 구조다. 예제의 head는 생산자가 다음에 기록할 위치이고, tail은 소비자가 다음에 읽을 위치다. 생산자만 head를 바꾸고 소비자만 tail을 바꾼다. 상대편 인덱스는 읽어서 진행 가능 여부를 판단한다.

배열 크기는 4지만 항목은 3개까지만 저장한다. head == tail이면 비어 있고, next(head) == tail이면 가득 찬 것으로 정한다. 한 칸을 비워 두면 별도의 공유 개수 없이 두 상태를 구분할 수 있다. 배열 크기는 2 이상이어야 하며, 여기서는 나머지 연산으로 다음 위치를 계산하므로 2의 거듭제곱일 필요는 없다.

생산자는 먼저 소비자의 읽기 위치를 확인한다. 공간이 있으면 자기 슬롯에 물체 번호를 쓰고, 그다음 head를 공개한다. 소비자는 공개된 head를 확인하고 슬롯을 지역 변수에 복사한 뒤 tail을 공개한다. 소비자가 복사하기 전에 칸을 반환하면 생산자가 같은 칸에 새 값을 써 버릴 수 있다.

코드에서는 공개하는 저장에 memory_order_release를 사용하고, 상대편 공개값을 읽는 데 memory_order_acquire를 사용한다. 해제 저장과 획득 읽기가 해당 공개를 연결하면, 저장 전에 끝난 슬롯 접근이 상대편의 이후 접근보다 먼저 일어난 것으로 정해진다. 따라서 슬롯 배열 자체를 원자 객체로 만들지 않고도 올바른 순서로 읽고 쓸 수 있다.

이 연결은 양쪽 방향에 필요하다. 생산자의 head 공개는 “이 칸의 기록이 끝났다”는 뜻이다. 소비자의 tail 공개는 “이 칸의 읽기가 끝났으므로 다시 써도 된다”는 뜻이다. 처음 방향만 생각하면 데이터 전달은 보호하더라도 슬롯 재사용은 보호하지 못할 수 있다.

자신만 수정하는 인덱스를 읽는 데는 memory_order_relaxed를 사용한다. 상대편 슬롯 접근과 순서를 연결할 필요가 없는 읽기이기 때문이다. 상대편 인덱스가 진행 중일 때 이전 값을 관찰하면 당장 진행할 수 없다고 판단할 수는 있어도, 소유권 규칙 안에서는 아직 반환되지 않은 슬롯을 마음대로 사용하지 않는다. 생산자는 실패 결과에 정해진 넘침 정책을 적용하고, 소비자는 다음 호출에서 다시 확인한다.

여기에 두 번째 생산자를 추가하면 이야기가 달라진다. 두 ISR이 같은 head를 읽고 같은 슬롯에 쓸 수 있기 때문이다. 인덱스가 원자 변수라는 사실만으로 슬롯 예약까지 원자적으로 이루어지지는 않는다. 같은 ISR의 중첩 실행도 동시에 생산하는 실행 흐름을 만들 수 있다. 이 구현의 사용 조건에는 생산자와 소비자가 각각 동시에 하나만 실행된다는 제한이 포함된다.

또한 C11은 모든 원자 타입이 내부 잠금 없이 구현된다고 보장하지 않는다. PC 스레드에서는 사용할 수 있는 구현도 MCU ISR에서는 적합하지 않을 수 있다. 대상 컴파일러의 ISR 규칙, 정렬 요구, 생성된 명령을 확인하고 필요하면 atomic_is_lock_free()로 해당 객체의 구현 특성을 확인한다. 잠금 없는 구현이라는 조건 하나가 ISR 적합성이나 최대 실행 시간까지 보장하지는 않는다.

완성 코드

다음 프로그램을 isr_sharing.c로 저장한다. hal_sim_ 접두어 함수는 PC 시뮬레이션 계층이다. 센서 신호 묶음마다 스레드를 만들고 종료를 기다린 뒤 소비 작업을 호출한다. 간단한 협동형 스케줄러 시뮬레이터는 매 회차 등록된 작업을 한 번 실행하며, 작업은 항목을 최대 두 개 꺼내고 반환한다. 여기서 회차는 실행 순서이며 실제 시간을 뜻하지 않는다.

센서 신호 수는 차례로 2개, 4개, 2개다. 출력 순서를 고정하기 위해 생산과 소비 단계를 기다림으로 구분했다. 따라서 이 실행은 동시에 엇갈리는 모든 경우를 시험하는 부하 시험이 아니다. 보호 구간의 사용법, 링 버퍼의 경계 조건, 넘침 정책을 재현하는 예제다. 링 버퍼 함수의 공개 순서는 이 실행 순서에 기대지 않도록 작성했다.

#include <inttypes.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

enum {
    RING_SIZE = 4,
    WORK_BUDGET = 2
};

/* [01] Shared objects and their owners. */
typedef struct {
    uint32_t slots[RING_SIZE];
    atomic_uint head;
    atomic_uint tail;
} Ring;

typedef struct {
    uint32_t pending;
    uint32_t dropped;
} Stats;

static Ring events;
static pthread_mutex_t hal_sim_irq_gate =
    PTHREAD_MUTEX_INITIALIZER;

/* Protected by hal_sim_irq_gate. */
static uint32_t pending_edges;
static uint32_t dropped_events;

/* [02] POSIX thread error handling. */
static void check_pthread(int error, const char *operation)
{
    if (error != 0) {
        fprintf(stderr, "%s: %s\n", operation, strerror(error));
        exit(EXIT_FAILURE);
    }
}

static void hal_sim_enter_critical(void)
{
    check_pthread(pthread_mutex_lock(&hal_sim_irq_gate),
                  "pthread_mutex_lock");
}

static void hal_sim_leave_critical(void)
{
    check_pthread(pthread_mutex_unlock(&hal_sim_irq_gate),
                  "pthread_mutex_unlock");
}

/* [03] Initialize before starting a producer. */
static void ring_init(Ring *ring)
{
    atomic_init(&ring->head, 0u);
    atomic_init(&ring->tail, 0u);
}

static unsigned next_index(unsigned index)
{
    return (index + 1u) % (unsigned)RING_SIZE;
}

/* [04] Called by the single producer only. */
static bool ring_push(Ring *ring, uint32_t value)
{
    unsigned head =
        atomic_load_explicit(&ring->head, memory_order_relaxed);
    unsigned next = next_index(head);
    unsigned tail =
        atomic_load_explicit(&ring->tail, memory_order_acquire);

    if (next == tail) {
        return false;
    }

    ring->slots[head] = value;
    atomic_store_explicit(&ring->head, next, memory_order_release);
    return true;
}

/* [05] Called by the single consumer only. */
static bool ring_pop(Ring *ring, uint32_t *value)
{
    unsigned tail =
        atomic_load_explicit(&ring->tail, memory_order_relaxed);
    unsigned head =
        atomic_load_explicit(&ring->head, memory_order_acquire);

    if (tail == head) {
        return false;
    }

    *value = ring->slots[tail];
    atomic_store_explicit(&ring->tail, next_index(tail),
                          memory_order_release);
    return true;
}

/* [06] Simulated ISR: no printing or waiting for buffer space. */
static void sensor_isr(uint32_t sequence)
{
    hal_sim_enter_critical();
    ++pending_edges;
    if (!ring_push(&events, sequence)) {
        ++dropped_events;
    }
    hal_sim_leave_critical();
}

/* [07] Copy and clear belong to one protected operation. */
static Stats take_stats(void)
{
    Stats result;

    hal_sim_enter_critical();
    result.pending = pending_edges;
    result.dropped = dropped_events;
    pending_edges = 0u;
    hal_sim_leave_critical();

    return result;
}

/* [08] A request remains alive until its worker is joined. */
typedef struct {
    uint32_t first_sequence;
    unsigned count;
} Burst;

static void *hal_sim_sensor_thread(void *argument)
{
    const Burst *burst = argument;

    for (unsigned i = 0u; i < burst->count; ++i) {
        sensor_isr(burst->first_sequence + (uint32_t)i);
    }
    return NULL;
}

static void hal_sim_raise_burst(uint32_t first_sequence,
                                unsigned count)
{
    Burst burst = { first_sequence, count };
    pthread_t thread;

    check_pthread(
        pthread_create(&thread, NULL, hal_sim_sensor_thread, &burst),
        "pthread_create");
    check_pthread(pthread_join(thread, NULL), "pthread_join");
}

/* [09] One cooperative task invocation has a fixed item budget. */
static void conveyor_task(unsigned tick)
{
    Stats stats = take_stats();

    printf("tick=%u edges=%" PRIu32 " dropped=%" PRIu32 "\n",
           tick, stats.pending, stats.dropped);

    for (unsigned i = 0u; i < (unsigned)WORK_BUDGET; ++i) {
        uint32_t sequence;

        if (!ring_pop(&events, &sequence)) {
            break;
        }
        printf("  inspect=%" PRIu32 "\n", sequence);
    }
}

/* [10] Minimal cooperative scheduler simulator. */
typedef void (*SimTask)(unsigned tick);

static void scheduler_sim_step(unsigned tick)
{
    static const SimTask tasks[] = { conveyor_task };
    const size_t count = sizeof tasks / sizeof tasks[0];

    for (size_t i = 0u; i < count; ++i) {
        tasks[i](tick);
    }
}

/* [11] Fixed input makes overflow and wraparound reproducible. */
int main(void)
{
    static const unsigned bursts[] = { 2u, 4u, 2u };
    const size_t count = sizeof bursts / sizeof bursts[0];
    uint32_t first_sequence = 101u;

    ring_init(&events);

    for (size_t i = 0u; i < count; ++i) {
        hal_sim_raise_burst(first_sequence, bursts[i]);
        first_sequence += (uint32_t)bursts[i];
        scheduler_sim_step((unsigned)i + 1u);
    }

    puts("drain");
    uint32_t sequence;
    while (ring_pop(&events, &sequence)) {
        printf("  inspect=%" PRIu32 "\n", sequence);
    }

    Stats final_stats = take_stats();
    printf("done pending=%" PRIu32 " dropped=%" PRIu32 "\n",
           final_stats.pending, final_stats.dropped);

    check_pthread(pthread_mutex_destroy(&hal_sim_irq_gate),
                  "pthread_mutex_destroy");
    return EXIT_SUCCESS;
}

줄별 해설

[01] 공유 객체 선언. slots는 일반 배열이고 두 인덱스만 원자 객체다. 슬롯 접근을 안전하게 만드는 것은 배열의 특별한 성질이 아니라 인덱스를 통한 공개와 반환이다. pending_edges와 dropped_events는 같은 잠금 안에서만 접근한다. 정적 저장 기간을 가지므로 두 카운터의 초기값은 0이다.

[02] 오류 검사와 보호 구간. POSIX 스레드 함수는 오류 번호를 반환하므로 반환값을 직접 검사한다. 잠금 진입 실패를 무시하고 공유 데이터를 계속 사용하면 보호 규칙이 깨진다. 예제는 오류 내용을 표준 오류로 출력하고 종료한다. 정상 실행 결과에는 이 경로의 출력이 나타나지 않는다.

[03] 초기화와 다음 위치. atomic_init()은 생산자를 시작하기 전에 호출한다. 사용 중인 링 버퍼에 다시 호출하여 초기화해서는 안 된다. next_index()는 0, 1, 2, 3, 0 순으로 위치를 돌려준다. 두 인덱스가 0으로 같으므로 시작 상태는 비어 있다.

[04] 생산자의 각 접근. 처음 두 줄은 자신이 소유한 쓰기 위치와 다음 위치를 구한다. 이어지는 획득 읽기는 소비자가 어디까지 슬롯을 반환했는지 확인한다. next == tail이면 배열이나 인덱스를 바꾸지 않고 실패한다. 성공 경로에서는 슬롯 대입이 해제 저장보다 앞에 놓인다. head를 새 값으로 저장하는 줄이 항목의 공개 지점이다.

[05] 소비자의 각 접근. 소비자는 자신의 읽기 위치를 읽고 생산자의 공개 위치를 획득한다. 비어 있으면 출력 인자가 가리키는 값은 바꾸지 않는다. 성공하면 슬롯을 먼저 복사하고 그 뒤 읽기 위치를 공개한다. 함수가 반환된 뒤에는 슬롯이 재사용되어도 지역 변수 sequence에 복사된 번호는 유지된다.

[06] 감지와 전달의 구분. 신호 하나마다 pending_edges를 먼저 증가시킨다. 링 버퍼에 넣지 못한 신호도 감지 횟수에는 포함된다. 실패했을 때만 dropped_events를 증가시킨다. 버퍼가 빌 때까지 반복하지 않으므로 소비자가 늦어져도 ISR 작업이 버퍼 공간을 기다리며 계속 머무르지 않는다. 진입 함수의 잠금 대기는 PC에서 ISR 진입을 모사하기 위한 부분이다.

[07] 묶음 가져오기. 두 카운터를 지역 구조체에 복사하고, 그중 미처리 감지 횟수만 0으로 만든다. 손실 횟수는 시작 이후의 누적값으로 유지한다. 두 값을 같은 보호 구간에서 가져오므로 복사 도중 센서 스레드가 카운터를 바꾸지 못한다. 다만 이 구조체와 링 버퍼 전체를 하나의 스냅샷으로 가져오는 것은 아니다.

[08] 시뮬레이션 요청 수명. 작업 스레드에 전달하는 burst는 함수의 지역 변수다. pthread_join()으로 작업 완료를 기다린 뒤 함수가 반환하므로 작업 스레드가 사용하는 동안 살아 있다. 기다림을 제거하고 바로 반환하면 이 포인터의 수명 조건이 깨진다. 또한 이 기다림 덕분에 서로 다른 묶음의 생산자가 겹치지 않는다.

[09] 소비 작업. 통계 출력은 보호 구간 밖에서 수행한다. 반복문은 최대 두 항목만 꺼내므로 한 번의 호출이 처리할 항목 수가 제한된다. 여기서는 검사 대상을 출력하지만, 출력에 걸리는 시간까지 제한된다는 뜻은 아니다. 함수 호출마다 작업량을 나누는 구조와 정확한 실행 시간 보장은 구분해야 한다.

[10]과 [11] 실행 순서. 스케줄러는 등록된 함수를 차례로 호출하고 함수가 반환되어야 다음으로 진행한다. 입력 묶음의 시작 번호는 전달 성공 여부와 관계없이 이동한다. 따라서 손실된 번호가 나중에 재사용되지 않는다. 마지막 반복문은 남은 항목을 비우며, 모든 센서 스레드가 종료된 뒤 잠금을 파괴한다.

실행 결과

macOS 또는 POSIX 스레드를 지원하는 Linux 환경에서 다음 명령으로 빌드하고 실행한다. 별도의 라이브러리 소스나 외부 입력 파일은 필요하지 않다.

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

정상 실행의 예상 표준 출력은 다음과 같다.

tick=1 edges=2 dropped=0
  inspect=101
  inspect=102
tick=2 edges=4 dropped=1
  inspect=103
  inspect=104
tick=3 edges=2 dropped=1
  inspect=105
  inspect=107
drain
  inspect=108
done pending=0 dropped=1

첫 묶음의 두 항목은 모두 들어가고 모두 소비된다. 두 번째 묶음에서는 103, 104, 105가 들어가며 106은 가득 찬 버퍼 때문에 버려진다. 소비 작업이 103과 104를 꺼내므로 105가 남는다. 세 번째 묶음의 107과 108은 빈 두 칸에 들어가고, 소비자는 먼저 들어온 105 다음에 107을 꺼낸다. 마지막에 남은 번호는 108이다.

각 회차의 edges를 더하면 8이며, 검사 출력은 7개이고 누적 손실은 1이다. 버퍼를 모두 비운 이 실행에서는 감지 횟수와 전달 결과가 이처럼 맞아야 한다. 마지막의 pending=0은 마지막 통계 수집 이후 새 감지가 없었다는 뜻이다. 일반적으로 이 값이 0이라고 해서 링 버퍼까지 비었다고 판단할 수는 없다.

실무에서 자주 틀리는 것

읽고 지우는 두 연산을 따로 보호한다

다음은 원자 카운터 edges를 사용하더라도 잘못된 패턴이다. 두 줄 사이에서 ISR이 증가시키면 두 번째 줄이 새 증가분까지 지운다. 각 줄의 원자성과 두 줄 전체의 원자성은 다르다.

/* Wrong: edges is an atomic_uint. */
unsigned taken = atomic_load(&edges);
atomic_store(&edges, 0u);

단일 카운터만 가져와 초기화한다면 교환 연산 하나로 묶을 수 있다. 이 대안에서는 ISR도 atomic_fetch_add_explicit() 같은 원자 증가를 사용해야 한다. 아래 카운터는 다른 데이터의 준비 상태를 알리는 용도가 아니므로 완화 순서로 충분하다. 여러 필드의 일관된 묶음이 필요하면 완성 코드처럼 구간 전체를 보호한다.

/* Correct: take the count and clear it in one operation. */
unsigned taken =
    atomic_exchange_explicit(&edges, 0u, memory_order_relaxed);

인덱스를 먼저 공개한다

다음 두 줄은 순서가 반대다. 해제 저장이 있어도 그 뒤에 수행하는 슬롯 기록까지 앞선 공개에 포함되는 것은 아니다. 소비자가 새 인덱스를 확인하는 시점에 슬롯에는 이전 값이 남아 있을 수 있다.

/* Wrong: publish before writing the payload. */
atomic_store_explicit(&ring->head, next, memory_order_release);
ring->slots[head] = value;

슬롯 기록을 끝낸 뒤 인덱스를 공개한다. 소비자 쪽에서도 획득 읽기가 필요하다. 생산자의 저장 순서만 고쳐 놓고 소비자의 상대편 인덱스 읽기를 완화 순서로 바꾸면 일반 슬롯 데이터에 필요한 동기화 연결을 잃는다.

/* Correct: write, then publish. */
ring->slots[head] = value;
atomic_store_explicit(&ring->head, next, memory_order_release);

가득 차면 생산자가 읽기 위치를 옮긴다

오래된 항목 하나를 버리겠다며 생산자가 tail을 수정하는 경우가 있다. 다음 수정은 소비자에게만 있던 쓰기 권한을 생산자에게도 준다. 소비자가 슬롯을 읽는 중인지 확인하지 않은 채 그 슬롯을 재사용할 수 있어 기존 SPSC 알고리즘의 조건이 깨진다.

/* Wrong inside ring_push(): producer changes consumer state. */
if (next == tail) {
    atomic_store_explicit(&ring->tail, next_index(tail),
                          memory_order_release);
}

이 장의 정책에서는 생산자가 실패를 반환하고 호출자가 손실을 기록한다. 오래된 항목을 버리는 정책이 필요하다면 그 정책에 맞는 별도의 동기화 설계가 필요하다. 아래 수정은 완성 코드의 두 위치에 각각 대응한다.

/* Correct inside ring_push(). */
if (next == tail) {
    return false;
}

/* Correct inside the protected part of sensor_isr(). */
if (!ring_push(&events, sequence)) {
    ++dropped_events;
}

출력까지 보호 구간에 넣는다

다음 패턴은 카운터 읽기 자체를 보호하지만 출력이 끝날 때까지 센서 쪽 진행을 막는다. PC 표준 출력은 내부 잠금이나 입출력 대기를 포함할 수 있다. MCU에서 같은 구조로 인터럽트를 마스킹한 채 통신 출력을 수행하면 긴 지연이 생길 수 있다.

/* Wrong: output extends the protected interval. */
hal_sim_enter_critical();
printf("pending=%" PRIu32 "\n", pending_edges);
pending_edges = 0u;
hal_sim_leave_critical();

복사와 초기화만 안에서 하고 출력은 밖으로 옮긴다. 보호 구간에서 얻은 지역 복사본은 공유 변수가 나중에 바뀌어도 안전하게 사용할 수 있다.

/* Correct: use a local snapshot outside the protected interval. */
uint32_t taken;

hal_sim_enter_critical();
taken = pending_edges;
pending_edges = 0u;
hal_sim_leave_critical();

printf("pending=%" PRIu32 "\n", taken);

한눈에 보기

공유 데이터마다 소유자와 보호 규칙을 기록한다

대상	수정 주체	보호 규칙	확인할 경계
미수집 감지 횟수	ISR 증가, 메인 초기화	같은 크리티컬 섹션	복사와 초기화 사이의 감지
누적 손실 횟수	ISR	쓰기와 읽기에 같은 보호 적용	장시간 운전 시 카운터 범위
head	생산자만	슬롯 기록 뒤 해제 저장	다음 위치가 tail인지
tail	소비자만	슬롯 복사 뒤 해제 저장	현재 위치가 head인지
슬롯 배열	생산자 기록, 소비자 읽기	상대편 인덱스의 획득 읽기	소비 완료 전 재사용 여부
PC 실행 순서	시뮬레이션 계층	생산 스레드 종료 후 작업 호출	동시 실행 검증과 구분

공유 변수마다 누가 쓰는지와 어떤 동작을 묶어야 하는지를 먼저 적으면 보호 수단을 선택하기 쉬워진다. 단일 필드의 원자 접근, 여러 필드의 일관된 복사, 슬롯 소유권의 이전은 서로 다른 요구다. 한 종류의 선언이나 잠금으로 모두 해결하려고 하기보다 각 데이터의 사용 규칙을 코드 가까이에 남기는 편이 낫다.

예제의 카운터는 부호 없는 32비트 값이므로 범위를 넘으면 순환한다. 짧은 고정 입력에서는 문제가 없지만 장시간 누적 손실을 관리할 때는 허용 범위, 초기화 시점, 넘침 표시를 정해야 한다. 버퍼 크기를 늘리는 것도 일시적인 입력 집중을 흡수할 뿐이다. 입력이 지속적으로 소비보다 빠르면 고정 크기 버퍼는 결국 가득 찬다.

연습 문제

1. 나머지 코드는 그대로 두고 RING_SIZE를 5로 바꾸었다. 손실되는 번호, 각 회차에 검사되는 번호, 마지막 배출 단계에서 검사되는 번호를 구하라.
2. take_stats()가 잠금을 풀고 나서 pending_edges = 0u;를 실행하도록 바뀌었다. 감지 기록 하나가 사라지는 실행 순서를 설명하라. PC에서 동기화되지 않은 일반 변수 접근이 추가로 갖는 문제도 설명하라.
3. 생산자의 head 저장은 해제 순서를 유지하고 소비자의 head 읽기만 완화 순서로 바꾸었다. 인덱스가 원자 변수인데도 슬롯 접근의 안전성을 설명할 수 없는 이유를 쓰라.
4. 다른 센서 ISR도 같은 링 버퍼의 ring_push()를 호출하게 하려 한다. 두 생산자가 서로 겹칠 수 있을 때 깨지는 조건과 구조를 유지하면서 해결할 수 있는 방법 하나를 설명하라.

정답과 해설

1. 실제 저장 용량은 4개가 되므로 손실되는 번호는 없다. 첫 회차는 101과 102를 검사한다. 두 번째 회차에서는 103부터 106까지 저장한 뒤 103과 104를 검사한다. 세 번째 회차에서는 남아 있던 105, 106 뒤에 107, 108이 들어가고 105와 106을 검사한다. 마지막 배출 단계는 107과 108을 검사한다. 누적 손실은 모든 회차에서 0이다.

2. 메인 루프가 기존 감지 횟수를 복사하고 잠금을 푼다. 그다음 ISR이 잠금 안에서 새 감지 하나를 더한다. 이어서 메인 루프가 0을 저장하면 새 감지가 지워진다. 복사와 초기화를 하나의 보호 구간에 넣어야 한다. 또한 PC에서 일반 변수에 대한 초기화를 잠금 밖에서 수행하면 ISR의 접근과 데이터 경쟁이 생길 수 있으므로, 단순한 손실 시나리오만으로 전체 동작을 한정할 수 없다.

3. 원자적인 인덱스 접근은 그 인덱스 자체의 접근을 보호한다. 일반 배열 슬롯에 대한 생산자의 기록과 소비자의 읽기를 연결하려면 해당 공개를 획득하는 읽기가 필요하다. 완화 읽기로 새 인덱스 값을 얻었다고 해서 앞선 슬롯 기록과 동기화되는 것은 아니다. 소비자의 상대편 인덱스 읽기를 획득 순서로 유지해야 한다.

4. 생산자가 하나라는 조건이 깨진다. 두 ISR이 같은 쓰기 위치를 선택하고 같은 슬롯에 기록할 수 있으며, 하나의 인덱스 저장이 다른 생산자의 진행을 나타내지 못할 수 있다. 센서별로 링 버퍼를 하나씩 두고 각 버퍼에 생산자를 하나만 배정하면 기존 규칙을 유지할 수 있다. 메인 루프는 두 버퍼의 소비자가 된다. 이 경우 버퍼 사이의 전체 도착 순서는 자동으로 보존되지 않으므로 필요하면 별도의 순서 기준을 정해야 한다.