Ring Buffer (ftrace)

리눅스 커널 ftrace의 핵심 데이터 저장소인 ring buffer(kernel/trace/ring_buffer.c)를 심층 분석합니다. per-CPU 페이지(Page) 기반 lockless 설계, NMI-safe 중첩 쓰기 메커니즘, ring_buffer_event 헤더와 페이로드(Payload) 인코딩, splice/mmap을 통한 zero-copy 읽기, TSC delta 타임스탬프 압축, 덮어쓰기/폐기 모드, 리사이즈와 스냅샷을 커널 소스 기반으로 분석합니다. kfifo 평탄 버퍼(Buffer)와의 구조적 차이, perf 링 버퍼(Ring Buffer)와의 비교, 실전 튜닝과 디버깅(Debugging)까지 포괄합니다.

링 버퍼 개요

리눅스 커널의 ring buffer는 ftrace 트레이싱 시스템의 핵심 데이터 저장소입니다. 2008년 Steven Rostedt에 의해 도입된 이 자료 구조는 다음과 같은 요구사항을 충족하도록 설계되었습니다.

ring buffer의 핵심 설계 원칙은 "쓰기 경로의 최소 오버헤드"입니다. 트레이싱은 관찰 대상 시스템에 최소한의 영향을 주어야 하므로, 이벤트 기록 경로에서 lock 획득, 메모리 할당, 시스템 콜이 발생하면 안 됩니다. 이를 위해 다음 기법들이 사용됩니다.

• per-CPU 분리: 각 CPU가 독립된 버퍼를 소유하므로 CPU 간 동기화가 불필요합니다.
• local_cmpxchg: 같은 CPU 내에서도 인터럽트(Interrupt) 중첩 처리를 위해 lock 대신 local_cmpxchg(CPU-local CAS)를 사용합니다.
• 사전 할당: 버퍼 페이지는 트레이싱 시작 전에 미리 할당됩니다. 쓰기 경로에서 kmalloc이 호출되지 않습니다.
• commit 분리: 예약(reserve)과 확정(commit)을 분리하여, 부분 쓰기 이벤트가 읽기 측에 노출되지 않습니다.

설계 진화 타임라인

lockless 설계 철학

ring buffer가 lock 대신 lockless 기법을 선택한 이유는 트레이싱의 특수한 요구사항에 있습니다.

1. NMI 안전성 — NMI에서는 spin_lock을 잡을 수 없습니다. 이미 lock을 잡은 상태에서 NMI가 발생하면 데드락이 걸리므로, lock 기반 설계 자체가 불가능합니다.
2. 관찰자 효과 최소화 — 트레이싱은 관찰 대상에 영향을 최소화해야 합니다. lock 경합은 스케줄링 동작을 변화시키며, Heisenbug(트레이싱으로 인해 버그가 사라지는 현상)를 방지해야 합니다.
3. 인터럽트 컨텍스트 지원 — IRQ 핸들러(Handler) 내부 이벤트도 기록해야 합니다. irq_disable + lock 방식은 인터럽트 지연(Latency)을 유발합니다.
4. per-CPU + local_cmpxchg 조합 — CPU 간에는 per-CPU 분리로 동기화가 불필요하고, CPU 내에서는 local_cmpxchg로 인터럽트 중첩을 처리합니다. 결과적으로 글로벌 lock 없이 완전한 동기화를 달성합니다.

per-CPU 아키텍처

ring buffer의 가장 중요한 설계 결정은 CPU별 독립 버퍼입니다. struct trace_buffer는 CPU 개수만큼의 struct ring_buffer_per_cpu 포인터 배열을 보유합니다.

/* include/linux/ring_buffer.h (simplified) */
struct trace_buffer {
    unsigned                    flags;
    int                         cpus;
    atomic_t                    record_disabled;
    cpumask_var_t               cpumask;
    struct lock_class_key       *reader_lock_key;

    struct mutex                mutex;
    struct ring_buffer_per_cpu  **buffers;  /* per-CPU 버퍼 배열 */

    u64                         (*clock)(void);
    struct rb_irq_work          irq_work;
};

각 ring_buffer_per_cpu는 해당 CPU에서만 쓰기가 발생합니다. 이로 인해 다음과 같은 이점이 있습니다.

/* kernel/trace/ring_buffer.c */
struct ring_buffer_per_cpu {
    int                         cpu;
    atomic_t                    record_disabled;
    atomic_t                    resize_disabled;
    struct trace_buffer         *buffer;

    raw_spinlock_t              reader_lock;    /* 읽기 측 직렬화 */
    unsigned long               nr_pages;       /* 페이지 수 */
    struct list_head            *pages;         /* 페이지 순환 리스트 */

    struct buffer_page          *head_page;     /* 가장 오래된 데이터 */
    struct buffer_page          *tail_page;     /* 현재 쓰기 위치 */
    struct buffer_page          *commit_page;   /* 마지막 commit 위치 */
    struct buffer_page          *reader_page;   /* 읽기 전용 분리 페이지 */

    local_t                     entries;        /* 총 이벤트 수 */
    local_t                     overrun;        /* 덮어쓴 이벤트 수 */
    local_t                     commit_overrun; /* commit 충돌 수 */
    local_t                     dropped_events; /* 폐기된 이벤트 수 */

    local_t                     committing;     /* 중첩 commit 추적 */
    local_t                     commits;        /* commit 완료 카운터 */
    unsigned long               read;           /* reader_page 내 읽기 오프셋 */
    unsigned long               read_bytes;
    u64                         write_stamp;    /* 마지막 쓰기 타임스탬프 */
    u64                         before_stamp;   /* 예약 전 타임스탬프 */

    rb_time_t                   write_stamp_rb;
    rb_time_t                   before_stamp_rb;
    /* ... irq_work, wakeup 등 생략 ... */
};

페이지 기반 설계

ring buffer는 연속 메모리(flat buffer)가 아닌 buffer_page 순환 연결 리스트로 구성됩니다. 이 결정에는 여러 이유가 있습니다.

reader_page swap 상세

reader_page는 ring buffer에서 가장 독특한 메커니즘입니다. 읽기 측은 writer ring에서 페이지를 직접 읽지 않고, head_page를 분리하여 자신의 전용 페이지로 교환합니다.

/* rb_get_reader_page() 핵심 로직 (간략화) */
static struct buffer_page *rb_get_reader_page(
    struct ring_buffer_per_cpu *cpu_buffer)
{
    struct buffer_page *reader = cpu_buffer->reader_page;
    struct buffer_page *head;

    /* reader_page에 아직 읽지 않은 데이터가 있으면 그대로 반환 */
    if (cpu_buffer->read < rb_page_commit(reader))
        return reader;

    /* reader_page 소비 완료 → head_page와 swap */
    head = cpu_buffer->head_page;

    /* cmpxchg로 head_page의 list->prev 포인터를 교체하여
     * reader_page를 ring에 삽입하고 head_page를 분리 */
    reader->list.next = head->list.next;
    rb_set_list_to_head(&reader->list);

    /* 원자적 swap: head를 ring에서 분리, reader를 삽입 */
    unsigned long ret = cmpxchg(&head->list.prev->next,
                                  &head->list, &reader->list);

    cpu_buffer->reader_page = head;
    cpu_buffer->read = 0;
    return head;
}

페이지 끝 패딩(Padding)

이벤트가 현재 페이지에 들어가지 않으면, 남은 공간에 PADDING 이벤트를 삽입하고 다음 페이지로 이동합니다. PADDING 이벤트는 읽기 측에서 자동으로 건너뜁니다.

/* 페이지 끝 남은 공간에 PADDING 이벤트 삽입 */
static void rb_event_set_padding(
    struct ring_buffer_event *event)
{
    /* type_len = RINGBUF_TYPE_PADDING (29) */
    event->type_len = RINGBUF_TYPE_PADDING;
    event->time_delta = 0;

    /* array[0]에 남은 공간 크기 저장 */
    event->array[0] = rb_page_size(page) -
                      local_read(&page->write);
}

/*
 * 페이지 구조 예시 (4KB 페이지, BUF_PAGE_SIZE = 4080):
 *
 * +------------------------------------+
 * | buffer_data_page header (16 bytes) |
 * | - time_stamp: u64                  |
 * | - commit: local_t                  |
 * +------------------------------------+
 * | event 1 (header + payload)         |
 * | event 2 (header + payload)         |
 * | event 3 (header + payload)         |
 * | ...                                |
 * | PADDING event (남은 공간)          |
 * +------------------------------------+
 *
 * BUF_PAGE_SIZE = PAGE_SIZE - sizeof(buffer_data_page)
 *              = 4096 - 16 = 4080 bytes (사용 가능)
 */

서브버퍼 크기 설정 (5.15+)

Linux 5.15부터 ring buffer의 서브버퍼(페이지) 크기를 PAGE_SIZE가 아닌 값으로 설정할 수 있습니다. 대형 이벤트(예: 긴 문자열, 스택 트레이스)가 단일 페이지를 초과할 때 유용합니다.

# 서브버퍼 크기 설정 (기본: PAGE_SIZE = 4096)
echo 8192 > /sys/kernel/tracing/buffer_subbuf_size_kb    # 8KB 서브버퍼

# 현재 설정 확인
cat /sys/kernel/tracing/buffer_subbuf_size_kb

# 주의: 변경 시 기존 버퍼 내용이 삭제됨

ring_buffer 구조체(Struct)

struct trace_buffer(이전 명칭 struct ring_buffer)는 전체 ring buffer 인스턴스를 대표하는 최상위 구조체입니다.

clock 함수

ring buffer는 clock 함수 포인터를 통해 타임스탬프를 얻습니다. 기본값은 trace_clock_local()로, sched_clock()의 CPU-local 변형입니다.

/* 사용 가능한 trace clock 소스 */
static struct {
    char *name;
    u64 (*func)(void);
} trace_clocks[] = {
    { "local",     trace_clock_local },     /* 기본: CPU TSC, 빠름 */
    { "global",    trace_clock_global },    /* 전역 동기화, 느림 */
    { "counter",   trace_clock_counter },   /* atomic 카운터 */
    { "uptime",    ktime_get_boot_fast_ns },/* 부팅 후 시간 */
    { "perf",      ktime_get_mono_fast_ns },/* monotonic */
    { "mono",      ktime_get_mono_fast_ns },/* monotonic */
    { "mono_raw",  ktime_get_raw_fast_ns }, /* 보정 없는 monotonic */
    { "boot",      ktime_get_boot_fast_ns },/* 부팅 후 (suspend 포함) */
    { "tai",       ktime_get_tai_fast_ns }, /* TAI 시간 */
};

buffer_page 구조체

struct buffer_page는 ring buffer의 물리적 저장 단위입니다. 각 buffer_page는 PAGE_SIZE(보통 4KB) 크기의 데이터 영역과 메타데이터를 가집니다.

/* kernel/trace/ring_buffer.c */
struct buffer_page {
    struct list_head  list;       /* 순환 연결 리스트 */
    local_t           write;      /* 현재 쓰기 오프셋 (페이지 내) */
    unsigned          read;       /* reader_page 내 읽기 오프셋 */
    local_t           entries;    /* 이 페이지의 이벤트 수 */
    unsigned long     real_end;   /* 실제 데이터 끝 (패딩 전) */
    struct buffer_data_page *page; /* 실제 데이터 페이지 */
};

struct buffer_data_page {
    u64              time_stamp; /* 이 페이지의 기준 타임스탬프 */
    local_t          commit;     /* 확정된 데이터 끝 오프셋 */
    unsigned char    data[];     /* 이벤트 데이터 (가변 길이) */
};

주요 필드의 역할을 정리하면 다음과 같습니다.

head_page 플래그 인코딩

head_page를 구별하기 위해 list.prev->next 포인터의 최하위 비트에 플래그를 인코딩합니다.

#define RB_PAGE_NORMAL   0   /* 일반 페이지 */
#define RB_PAGE_HEAD     1   /* head_page (가장 오래된 데이터) */
#define RB_PAGE_UPDATE   2   /* swap 진행 중 */

/* 포인터에 플래그 비트 설정/해제 */
static inline struct list_head *
rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer,
                 struct buffer_page *head)
{
    unsigned long *ptr = (unsigned long *)&head->list.prev->next;
    *ptr |= RB_PAGE_HEAD;
    /* ... */
}

ring_buffer_event

ring buffer에 저장되는 각 이벤트는 struct ring_buffer_event 헤더와 페이로드로 구성됩니다. 헤더는 공간 효율을 위해 4바이트로 압축됩니다.

/* include/linux/ring_buffer.h */
struct ring_buffer_event {
    u32  type_len:5,     /* 이벤트 타입 + 길이 인코딩 */
         time_delta:27;  /* 이전 이벤트로부터의 시간 차이 (ns) */
    u32  array[];         /* 확장 길이 또는 페이로드 시작 */
};

/* type_len 상수 */
enum ring_buffer_type {
    RINGBUF_TYPE_DATA_TYPE_LEN_MAX = 28,  /* 1~28: 인라인 길이 */
    RINGBUF_TYPE_PADDING           = 29,  /* 페이지 끝 패딩 */
    RINGBUF_TYPE_TIME_EXTEND       = 30,  /* 시간 확장 */
    RINGBUF_TYPE_TIME_STAMP        = 31,  /* 절대 타임스탬프 */
};

/* 이벤트 크기 계산 헬퍼 */
static inline unsigned rb_event_data_length(
    struct ring_buffer_event *event)
{
    unsigned length;
    if (event->type_len)
        length = event->type_len * 4;
    else
        length = event->array[0];  /* 대형 이벤트 */
    return length;
}

쓰기 경로

ring buffer의 쓰기 경로는 성능에 가장 민감한 부분입니다. 이벤트 하나를 기록하는 전체 흐름을 추적합니다.

reserve 상세: local_cmpxchg 기반 공간 예약

/* rb_reserve_next_event 핵심 로직 (간략화) */
static noinline struct ring_buffer_event *
rb_reserve_next_event(struct trace_buffer *buffer,
                      struct ring_buffer_per_cpu *cpu_buffer,
                      unsigned long length)
{
    struct buffer_page *tail_page;
    struct ring_buffer_event *event;
    unsigned long tail, write;

again:
    tail_page = cpu_buffer->tail_page;
    tail = local_read(&tail_page->write);
    write = tail + length;

    /* 현재 페이지에 공간이 있는가? */
    if (likely(write <= BUF_PAGE_SIZE)) {
        /* Fast path: local_cmpxchg로 원자적 예약 */
        if (local_cmpxchg(&tail_page->write,
                          tail, write) != tail)
            goto again;  /* 중첩(IRQ/NMI) 발생, 재시도 */

        event = (struct ring_buffer_event *)
                (tail_page->page->data + tail);
        rb_update_event(cpu_buffer, event, length,
                        0, delta);
        return event;
    }

    /* Slow path: 페이지 전환 필요 */
    return rb_move_tail(cpu_buffer, tail, tail_page, ...);
}

commit 상세

/* ring_buffer_unlock_commit 핵심 로직 */
void ring_buffer_unlock_commit(
    struct trace_buffer *buffer)
{
    struct ring_buffer_per_cpu *cpu_buffer;

    cpu_buffer = buffer->buffers[raw_smp_processor_id()];

    /* commit 포인터 전진 시도 */
    rb_commit(cpu_buffer);

    /* 대기 중인 reader가 있으면 wakeup */
    rb_wakeups(buffer, cpu_buffer);

    preempt_enable_notrace();
}

static inline void rb_commit(
    struct ring_buffer_per_cpu *cpu_buffer)
{
    unsigned long commits;

    /* committing 카운터 감소 -- 중첩 추적 */
    commits = local_read(&cpu_buffer->commits);

    /* 가장 바깥쪽 컨텍스트만 실제 commit 수행 */
    while (local_read(&cpu_buffer->committing) == commits) {
        /* commit 포인터를 write 위치까지 전진 */
        local_set(&cpu_buffer->commit_page->page->commit,
                  local_read(&cpu_buffer->commit_page->write));
        /* 다음 페이지로 commit_page 전진 (필요시) */
        rb_set_commit_to_write(cpu_buffer);
        break;
    }
}

NMI-safe 중첩 쓰기

ring buffer가 해결해야 하는 가장 어려운 문제 중 하나는 인터럽트 중첩입니다. 일반 컨텍스트에서 이벤트를 쓰는 도중 IRQ가 발생하고, IRQ 핸들러 내에서 또 이벤트를 쓰고, 심지어 NMI가 발생할 수 있습니다.

중첩 안전성의 핵심은 다음 두 가지 메커니즘입니다.

nesting 카운터

/* 중첩 추적 */
static inline int rb_is_nested(
    struct ring_buffer_per_cpu *cpu_buffer)
{
    /* committing 카운터가 commits보다 크면 중첩 상태 */
    return local_read(&cpu_buffer->committing) !=
           local_read(&cpu_buffer->commits);
}

/* reserve 시: committing 증가 */
local_inc(&cpu_buffer->committing);

/* commit 시: commits 증가 후, 가장 바깥쪽이면 실제 commit */
local_inc(&cpu_buffer->commits);
if (local_read(&cpu_buffer->committing) ==
    local_read(&cpu_buffer->commits))
    rb_set_commit_to_write(cpu_buffer);

cmpxchg 기반 예약

local_cmpxchg를 사용한 공간 예약은 인터럽트에 의한 동시 접근을 자연스럽게 처리합니다.

중첩 깊이와 컨텍스트 분류

ring buffer는 최대 4단계 중첩을 지원합니다. 각 컨텍스트 레벨은 preempt_count로 식별됩니다.

/* 컨텍스트 레벨 식별 (간략화) */
static inline int rb_ctx_idx(int pc)
{
    if (in_nmi())
        return 3;     /* NMI 컨텍스트 */
    if (pc & HARDIRQ_MASK)
        return 2;     /* Hard IRQ 컨텍스트 */
    if (pc & SOFTIRQ_OFFSET)
        return 1;     /* Soft IRQ 컨텍스트 */
    return 0;         /* Normal 컨텍스트 */
}

/*
 * 각 컨텍스트 레벨은 독립된 nesting 추적:
 * - Normal에서 reserve → committing = 1
 * - IRQ 발생, IRQ에서 reserve → committing = 2
 * - NMI 발생, NMI에서 reserve → committing = 3
 * - NMI commit → commits = 1 (커밋 보류)
 * - IRQ commit → commits = 2 (커밋 보류)
 * - Normal commit → commits = 3 (committing == commits → 실제 commit!)
 */

NMI 중첩과 타임스탬프 문제

NMI 중첩이 발생하면 타임스탬프 순서가 역전될 수 있습니다. 예를 들어, Normal 컨텍스트에서 타임스탬프 T1을 획득한 후 NMI가 발생하여 T2(T2 > T1)를 획득하면, 페이지 내에서 T2 이벤트가 T1 이벤트보다 먼저 기록됩니다.

/* 타임스탬프 역전 감지 및 보정 */
static inline u64
rb_time_delta(struct ring_buffer_event *event,
              u64 ts, u64 prev_ts)
{
    u64 delta = ts - prev_ts;

    /* NMI 중첩으로 역전 발생 시 */
    if (unlikely((s64)delta < 0)) {
        /* delta를 0으로 처리하고 절대 타임스탬프 삽입 */
        delta = 0;
        /* rb_add_time_stamp() 호출하여 TIME_STAMP 마커 삽입 */
    }
    return delta;
}

/*
 * before_stamp / write_stamp 프로토콜:
 *
 * 1. before_stamp = 현재 타임스탬프 (예약 전)
 * 2. local_cmpxchg로 공간 예약
 * 3. write_stamp = 현재 타임스탬프 (예약 후)
 *
 * 중첩 발생 시:
 * - 상위 컨텍스트의 write_stamp가 하위보다 클 수 있음
 * - before_stamp vs write_stamp 비교로 중첩 감지
 * - 감지되면 절대 타임스탬프로 재동기화
 */

읽기 경로

ring buffer의 읽기 경로는 쓰기 경로와 달리 reader_lock(raw_spinlock)으로 보호됩니다. 주요 읽기 API는 다음과 같습니다.

/* ring_buffer_consume 핵심 로직 */
struct ring_buffer_event *
ring_buffer_consume(struct trace_buffer *buffer, int cpu,
                    u64 *ts, unsigned long *lost_events)
{
    struct ring_buffer_per_cpu *cpu_buffer;
    struct ring_buffer_event *event = NULL;

    cpu_buffer = buffer->buffers[cpu];

    raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);

    /* reader_page에서 다음 이벤트 가져오기 */
    event = rb_buffer_peek(cpu_buffer, ts, lost_events);
    if (!event)
        goto out;

    /* 읽기 포인터 전진 */
    rb_advance_reader(cpu_buffer);

out:
    raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
    return event;
}

peek vs consume 차이

peek은 이벤트를 읽되 읽기 포인터를 전진시키지 않으므로, 같은 이벤트를 여러 번 읽을 수 있습니다. /sys/kernel/tracing/trace 파일이 이 방식을 사용합니다. 반면 consume은 읽기 포인터를 전진시켜 이벤트를 "소비"하므로, trace_pipe처럼 실시간 스트리밍에 적합합니다.

rb_advance_reader 상세

이벤트를 소비(consume)할 때 rb_advance_reader가 읽기 포인터를 전진시킵니다. reader_page 내에서 다음 이벤트로 이동하거나, 페이지 소진 시 새 reader_page를 확보합니다.

/* rb_advance_reader 핵심 로직 (간략화) */
static void rb_advance_reader(
    struct ring_buffer_per_cpu *cpu_buffer)
{
    struct ring_buffer_event *event;
    struct buffer_page *reader;
    unsigned length;

    reader = cpu_buffer->reader_page;
    event = rb_reader_event(cpu_buffer);

    /* 현재 이벤트의 전체 크기(헤더 + 페이로드 + 정렬 패딩) */
    length = rb_event_length(event);

    /* PADDING 이벤트면 페이지 끝까지 건너뛰기 */
    if (event->type_len == RINGBUF_TYPE_PADDING) {
        cpu_buffer->read = rb_page_commit(reader);
        return;
    }

    /* TIME_EXTEND/TIME_STAMP은 건너뛰기 */
    if (event->type_len >= RINGBUF_TYPE_TIME_EXTEND) {
        cpu_buffer->read += length;
        return;
    }

    /* 일반 데이터 이벤트: 읽기 포인터 전진 */
    cpu_buffer->read += length;
    local_dec(&cpu_buffer->entries);
    cpu_buffer->read_bytes += length;
}

/*
 * reader_page 소진 시:
 * rb_get_reader_page() → head_page와 swap
 * → 새 reader_page에서 cpu_buffer->read = 0 재설정
 */

lost_events 추적

읽기 API에서 lost_events 매개변수는 마지막 읽기 이후 손실된 이벤트 수를 반환합니다. 이는 overwrite 모드에서 reader가 소비하지 못한 사이에 덮어쓰기가 발생한 경우를 추적합니다.

/* lost_events 계산 */
static struct ring_buffer_event *
rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer,
               u64 *ts, unsigned long *lost_events)
{
    struct buffer_page *reader;

    reader = rb_get_reader_page(cpu_buffer);
    if (!reader)
        return NULL;

    /* reader_page swap 시 분리된 페이지의 overrun을 기록 */
    if (lost_events)
        *lost_events = local_read(&cpu_buffer->lost_events);

    /* 다음 이벤트 반환 */
    return rb_reader_event(cpu_buffer);
}

iterator 인터페이스

iterator는 ring buffer를 순차적으로 탐색하기 위한 인터페이스입니다. trace 파일 읽기 시 사용됩니다.

/* iterator 생명주기 */
struct ring_buffer_iter *iter;

/* 1. iterator 시작 -- 쓰기를 비활성화 (정확한 스냅샷) */
iter = ring_buffer_read_prepare(buffer, cpu, GFP_KERNEL);
ring_buffer_read_prepare_sync();
ring_buffer_read_start(iter);

/* 2. 순차 읽기 */
while ((event = ring_buffer_iter_peek(iter, ts))) {
    /* 이벤트 처리 */
    process_event(event);
    ring_buffer_iter_advance(iter);
}

/* 3. iterator 종료 -- 쓰기 재활성화 */
ring_buffer_read_finish(iter);

iterator 내부 구조

struct ring_buffer_iter {
    struct ring_buffer_per_cpu *cpu_buffer;
    unsigned long               head;       /* 현재 읽기 위치 */
    unsigned long               next_event;
    struct buffer_page          *head_page;
    struct buffer_page          *cache_reader_page;
    unsigned long               cache_read;
    u64                         read_stamp;
    u64                         page_stamp;
    struct ring_buffer_event    *event;
    int                         missed_events;
};

splice 인터페이스

trace_pipe를 통한 대량 데이터 전송에서 splice 시스템 콜은 zero-copy 전송을 가능하게 합니다. ring buffer의 ring_buffer_read_page가 전체 buffer_page를 파이프 버퍼에 직접 전달합니다.

/* splice 읽기 경로 (간략화) */
static ssize_t
tracing_buffers_splice_read(struct file *filp,
                           loff_t *ppos,
                           struct pipe_inode_info *pipe,
                           size_t len,
                           unsigned int flags)
{
    struct ftrace_buffer_info *info = filp->private_data;
    struct trace_buffer *buffer = info->buffer;

    /* 페이지 단위로 ring buffer에서 읽기 */
    ref = ring_buffer_alloc_read_page(buffer, cpu);
    ring_buffer_read_page(buffer, ref, len, cpu, 1);

    /* 읽은 페이지를 pipe 버퍼에 직접 매핑 */
    spd.pages[i] = ref->page;
    spd.partial[i].len = PAGE_SIZE;

    /* splice_to_pipe로 사용자 공간에 전달 */
    return splice_to_pipe(pipe, &spd);
}

splice의 장점은 커널 버퍼 페이지를 사용자 공간으로 직접 매핑(Mapping)하여 memcpy를 피하는 것입니다. 대량 트레이스 데이터를 파일로 저장할 때 특히 효과적입니다.

# splice를 활용한 고속 트레이스 수집
cat /sys/kernel/tracing/trace_pipe > /tmp/trace.log &

# 또는 trace-cmd로 splice 기반 수집
trace-cmd record -e sched_switch -o /tmp/trace.dat

splice vs 일반 read 성능 비교

/* ring_buffer_read_page: splice용 전체 페이지 읽기 */
int ring_buffer_read_page(struct trace_buffer *buffer,
                         struct buffer_data_read_page *data_page,
                         size_t len, int cpu, int full)
{
    struct ring_buffer_per_cpu *cpu_buffer;
    struct buffer_page *reader;
    unsigned long flags;

    cpu_buffer = buffer->buffers[cpu];
    raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);

    reader = rb_get_reader_page(cpu_buffer);
    if (!reader)
        goto out_unlock;

    /* 페이지 내 commit된 데이터가 있으면 */
    if (rb_page_commit(reader) >= len || !full) {
        /* reader_page의 데이터를 data_page로 교환 */
        swap(reader->page, data_page->data);
        /* 새 빈 페이지가 reader_page에 들어감 */
    }

out_unlock:
    raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);
    return ret;
}

mmap 인터페이스

Linux 6.6부터 ring buffer는 mmap 인터페이스를 지원합니다. 사용자 공간에서 시스템 콜 없이 직접 이벤트를 소비할 수 있어, 고빈도 트레이싱의 오버헤드를 더욱 줄입니다.

/* 사용자 공간에서 mmap 기반 이벤트 소비 (예시) */
int fd = open("/sys/kernel/tracing/per_cpu/cpu0/trace_pipe_raw",
             O_RDONLY);

/* mmap: meta page + 서브버퍼 배열 */
size_t meta_len = PAGE_SIZE;
size_t data_len = nr_subbufs * subbuf_size;
void *mmap_addr = mmap(NULL, meta_len + data_len,
                       PROT_READ, MAP_SHARED, fd, 0);

struct ring_buffer_meta *meta = mmap_addr;
void *data_start = mmap_addr + meta_len;

/* reader 서브버퍼에서 이벤트 소비 */
while (1) {
    int reader_id = meta->reader.id;
    void *subbuf = data_start + reader_id * subbuf_size;

    /* 서브버퍼 내 이벤트 순회 */
    process_events_in_subbuf(subbuf, meta->reader.read);

    /* 다음 서브버퍼 요청 */
    ioctl(fd, TRACE_MMAP_IOCTL_GET_READER);
}

타임스탬프 메커니즘

ring buffer의 타임스탬프 시스템은 공간 효율과 정밀도를 모두 달성하기 위한 정교한 설계를 가지고 있습니다.

/* 타임스탬프 delta 계산 */
static inline void rb_update_event(
    struct ring_buffer_per_cpu *cpu_buffer,
    struct ring_buffer_event *event,
    struct rb_event_info *info)
{
    u64 delta = info->delta;

    /* 27비트 이내: 헤더에 직접 저장 */
    if (delta < (1ULL << 27)) {
        event->time_delta = delta;
    } else {
        /* 27비트 초과: TIME_EXTEND 이벤트 삽입 */
        event->type_len = RINGBUF_TYPE_TIME_EXTEND;
        event->time_delta = delta & ((1ULL << 27) - 1);
        event->array[0] = delta >> 27;
    }
}

절대 타임스탬프 (TIME_STAMP)

RINGBUF_TYPE_TIME_STAMP(type_len=31)은 절대 타임스탬프 재동기화에 사용됩니다. 주로 다음 상황에서 삽입됩니다.

• 새 페이지의 첫 이벤트 (페이지 전환 시)
• NMI 중첩 후 타임스탬프 일관성 복구
• 명시적 동기화 마커 요청 시

덮어쓰기 vs 폐기 모드

ring buffer가 가득 찼을 때 새 이벤트를 어떻게 처리할지 두 가지 정책이 있습니다.

/* overwrite 모드에서 tail이 head를 따라잡았을 때 */
static void rb_handle_head_page(
    struct ring_buffer_per_cpu *cpu_buffer,
    struct buffer_page *tail_page,
    struct buffer_page *next_page)
{
    /* overwrite 모드: head_page를 다음으로 전진 */
    if (cpu_buffer->buffer->flags & RB_FL_OVERWRITE) {
        /* 가장 오래된 페이지의 이벤트 수를 overrun에 추가 */
        local_add(rb_page_entries(next_page),
                  &cpu_buffer->overrun);

        /* head_page를 한 칸 전진 */
        rb_inc_page(&cpu_buffer->head_page);
    } else {
        /* discard 모드: 이벤트 기록 거부 */
        local_inc(&cpu_buffer->dropped_events);
    }
}

# overwrite 모드 제어
echo "overwrite" > /sys/kernel/tracing/trace_options     # 덮어쓰기 활성화 (기본)
echo "nooverwrite" > /sys/kernel/tracing/trace_options   # 폐기 모드

# 이벤트 손실 확인
cat /sys/kernel/tracing/per_cpu/cpu0/stats
# entries: 12345
# overrun: 678          ← overwrite 모드에서 덮어쓴 수
# commit overrun: 0
# dropped events: 0     ← discard 모드에서 폐기한 수

덮어쓰기 내부 메커니즘

overwrite 모드에서 tail_page가 head_page를 따라잡으면, head_page의 데이터를 포기하고 한 칸 전진시킵니다. 이때 reader_page와의 충돌도 고려해야 합니다.

/* rb_move_tail에서 overwrite 처리 (간략화) */
static noinline struct ring_buffer_event *
rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
             unsigned long tail,
             struct buffer_page *tail_page)
{
    struct buffer_page *next_page = tail_page->list.next;

    /* tail_page에 PADDING 이벤트 삽입 (남은 공간 표시) */
    rb_event_set_padding(event);
    local_sub(length, &cpu_buffer->entries_bytes);

    /* 다음 페이지가 head_page인가? (버퍼 가득 참) */
    if (next_page == cpu_buffer->head_page) {
        if (!(cpu_buffer->buffer->flags & RB_FL_OVERWRITE)) {
            /* discard 모드: 이벤트 거부 */
            local_inc(&cpu_buffer->dropped_events);
            return NULL;
        }

        /* overwrite 모드: head_page를 밀어내기 */
        rb_handle_head_page(cpu_buffer, tail_page, next_page);
    }

    /* tail_page를 next_page로 전진 */
    rb_tail_page_update(cpu_buffer, tail_page, next_page);

    /* 새 tail_page에서 이벤트 예약 재시도 */
    return rb_reserve_next_event(cpu_buffer, length);
}

overwrite와 reader 충돌

overwrite 모드에서 head_page를 전진시킬 때, reader가 이미 해당 페이지를 분리(swap)했을 수 있습니다. 이 충돌은 RB_PAGE_HEAD 플래그로 감지합니다.

버퍼 가득 참 알림

discard 모드에서 버퍼가 가득 차면, 대기 중인 reader에게 wakeup 신호를 보내 소비를 촉진합니다.

/* irq_work 기반 reader wakeup */
static void rb_wakeups(struct trace_buffer *buffer,
                       struct ring_buffer_per_cpu *cpu_buffer)
{
    /* full 상태이면 waiters 깨우기 */
    if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
        return;

    /* irq_work로 안전하게 wakeup (NMI 컨텍스트에서도 동작) */
    if (waitqueue_active(&buffer->irq_work.waiters))
        irq_work_queue(&buffer->irq_work.work);

    if (waitqueue_active(&cpu_buffer->irq_work.waiters))
        irq_work_queue(&cpu_buffer->irq_work.work);
}

리사이즈와 스냅샷

ring buffer는 런타임에 크기를 동적으로 조정할 수 있으며, 특정 시점의 버퍼 상태를 스냅샷으로 저장할 수 있습니다.

동적 리사이즈

/* 버퍼 크기 변경 */
int ring_buffer_resize(struct trace_buffer *buffer,
                      unsigned long size, int cpu)
{
    /* 새 페이지 수 계산 */
    unsigned long nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);

    /* 증가: 새 페이지를 할당하여 링에 삽입 */
    /* 감소: 여분 페이지를 링에서 제거하고 해제 */

    /* 축소 시 트레이싱 일시 정지 필요 */
    atomic_inc(&cpu_buffer->resize_disabled);
    schedule_work_on(cpu, &cpu_buffer->update_pages_work);

    /* ... 워커에서 실제 리사이즈 수행 ... */
}

# 버퍼 크기 조정 (KB 단위)
echo 16384 > /sys/kernel/tracing/buffer_size_kb       # 전체 CPU 16MB
echo 32768 > /sys/kernel/tracing/per_cpu/cpu0/buffer_size_kb  # CPU0만 32MB

# 현재 크기 확인
cat /sys/kernel/tracing/buffer_size_kb

스냅샷

스냅샷은 현재 ring buffer의 상태를 순간적으로 저장하는 기능입니다. 구현은 매우 효율적입니다 -- 별도의 ring buffer 인스턴스(snapshot buffer)와 현재 버퍼의 페이지 배열을 swap하는 것입니다.

/* 스냅샷 = 두 ring buffer의 swap */
void ring_buffer_swap(struct trace_buffer *buffer_a,
                     struct trace_buffer *buffer_b)
{
    for_each_buffer_cpu(buffer_a, cpu) {
        /* 각 CPU의 페이지 리스트를 교환 */
        swap(buffer_a->buffers[cpu]->pages,
             buffer_b->buffers[cpu]->pages);
        swap(buffer_a->buffers[cpu]->head_page,
             buffer_b->buffers[cpu]->head_page);
        /* ... 나머지 포인터도 swap ... */
    }
}

# 스냅샷 활용
echo 1 > /sys/kernel/tracing/snapshot      # 현재 상태 저장
cat /sys/kernel/tracing/snapshot             # 스냅샷 읽기
echo 0 > /sys/kernel/tracing/snapshot      # 스냅샷 해제

# trace-cmd에서 스냅샷
trace-cmd snapshot

스냅샷 트리거

수동 스냅샷 외에, 특정 조건이 만족되면 자동으로 스냅샷을 찍는 트리거를 설정할 수 있습니다.

# 조건부 스냅샷: sched_switch에서 특정 PID 감지 시
echo 'snapshot if prev_pid == 1234' > \
    /sys/kernel/tracing/events/sched/sched_switch/trigger

# stacktrace + snapshot 결합
echo 'snapshot:stacktrace if next_comm ~ "bad_proc"' > \
    /sys/kernel/tracing/events/sched/sched_switch/trigger

# 지연 임계값 초과 시 자동 스냅샷 (irqsoff tracer)
echo 1000 > /sys/kernel/tracing/tracing_thresh   # 1000us 임계값
echo irqsoff > /sys/kernel/tracing/current_tracer

# 스냅샷 확인
cat /sys/kernel/tracing/snapshot

리사이즈 제약사항

kfifo 비교

커널에서 순환 버퍼로 널리 사용되는 kfifo와 ring buffer의 구조적 차이를 비교합니다.

사용 시점 가이드

kfifo 내부 구현 비교

/* kfifo: 평탄 버퍼 + power-of-2 마스크 트릭 */
struct __kfifo {
    unsigned int    in;     /* 쓰기 인덱스 (단조 증가) */
    unsigned int    out;    /* 읽기 인덱스 (단조 증가) */
    unsigned int    mask;   /* size - 1 (power-of-2) */
    unsigned int    esize;  /* 요소 크기 */
    void            *data;  /* 연속 버퍼 */
};

/* kfifo_in: 단일 생산자 lock-free 쓰기 */
static inline unsigned int
__kfifo_in(struct __kfifo *fifo,
           const void *buf, unsigned int len)
{
    unsigned int off = fifo->in & fifo->mask;
    memcpy(fifo->data + off, buf, len);
    smp_wmb();   /* 데이터 쓰기 → in 갱신 순서 보장 */
    fifo->in += len;
    return len;
}

/* ring_buffer_lock_reserve: per-CPU lockless + NMI-safe */
/* → 가변 길이, 타임스탬프, 중첩 지원, splice/mmap */
/* kfifo보다 복잡하지만, 트레이싱 요구사항에 필수적 */

BPF ringbuf 비교

커널에는 ring buffer 구현이 세 가지 존재합니다. 각각의 설계 목표가 다릅니다.

ftrace 통합

ring buffer는 ftrace 인프라의 핵심 저장소로, 다양한 tracer와 trace event가 이벤트를 기록하는 단일 경로를 제공합니다.

Trace Event 기록 경로

/* TRACE_EVENT(sched_switch, ...) 가 생성하는 기록 함수 (간략화) */
static void
trace_event_raw_event_sched_switch(void *__data,
    bool preempt, struct task_struct *prev,
    struct task_struct *next, unsigned int prev_state)
{
    struct trace_event_buffer fbuffer;
    struct trace_event_raw_sched_switch *entry;
    int __data_size;

    /* 1. ring buffer에서 공간 예약 */
    entry = trace_event_buffer_reserve(&fbuffer, ...,
                sizeof(*entry) + __data_size);
    if (!entry)
        return;

    /* 2. 이벤트 페이로드 채우기 */
    entry->prev_pid   = prev->pid;
    entry->prev_prio  = prev->prio;
    entry->prev_state = prev_state;
    entry->next_pid   = next->pid;
    entry->next_prio  = next->prio;
    memcpy(entry->prev_comm, prev->comm, TASK_COMM_LEN);
    memcpy(entry->next_comm, next->comm, TASK_COMM_LEN);

    /* 3. commit */
    trace_event_buffer_commit(&fbuffer);
}

트레이싱 인스턴스와 ring buffer

# 독립 인스턴스 생성 (별도 ring buffer 할당)
mkdir /sys/kernel/tracing/instances/my_trace
mkdir /sys/kernel/tracing/instances/net_trace

# 각 인스턴스에 별도 설정
echo 8192 > /sys/kernel/tracing/instances/my_trace/buffer_size_kb
echo sched_switch > /sys/kernel/tracing/instances/my_trace/set_event

echo 4096 > /sys/kernel/tracing/instances/net_trace/buffer_size_kb
echo net:* > /sys/kernel/tracing/instances/net_trace/set_event

# 독립적으로 읽기
cat /sys/kernel/tracing/instances/my_trace/trace_pipe &
cat /sys/kernel/tracing/instances/net_trace/trace_pipe &

# 인스턴스 삭제 (ring buffer 해제)
rmdir /sys/kernel/tracing/instances/my_trace

function tracer의 ring buffer 사용

/* function tracer 기록 함수 (간략화) */
static void
function_trace_call(unsigned long ip,
                    unsigned long parent_ip,
                    struct ftrace_ops *ops,
                    struct ftrace_regs *fregs)
{
    struct trace_array *tr = ops->private;
    struct trace_buffer *buffer;
    struct ring_buffer_event *event;
    struct ftrace_entry *entry;
    int pc;

    pc = preempt_count();
    buffer = tr->array_buffer.buffer;

    /* ring buffer에서 공간 예약 */
    event = trace_buffer_lock_reserve(buffer,
                TRACE_FN, sizeof(*entry), pc);
    if (!event)
        return;

    /* 이벤트 데이터 채우기 */
    entry = ring_buffer_event_data(event);
    entry->ip         = ip;          /* 호출된 함수 주소 */
    entry->parent_ip  = parent_ip;   /* 호출자 주소 */

    /* commit */
    ring_buffer_unlock_commit(buffer);
}

/*
 * function_graph tracer는 진입/탈출 2개 이벤트 기록:
 * - TRACE_GRAPH_ENT: 함수 진입 (ip, depth)
 * - TRACE_GRAPH_RET: 함수 탈출 (ip, depth, calltime, rettime)
 * → 함수 실행 시간 = rettime - calltime
 */

히스토그램 트리거와 ring buffer

ftrace의 히스토그램 트리거(hist trigger)는 ring buffer의 이벤트를 소비하지 않고, 쓰기 경로에서 직접 집계합니다. 이는 ring buffer의 오버헤드를 피하면서 통계를 수집하는 효율적인 방법입니다.

# 히스토그램 트리거: sched_switch의 prev_state별 카운트
echo 'hist:keys=prev_state:sort=hitcount' > \
    /sys/kernel/tracing/events/sched/sched_switch/trigger

# 결과 확인 (ring buffer 소비 없이 집계)
cat /sys/kernel/tracing/events/sched/sched_switch/hist

# 지연 분포 히스토그램
echo 'hist:keys=common_pid:vals=hitcount:sort=hitcount.descending' > \
    /sys/kernel/tracing/events/sched/sched_wakeup_new/trigger

perf 링 버퍼

perf_event 서브시스템은 ftrace의 ring buffer와는 완전히 다른 별도의 링 버퍼를 사용합니다. 두 구현의 설계 철학과 구조를 비교합니다.

perf mmap 프로토콜

perf 링 버퍼의 사용자 공간 프로토콜은 ring buffer와 다릅니다. perf는 data_head/data_tail 두 인덱스로 생산자-소비자를 조율합니다.

/* perf_event mmap 프로토콜 (사용자 공간 소비) */
struct perf_event_mmap_page *header;
void *data;

/* mmap: 메타 페이지 + 데이터 영역 */
header = mmap(NULL, (1 + 16) * PAGE_SIZE,
              PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
data = (void *)header + PAGE_SIZE;

/* 이벤트 소비 루프 */
while (1) {
    __u64 head = __atomic_load_n(&header->data_head,
                                  __ATOMIC_ACQUIRE);
    __u64 tail = header->data_tail;

    while (tail != head) {
        struct perf_event_header *ev =
            data + (tail & (data_size - 1));
        process_perf_event(ev);
        tail += ev->size;
    }

    /* 소비 완료 알림 */
    __atomic_store_n(&header->data_tail, tail,
                      __ATOMIC_RELEASE);
}

aux_buffer (Intel PT, ARM CoreSight)

perf 링 버퍼는 하드웨어 트레이스 장치(Intel PT, ARM CoreSight)를 위한 aux_buffer를 별도로 제공합니다. 이 영역은 하드웨어가 직접 쓰기하므로 소프트웨어 오버헤드가 없습니다. ftrace ring buffer에는 이에 대응하는 기능이 없습니다.

# Intel PT 수집 (aux_buffer 사용)
perf record -e intel_pt// -a --aux-size=16M -- sleep 5
perf script --itrace=b

# ARM CoreSight 수집
perf record -e cs_etm/@tmc_etr0/ -a -- sleep 5

성능 특성

ring buffer의 성능은 트레이싱 시스템 전체의 오버헤드를 결정합니다. 주요 성능 특성을 분석합니다.

쓰기 오버헤드

확장성

per-CPU 설계로 CPU 간 경합이 없어 이론적으로 완벽한 선형 확장성을 보입니다. 실제로는 메모리 대역폭(Bandwidth)이 병목(Bottleneck)이 될 수 있습니다.

캐시라인 동작

ring buffer의 성능에서 캐시라인 분리가 중요한 역할을 합니다.

메모리 대역폭과 버퍼 크기

per-CPU 설계로 CPU 간 경합은 없지만, 대량 이벤트 기록 시 메모리 대역폭이 병목이 될 수 있습니다. 특히 function tracer처럼 모든 함수 호출을 기록하는 경우, 각 이벤트는 약 28바이트이고 함수 호출 빈도는 초당 수백만 건에 달합니다.

메모리 대역폭 계산 예시:

• function tracer 이벤트 크기: ~28바이트
• 평균 함수 호출 빈도: ~5M/초/CPU
• 쓰기 대역폭: 28 × 5M = 140 MB/초/CPU
• 64 코어 시스템: 140 × 64 = 8.96 GB/초
• DDR4-3200 단일 채널: ~25.6 GB/초

메모리 컨트롤러 대역폭의 약 35%를 소비하므로, set_ftrace_filter를 통한 함수 필터링으로 완화가 필수입니다. 버퍼가 L3 캐시에 들어가는 크기라면 메모리 트래픽이 대폭 감소합니다 (예: 2MB/CPU 버퍼 + 8MB L3 슬라이스 → 대부분 캐시 히트).

트레이싱 지연 영향

NUMA 고려사항

NUMA 시스템에서 ring buffer의 페이지가 원격 노드에 할당되면 쓰기 지연이 증가합니다. ring buffer는 alloc_pages_node를 사용하여 각 CPU의 로컬 노드에서 페이지를 할당합니다.

/* NUMA-aware 페이지 할당 */
static struct buffer_page *
rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,
                  unsigned long nr_pages)
{
    struct buffer_page *bpage;
    struct list_head *pages = &cpu_buffer->new_pages;
    unsigned long i;

    for (i = 0; i < nr_pages; i++) {
        bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),
                             GFP_KERNEL,
                             cpu_to_node(cpu_buffer->cpu));
        if (!bpage)
            goto free_pages;

        /* 데이터 페이지도 로컬 노드에서 할당 */
        bpage->page = alloc_pages_node(
            cpu_to_node(cpu_buffer->cpu),
            GFP_KERNEL | __GFP_ZERO, 0);
        if (!bpage->page)
            goto free_pages;

        list_add_tail(&bpage->list, pages);
    }
    return pages;
}

튜닝 & 디버깅

ring buffer를 효과적으로 활용하기 위한 실전 튜닝과 디버깅 방법을 정리합니다.

버퍼 크기 튜닝

# 현재 버퍼 크기 확인
cat /sys/kernel/tracing/buffer_size_kb
# 1408  (기본값, 약 1.4MB per CPU)

# 전체 CPU 버퍼 크기 설정
echo 16384 > /sys/kernel/tracing/buffer_size_kb    # 16MB per CPU

# CPU별 개별 설정
echo 65536 > /sys/kernel/tracing/per_cpu/cpu0/buffer_size_kb

# 전체 통계 확인
for cpu in /sys/kernel/tracing/per_cpu/cpu*; do
    echo "=== $(basename $cpu) ==="
    cat "$cpu/stats"
done

# 버퍼 리셋 (모든 이벤트 삭제)
echo > /sys/kernel/tracing/trace

실전 설정 가이드

이벤트 손실 진단

# 이벤트 손실 상세 진단
cat /sys/kernel/tracing/per_cpu/cpu0/stats
# entries: 45123        ← 현재 버퍼에 있는 이벤트 수
# overrun: 12345        ← overwrite로 덮어쓴 이벤트 수
# commit overrun: 0     ← NMI 중첩 commit 충돌 (보통 0)
# bytes: 3612800        ← 총 기록 바이트
# oldest event ts: 123456789.012345
# now ts: 123456890.012345
# dropped events: 0     ← discard 모드에서 폐기된 이벤트
# read events: 30000    ← 소비된 이벤트 수

# 버퍼 사용률 확인
for cpu in /sys/kernel/tracing/per_cpu/cpu*; do
    entries=$(grep "entries:" "$cpu/stats" | awk '{print $2}')
    overrun=$(grep "overrun:" "$cpu/stats" | awk '{print $2}')
    echo "$(basename $cpu): entries=$entries overrun=$overrun"
done

이벤트 필터링으로 오버헤드 감소

# 이벤트 필터: 특정 조건만 기록하여 ring buffer 부하 감소

# sched_switch에서 특정 PID만 기록
echo 'prev_pid == 1234 || next_pid == 1234' > \
    /sys/kernel/tracing/events/sched/sched_switch/filter

# block 이벤트에서 특정 디바이스만
echo 'dev == 0x800010' > \
    /sys/kernel/tracing/events/block/block_rq_issue/filter

# function tracer: 특정 함수만 트레이싱
echo 'do_sys_open' > /sys/kernel/tracing/set_ftrace_filter
echo 'ext4_*' >> /sys/kernel/tracing/set_ftrace_filter

# function_graph: 깊이 제한
echo 5 > /sys/kernel/tracing/max_graph_depth

# 필터 해제
echo 0 > /sys/kernel/tracing/events/sched/sched_switch/filter

per_cpu 디렉토리 구조

# per-CPU 디렉토리 구조
ls /sys/kernel/tracing/per_cpu/cpu0/
# buffer_size_kb   ← 이 CPU 버퍼 크기
# stats            ← entries, overrun, dropped 등 통계
# trace            ← 이 CPU만의 트레이스 (비소비)
# trace_pipe       ← 이 CPU만의 실시간 스트림 (소비)
# trace_pipe_raw   ← 이 CPU 바이너리 스트림 (splice/mmap)
# snapshot         ← 이 CPU 스냅샷

# CPU별 독립 모니터링 예시
# 터미널 1: CPU 0 모니터링
cat /sys/kernel/tracing/per_cpu/cpu0/trace_pipe

# 터미널 2: CPU 1 모니터링
cat /sys/kernel/tracing/per_cpu/cpu1/trace_pipe

# CPU별 버퍼 크기 차등 설정
# 네트워크 인터럽트 처리 CPU에 큰 버퍼
echo 32768 > /sys/kernel/tracing/per_cpu/cpu0/buffer_size_kb
echo 4096 > /sys/kernel/tracing/per_cpu/cpu1/buffer_size_kb

자주 발생하는 문제와 해결

trace-cmd 활용 예제

# trace-cmd으로 이벤트 수집 (splice 기반, 손실 최소화)
trace-cmd record -e sched_switch -e sched_wakeup -b 16384

# 특정 CPU만 수집
trace-cmd record -e irq -C 0,1 -b 8192

# 함수 트레이싱 + 스택 추적
trace-cmd record -p function_graph -g do_sys_open --max-graph-depth 5

# 수집 결과 분석
trace-cmd report trace.dat | head -50

# 실시간 모니터링
trace-cmd stream -e sched_switch

커널 설정

ring buffer와 관련된 커널 설정 옵션을 정리합니다.

# 현재 커널의 ring buffer 관련 설정 확인
zcat /proc/config.gz | grep -i ring_buffer
# CONFIG_RING_BUFFER=y
# CONFIG_RING_BUFFER_ALLOW_SWAP=y
# # CONFIG_RING_BUFFER_BENCHMARK is not set
# # CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS is not set

# ring buffer 벤치마크 실행 (CONFIG_RING_BUFFER_BENCHMARK=m 필요)
modprobe ring_buffer_benchmark
# dmesg에서 결과 확인
dmesg | grep ring_buffer_benchmark

벤치마크 모듈

/* kernel/trace/ring_buffer_benchmark.c (발췌) */

/* 벤치마크 파라미터 */
static int write_iteration = 50;
module_param(write_iteration, int, 0644);
MODULE_PARM_DESC(write_iteration,
    "# of writes between timestamp readings");

static int producer_nice = MAX_NICE;
module_param(producer_nice, int, 0644);

static int consumer_nice = MAX_NICE;
module_param(consumer_nice, int, 0644);

/*
 * 벤치마크 결과 예시 (dmesg):
 *
 * ring_buffer_benchmark: Running ring buffer benchmark
 * ring_buffer_benchmark: Entries:   1000000
 * ring_buffer_benchmark: Total:     125 ms
 * ring_buffer_benchmark: Overruns:  0
 * ring_buffer_benchmark: Read:      1000000
 * ring_buffer_benchmark: Avg Write: 50 ns
 * ring_buffer_benchmark: Avg Read:  80 ns
 */

임베디드 시스템 최소 설정

# 최소 트레이싱 (ring buffer + trace events만)
CONFIG_RING_BUFFER=y
CONFIG_TRACING=y
CONFIG_TRACING_SUPPORT=y
CONFIG_EVENT_TRACING=y

# function tracer 비활성화 (mcount 오버헤드 제거)
# CONFIG_FUNCTION_TRACER is not set
# CONFIG_FUNCTION_GRAPH_TRACER is not set

# 스냅샷 비활성화 (메모리 절약)
# CONFIG_RING_BUFFER_ALLOW_SWAP is not set

# 디버그 비활성화 (오버헤드 제거)
# CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS is not set
# CONFIG_RING_BUFFER_BENCHMARK is not set

디버그 설정

# ring buffer 디버그 전체 활성화
CONFIG_RING_BUFFER=y
CONFIG_RING_BUFFER_VALIDATE_TIME_DELTAS=y  # 타임스탬프 검증
CONFIG_RING_BUFFER_BENCHMARK=m             # 벤치마크 모듈
CONFIG_RING_BUFFER_ALLOW_SWAP=y            # 스냅샷

# 관련 디버그 옵션
CONFIG_TRACING=y
CONFIG_FTRACE=y
CONFIG_FUNCTION_TRACER=y
CONFIG_FUNCTION_GRAPH_TRACER=y
CONFIG_DYNAMIC_FTRACE=y
CONFIG_FTRACE_SYSCALLS=y
CONFIG_TRACER_MAX_TRACE=y                  # 최대 지연 추적
CONFIG_TRACER_SNAPSHOT=y                   # 스냅샷 인터페이스
CONFIG_TRACER_SNAPSHOT_PER_CPU_SWAP=y      # CPU별 스냅샷
CONFIG_HIST_TRIGGERS=y                     # 히스토그램 트리거

부트 파라미터

# 부팅 초기 이벤트 캡처 (grub 커맨드라인)
linux /vmlinuz trace_buf_size=64M \
    trace_event=initcall:initcall_start,initcall:initcall_finish \
    ftrace_dump_on_oops=orig_cpu

커널 소스 분석: 쓰기 Fast Path

ring_buffer_write()와 rb_reserve_next_event()의 fast path를 커널 소스 수준에서 상세히 분석합니다. 이 경로는 트레이싱 오버헤드의 대부분을 결정하므로 최적화가 매우 중요합니다.

ring_buffer_write() 진입점(Entry Point)

ring_buffer_write()는 lock reserve와 commit을 하나로 결합한 편의 API입니다. 내부적으로 ring_buffer_lock_reserve()와 동일한 fast path를 거칩니다.

/* kernel/trace/ring_buffer.c — ring_buffer_write() 전체 흐름 */
int ring_buffer_write(struct trace_buffer *buffer,
                     unsigned long length,
                     void *data)
{
    struct ring_buffer_per_cpu *cpu_buffer;
    struct ring_buffer_event *event;
    int ret;

    /* 1. preemption 비활성화 — CPU 이동 방지 */
    preempt_disable_notrace();

    cpu_buffer = buffer->buffers[raw_smp_processor_id()];

    /* 2. 쓰기 비활성화 상태 확인 (iterator 사용 중이면 실패) */
    if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
        goto out;

    /* 3. 이벤트 공간 예약 (핵심 fast path) */
    event = rb_reserve_next_event(buffer, cpu_buffer, length);
    if (!event)
        goto out;

    /* 4. 페이로드 복사 */
    memcpy(rb_event_data(event), data, length);

    /* 5. commit 수행 */
    rb_commit(cpu_buffer);
    rb_wakeups(buffer, cpu_buffer);
    ret = 0;
out:
    preempt_enable_notrace();
    return ret;
}

rb_reserve_next_event() fast path 심층 분석

fast path에서 가장 중요한 것은 단일 local_cmpxchg로 공간 예약이 완료되는 것입니다. 이 경로에서 lock은 전혀 사용되지 않습니다.

/* rb_reserve_next_event() — fast path 상세 (kernel/trace/ring_buffer.c) */
static noinline struct ring_buffer_event *
rb_reserve_next_event(struct trace_buffer *buffer,
                      struct ring_buffer_per_cpu *cpu_buffer,
                      unsigned long length)
{
    struct ring_buffer_event *event;
    struct rb_event_info info;
    int nr_loops = 0;

    /* 이벤트 헤더 크기 추가 + 4바이트 정렬 */
    length = rb_calculate_event_length(length);

    /* 타임스탬프 획득 — TSC 또는 trace_clock */
    info.ts = rb_time_stamp(cpu_buffer);
    info.delta = info.ts - cpu_buffer->write_stamp;

    /* 중첩 카운터 증가 (NMI/IRQ 추적) */
    local_inc(&cpu_buffer->committing);

again:
    /* 무한 루프 방지 — 최대 1000회 재시도 */
    if (unlikely(++nr_loops > 1000)) {
        RB_WARN_ON(cpu_buffer, 1);
        goto out_fail;
    }

    info.tail_page = cpu_buffer->tail_page;
    info.tail = local_read(&info.tail_page->write);
    info.write = info.tail + length;

    /* ── Fast Path: 현재 페이지에 공간이 충분한 경우 ── */
    if (likely(info.write <= BUF_PAGE_SIZE)) {
        /* local_cmpxchg: 같은 CPU 내에서만 원자적
         * x86: LOCK prefix 없는 CMPXCHG → 매우 빠름
         * IRQ/NMI 중첩 시에만 CAS 실패 → goto again */
        if (local_cmpxchg(&info.tail_page->write,
                          info.tail, info.write) != info.tail)
            goto again;

        /* 공간 예약 성공 — 이벤트 포인터 계산 */
        event = __rb_page_index(info.tail_page, info.tail);

        /* 이벤트 헤더 초기화 (type_len, time_delta) */
        rb_update_event(cpu_buffer, event, &info);

        /* write_stamp 갱신 (다음 이벤트의 delta 계산용) */
        cpu_buffer->write_stamp = info.ts;
        local_inc(&info.tail_page->entries);

        return event;
    }

    /* ── Slow Path: 페이지 전환 필요 ── */
    event = rb_move_tail(cpu_buffer, &info);
    if (unlikely(IS_ERR(event)))
        goto out_fail;
    if (!event)
        goto again;  /* rb_move_tail이 페이지 전환 후 재시도 요청 */

    return event;

out_fail:
    local_dec(&cpu_buffer->committing);
    return NULL;
}

rb_move_tail() slow path 분석

rb_move_tail()은 현재 페이지에 공간이 부족할 때 호출됩니다. 이 함수는 현재 페이지 끝에 PADDING을 삽입하고 tail_page를 다음 페이지로 전진시킵니다.

/* rb_move_tail() — slow path: 페이지 전환 (간략화) */
static noinline struct ring_buffer_event *
rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
             struct rb_event_info *info)
{
    struct buffer_page *tail_page = info->tail_page;
    struct buffer_page *next_page;

    next_page = tail_page->list.next;

    /* 현재 페이지의 남은 공간에 PADDING 이벤트 삽입 */
    rb_event_set_padding(tail_page, info->tail);

    /* tail_page → next_page 경합: 다른 컨텍스트가
     * 이미 전진시켰을 수 있으므로 cmpxchg 사용 */
    if (cmpxchg(&cpu_buffer->tail_page,
                tail_page, next_page) != tail_page)
        return NULL;  /* 다른 컨텍스트가 이미 전진 → 재시도 */

    /* 다음 페이지가 head_page인가? (버퍼 가득 참) */
    if (next_page == cpu_buffer->head_page) {
        if (buffer->flags & RB_FL_OVERWRITE) {
            /* Overwrite 모드: head_page를 한 칸 전진
             * → 가장 오래된 페이지의 데이터 손실 */
            rb_inc_page(&cpu_buffer->head_page);
            local_inc(&cpu_buffer->overrun);
            local_sub(local_read(&next_page->entries),
                     &cpu_buffer->entries);
        } else {
            /* Discard 모드: 이벤트 폐기 */
            local_inc(&cpu_buffer->dropped_events);
            return ERR_PTR(-EBUSY);
        }
    }

    /* 새 페이지 초기화 */
    local_set(&next_page->write, 0);
    local_set(&next_page->entries, 0);
    next_page->page->time_stamp = info->ts;

    return NULL;  /* 호출자에게 재시도 요청 */
}

커널 소스 분석: per-CPU 버퍼 및 페이지 관리

ring_buffer_per_cpu 구조체의 내부 상태 관리와 페이지 할당/해제 메커니즘을 소스 레벨에서 추적합니다.

페이지 할당과 ring 구성

/* rb_allocate_pages() — ring buffer 초기화 시 페이지 할당 */
static int rb_allocate_pages(
    struct ring_buffer_per_cpu *cpu_buffer,
    unsigned long nr_pages)
{
    struct list_head *head = &cpu_buffer->pages;
    struct buffer_page *bpage, *tmp;
    LIST_HEAD(pages);
    unsigned long i;

    for (i = 0; i < nr_pages; i++) {
        /* buffer_page 메타데이터 할당 */
        bpage = kzalloc_node(sizeof(*bpage),
                            GFP_KERNEL, cpu_to_node(cpu));

        /* 실제 데이터 페이지 할당 (PAGE_SIZE) */
        bpage->page = rb_alloc_page(cpu_buffer, cpu);

        /* 임시 리스트에 추가 */
        list_add_tail(&bpage->list, &pages);
    }

    /* 임시 리스트를 순환 연결 리스트로 변환 */
    list_for_each_entry_safe(bpage, tmp, &pages, list) {
        list_del_init(&bpage->list);
        rb_link_page(bpage, head);
    }

    /* 포인터 초기화 */
    cpu_buffer->head_page = list_entry(
        cpu_buffer->pages.next, struct buffer_page, list);
    cpu_buffer->tail_page = cpu_buffer->head_page;
    cpu_buffer->commit_page = cpu_buffer->head_page;
    cpu_buffer->nr_pages = nr_pages;

    return 0;
}

/* buffer_data_page — 실제 데이터를 담는 페이지 헤더 */
struct buffer_data_page {
    u64     time_stamp;  /* 이 페이지 첫 이벤트의 절대 타임스탬프 */
    local_t commit;      /* 확정된 바이트 오프셋 */
    unsigned char data[];  /* BUF_PAGE_SIZE 바이트 */
};

/* BUF_PAGE_SIZE 계산 */
#define BUF_PAGE_SIZE (PAGE_SIZE - offsetof( \
    struct buffer_data_page, data))
/* x86 4KB 페이지: 4096 - 16 = 4080 바이트 */

런타임 리사이즈 메커니즘

ring buffer는 실행 중에 페이지를 추가하거나 제거할 수 있습니다. 이 과정에서 쓰기를 잠시 비활성화한 뒤 순환 리스트에 페이지를 삽입/제거합니다.

/* __rb_resize_buffer() — 런타임 리사이즈 핵심 */
static int __rb_resize_buffer(
    struct ring_buffer_per_cpu *cpu_buffer,
    unsigned long new_nr_pages)
{
    unsigned long old_nr_pages = cpu_buffer->nr_pages;

    if (new_nr_pages > old_nr_pages) {
        /* 확장: 새 페이지 할당 후 tail_page 뒤에 삽입 */
        unsigned long delta = new_nr_pages - old_nr_pages;
        rb_insert_pages(cpu_buffer, delta);
    } else if (new_nr_pages < old_nr_pages) {
        /* 축소: head_page 앞의 페이지들을 제거
         * 주의: head_page~tail_page 구간은 제거 불가 */
        unsigned long delta = old_nr_pages - new_nr_pages;
        rb_remove_pages(cpu_buffer, delta);
    }

    cpu_buffer->nr_pages = new_nr_pages;
    return 0;
}

/*
 * 페이지 삽입 위치:
 *
 * [head] → [A] → [B] → [tail] → [NEW1] → [NEW2] → [head]
 *                                 ^^^^^^^^^^^^^^^^^
 *                                 tail_page 뒤에 삽입
 *
 * 삽입 중 쓰기 비활성화로 ring 일관성 보장
 */

커널 소스 분석: 소비자 경로

ring_buffer_read() 및 관련 소비자 함수의 내부 동작을 커널 소스 수준에서 분석합니다.

rb_buffer_peek() 내부 동작

rb_buffer_peek()은 모든 읽기 API의 공통 핵심 함수입니다. reader_page에서 다음 유효한 이벤트를 찾아 반환합니다.

/* rb_buffer_peek() — 다음 유효 이벤트 탐색 */
static struct ring_buffer_event *
rb_buffer_peek(struct ring_buffer_per_cpu *cpu_buffer,
               u64 *ts, unsigned long *lost_events)
{
    struct ring_buffer_event *event;
    struct buffer_page *reader;
    int nr_loops = 0;

again:
    /* 무한 루프 방지 */
    if (++nr_loops > 2)
        return NULL;

    /* reader_page 확보 (소진 시 head_page와 swap) */
    reader = rb_get_reader_page(cpu_buffer);
    if (!reader)
        return NULL;

    /* 현재 읽기 위치의 이벤트 */
    event = rb_reader_event(cpu_buffer);

    /* 특수 이벤트 처리 */
    switch (event->type_len) {
    case RINGBUF_TYPE_PADDING:
        /* 페이지 끝 패딩 → 다음 reader_page로 전환 */
        if (!event->time_delta)
            return NULL;  /* 빈 페이지 */
        rb_advance_reader(cpu_buffer);
        goto again;

    case RINGBUF_TYPE_TIME_EXTEND:
        /* 시간 확장 이벤트 → 건너뛰고 다음 이벤트 */
        cpu_buffer->read_stamp += rb_event_time_delta(event);
        rb_advance_reader(cpu_buffer);
        goto again;

    case RINGBUF_TYPE_TIME_STAMP:
        /* 절대 타임스탬프 → read_stamp 갱신 후 건너뛰기 */
        cpu_buffer->read_stamp = rb_event_time_delta(event);
        rb_advance_reader(cpu_buffer);
        goto again;

    default:
        /* 데이터 이벤트 → 타임스탬프 계산 후 반환 */
        cpu_buffer->read_stamp += event->time_delta;
        if (ts)
            *ts = cpu_buffer->read_stamp;
        if (lost_events)
            *lost_events = local_read(
                &cpu_buffer->lost_events);
        return event;
    }
}

reader_page swap 상세 과정

rb_get_reader_page()의 swap 동작을 단계별로 분석합니다. 이 과정은 쓰기 경로와의 동시성을 처리해야 하므로 cmpxchg를 사용합니다.

/* rb_get_reader_page() — reader_page swap 상세 */
static struct buffer_page *
rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer)
{
    struct buffer_page *reader = cpu_buffer->reader_page;
    struct buffer_page *head;
    unsigned long overwrite;

    /* ① reader_page에 읽을 데이터가 남아있으면 그대로 반환 */
    if (cpu_buffer->read < rb_page_commit(reader))
        return reader;

    /* ② 버퍼에 데이터가 있는지 확인 */
    if (cpu_buffer->commit_page == cpu_buffer->reader_page)
        return NULL;  /* 버퍼 비어있음 */

    /* ③ head_page를 가져옴 */
    head = cpu_buffer->head_page;

    /* ④ overwrite 횟수 기록 (lost_events 추적용) */
    overwrite = local_read(&head->page->commit) -
                local_read(&head->entries);

    /* ⑤ reader_page(빈 페이지)를 head_page 위치에 삽입 */
    reader->list.next = head->list.next;
    reader->list.prev = head->list.prev;

    /* ⑥ 원자적 swap (쓰기 경로와의 경합 방지) */
    if (cmpxchg(&head->list.prev->next,
                &head->list, &reader->list) != &head->list) {
        /* swap 실패: 쓰기 경로가 동시에 수정 중 → 재시도 */
        local_inc(&cpu_buffer->committing);
        return NULL;
    }

    /* ⑦ swap 성공: head를 reader_page로 사용 */
    cpu_buffer->reader_page = head;
    cpu_buffer->read = 0;

    /* ⑧ head_page를 다음 페이지로 전진 */
    rb_inc_page(&cpu_buffer->head_page);

    /* ⑨ lost_events 갱신 */
    local_add(overwrite, &cpu_buffer->lost_events);

    return head;  /* 이전 head_page가 새 reader_page */
}

커널 소스 분석: 타임스탬프 delta 압축

ring buffer의 타임스탬프 처리는 공간 효율성의 핵심입니다. 이벤트마다 64비트 절대 타임스탬프를 저장하는 대신, 이전 이벤트와의 차이(delta)를 27비트로 압축합니다.

delta 인코딩 과정

/* 타임스탬프 delta 계산 및 인코딩 — rb_reserve_next_event() 내부 */

/* 1단계: 현재 타임스탬프 획득 */
u64 ts = rb_time_stamp(cpu_buffer);
/* x86: rdtsc() 기반, ns 단위 변환 */

/* 2단계: delta 계산 */
u64 delta = ts - cpu_buffer->write_stamp;
/* write_stamp: 마지막으로 쓴 이벤트의 절대 타임스탬프 */

/* 3단계: delta 크기에 따른 분기 */
if (delta < (1ULL << RB_EVNT_TIME_SHIFT)) {
    /* Case A: 27비트 이내 (최대 ~134ms)
     * → 이벤트 헤더의 time_delta에 직접 저장
     * → 추가 공간 불필요 (가장 빈번한 경우) */
    event->time_delta = delta;
} else {
    /* Case B: 27비트 초과
     * → TIME_EXTEND 이벤트 삽입 (8바이트 추가) */
    struct ring_buffer_event *ext;

    ext->type_len = RINGBUF_TYPE_TIME_EXTEND;
    ext->time_delta = delta & ((1ULL << 27) - 1);
    ext->array[0] = (u32)(delta >> 27);
    /* 27 + 32 = 59비트 delta → 최대 ~18.3년 */
}

/* 4단계: write_stamp 갱신 */
cpu_buffer->write_stamp = ts;

/*
 * 관련 상수 정의:
 * #define RB_EVNT_TIME_SHIFT  27
 * #define RB_EVNT_TIME_MASK   ((1ULL << RB_EVNT_TIME_SHIFT) - 1)
 *
 * 타임스탬프 복원 공식:
 * absolute_ts = page.time_stamp
 *             + event[0].time_delta
 *             + event[1].time_delta
 *             + ...
 *             + event[n].time_delta
 */

NMI 중첩과 타임스탬프 정합성

NMI가 쓰기 도중 발생하면 타임스탬프 순서가 뒤바뀔 수 있습니다. ring buffer는 이를 before_stamp와 write_stamp의 이중 체크로 감지하고 보정합니다.

/* NMI 중첩 시 타임스탬프 보정 로직 */

/*
 * 시나리오: 일반 컨텍스트가 이벤트 A를 쓰는 도중
 * NMI가 발생하여 이벤트 B를 기록
 *
 * 시간순서: A.ts=100, B.ts=120
 * 하지만 A가 먼저 공간을 예약했으므로 버퍼에는 A가 먼저 위치
 *
 * 해결: commit 시 타임스탬프 일관성 확인
 */

/* 예약 시: before_stamp 기록 */
rb_time_set(&cpu_buffer->before_stamp, info->ts);

/* local_cmpxchg 성공 후: write_stamp 갱신 */
cpu_buffer->write_stamp = info->ts;

/* commit 시: before_stamp != write_stamp이면 중첩 감지 */
if (rb_time_read(&cpu_buffer->before_stamp, &bts) &&
    bts != info->ts) {
    /* 중첩이 발생하여 write_stamp가 변경됨
     * → 다음 이벤트에 TIME_EXTEND 삽입하여 보정
     * → 이벤트의 time_delta를 0으로 설정 */
    info->add_timestamp = RB_ADD_STAMP_EXTEND;
}

페이지 기반 vs 바이트 단위 순환 버퍼

ring buffer의 페이지 기반 설계를 kfifo와 같은 바이트 단위 평탄 버퍼와 비교하여, 각 설계의 장단점과 적합한 사용 사례를 분석합니다.

구조적 비교: 페이지 링 vs 바이트 링

/* 바이트 단위 순환 버퍼 (kfifo 스타일) */
struct byte_ring {
    unsigned char *buffer;  /* 연속 메모리 */
    unsigned int size;      /* 2^n 크기 */
    unsigned int in;        /* 쓰기 오프셋 */
    unsigned int out;       /* 읽기 오프셋 */
};
/* 쓰기: buffer[in & (size-1)] = data
 * 읽기: data = buffer[out & (size-1)]
 * 데이터가 버퍼 끝에서 시작으로 래핑될 수 있음 */

/* 페이지 기반 순환 버퍼 (ftrace ring buffer) */
struct page_ring {
    struct buffer_page *pages;  /* 순환 연결 리스트 */
    struct buffer_page *tail;   /* 쓰기 페이지 */
    struct buffer_page *head;   /* 읽기 페이지 */
    unsigned long nr_pages;
};
/* 이벤트는 항상 단일 페이지 내에서 완결
 * 페이지 경계 넘기지 않음 → 래핑 처리 불필요
 * 대신 페이지 끝에 PADDING 삽입 */

ftrace 이벤트 포맷 레이아웃

ring buffer에 저장되는 이벤트의 실제 메모리 레이아웃을 상세히 분석합니다. ring_buffer_event 헤더와 trace event 페이로드가 어떻게 결합되는지 살펴봅니다.

이벤트 크기 인코딩 상세

/* ring_buffer_event 헤더 — 비트 필드 해석 */
struct ring_buffer_event {
    u32  type_len : 5;    /* 이벤트 타입 + 길이 */
    u32  time_delta : 27;  /* 이전 이벤트 이후 시간 차이 */
    u32  array[];           /* 페이로드 (가변 길이) */
};

/* type_len 해석:
 *
 * 1~28: 데이터 이벤트, 페이로드 = type_len × 4 바이트
 *       총 크기 = 4(헤더) + type_len × 4
 *
 * 0:    대형 데이터 이벤트 (페이로드 > 112 바이트)
 *       array[0] = 실제 페이로드 크기
 *       총 크기 = 4(헤더) + 4(array[0]) + ALIGN(array[0], 4)
 *
 * 29:   PADDING (페이지 끝 빈 공간 채움)
 *       time_delta가 0이면 빈 페이지
 *       아니면 array[0] = 패딩 크기
 *
 * 30:   TIME_EXTEND (긴 시간 간격)
 *       27비트 time_delta + array[0] 32비트 = 59비트
 *
 * 31:   TIME_STAMP (절대 타임스탬프)
 *       전체 64비트 타임스탬프 재설정
 */

/* 이벤트 크기 계산 함수 */
static inline unsigned rb_event_length(
    struct ring_buffer_event *event)
{
    switch (event->type_len) {
    case RINGBUF_TYPE_PADDING:  /* 29 */
        if (event->time_delta)
            return event->array[0] + RB_EVNT_HDR_SIZE;
        return RB_EVNT_HDR_SIZE;

    case RINGBUF_TYPE_TIME_EXTEND:  /* 30 */
    case RINGBUF_TYPE_TIME_STAMP:  /* 31 */
        return RB_EVNT_HDR_SIZE + 4;  /* 8바이트 */

    case 0:  /* 대형 이벤트 */
        return event->array[0] + RB_EVNT_HDR_SIZE + 4;

    default:  /* 1~28: 일반 이벤트 */
        return event->type_len * 4 + RB_EVNT_HDR_SIZE;
    }
}

/* RB_EVNT_HDR_SIZE = offsetof(ring_buffer_event, array) = 4 */

trace_entry 공통 헤더

ring_buffer_event의 페이로드 첫 부분은 trace_entry 공통 헤더입니다. 모든 trace event가 이 헤더를 공유하므로, 이벤트 타입과 실행 컨텍스트를 일관되게 식별할 수 있습니다.

/* trace_entry — 모든 trace event의 공통 헤더 */
struct trace_entry {
    unsigned short  type;      /* trace event 타입 ID */
    unsigned char   flags;     /* TRACE_FLAG_* 비트마스크 */
    unsigned char   preempt_count;  /* 선점 카운트 */
    int             pid;       /* 현재 프로세스 PID */
};

/* trace_entry.flags 비트 */
#define TRACE_FLAG_IRQS_OFF      0x01  /* IRQ 비활성화 상태 */
#define TRACE_FLAG_IRQS_NOSUPPORT 0x02
#define TRACE_FLAG_NEED_RESCHED  0x04  /* 재스케줄 필요 */
#define TRACE_FLAG_HARDIRQ       0x08  /* 하드 IRQ 컨텍스트 */
#define TRACE_FLAG_SOFTIRQ       0x10  /* 소프트 IRQ 컨텍스트 */
#define TRACE_FLAG_NMI           0x20  /* NMI 컨텍스트 */

/*
 * 전체 이벤트 메모리 레이아웃 (sched_switch 예시):
 *
 * ring_buffer_event (4B):
 *   type_len=14, time_delta=500
 *
 * trace_entry (8B):
 *   type=0x1234, flags=0x00, preempt=0, pid=1000
 *
 * trace_event_raw_sched_switch (이벤트 고유 필드):
 *   prev_comm[16], prev_pid, prev_prio, prev_state
 *   next_comm[16], next_pid, next_prio
 *
 * 총: 4 + 8 + 48 = 60 바이트 (4바이트 정렬)
 */

실전 예제: 커스텀 트레이서 구현

ring buffer API를 사용하여 커스텀 커널 모듈(Kernel Module) 트레이서를 구현하는 방법을 단계별로 설명합니다. 이를 통해 ring buffer의 실전 사용법을 익힐 수 있습니다.

기본 커스텀 트레이서 모듈

/* custom_tracer.c — ring buffer API를 사용하는 커스텀 트레이서
 *
 * 이 모듈은 ring buffer의 핵심 API 사용법을 보여줍니다:
 * 1. trace_buffer 할당
 * 2. 이벤트 예약 및 기록
 * 3. 이벤트 소비 및 읽기
 * 4. 버퍼 해제
 */

#include <linux/module.h>
#include <linux/ring_buffer.h>
#include <linux/slab.h>
#include <linux/ktime.h>

struct my_trace_event {
    u64  timestamp;
    u32  cpu;
    u32  pid;
    char msg[32];
};

static struct trace_buffer *my_buffer;

/* 이벤트 기록 함수 */
static void my_trace_write(const char *msg)
{
    struct ring_buffer_event *event;
    struct my_trace_event *entry;

    /* 1단계: ring buffer에서 공간 예약
     *  - preemption 비활성화
     *  - 타임스탬프 자동 획득
     *  - local_cmpxchg로 lockless 예약 */
    event = ring_buffer_lock_reserve(
        my_buffer, sizeof(struct my_trace_event));
    if (!event)
        return;  /* 버퍼 가득 참 (discard 모드) */

    /* 2단계: 이벤트 데이터 채우기
     *  - ring_buffer_event_data()로 페이로드 포인터 획득
     *  - 예약된 공간에 직접 쓰기 (zero-copy) */
    entry = ring_buffer_event_data(event);
    entry->timestamp = ktime_get_ns();
    entry->cpu = raw_smp_processor_id();
    entry->pid = current->pid;
    strscpy(entry->msg, msg, sizeof(entry->msg));

    /* 3단계: commit
     *  - commit 포인터 전진
     *  - preemption 재활성화
     *  - 대기 중인 reader wakeup */
    ring_buffer_unlock_commit(my_buffer);
}

/* 이벤트 소비 함수 */
static void my_trace_read(int cpu)
{
    struct ring_buffer_event *event;
    struct my_trace_event *entry;
    u64 ts;

    while ((event = ring_buffer_consume(
            my_buffer, cpu, &ts, NULL))) {
        entry = ring_buffer_event_data(event);
        pr_info("[%llu] CPU%u PID%u: %s\n",
                ts, entry->cpu, entry->pid, entry->msg);
    }
}

/* 모듈 초기화 */
static int __init my_tracer_init(void)
{
    /* ring buffer 할당: per-CPU 버퍼 자동 생성 */
    my_buffer = ring_buffer_alloc(
        4096 * 1024,  /* 4MB per CPU */
        RB_FL_OVERWRITE);  /* 덮어쓰기 모드 */
    if (!my_buffer)
        return -ENOMEM;

    /* 테스트 이벤트 기록 */
    my_trace_write("tracer initialized");

    pr_info("my_tracer: loaded, buffer=%lu KB/cpu\n",
            ring_buffer_size(my_buffer, 0) / 1024);
    return 0;
}

/* 모듈 종료 */
static void __exit my_tracer_exit(void)
{
    int cpu;

    /* 모든 CPU의 이벤트 소비 */
    for_each_online_cpu(cpu)
        my_trace_read(cpu);

    /* ring buffer 해제 */
    ring_buffer_free(my_buffer);
    pr_info("my_tracer: unloaded\n");
}

module_init(my_tracer_init);
module_exit(my_tracer_exit);
MODULE_LICENSE("GPL");

ftrace 프레임워크에 트레이서 등록

ftrace 프레임워크의 struct tracer를 사용하면 tracefs 인터페이스와 통합되는 정식 트레이서를 구현할 수 있습니다.

/* ftrace 프레임워크 정식 트레이서 등록 (간략화) */
#include <linux/ftrace.h>
#include "trace.h"  /* kernel/trace/trace.h (내부 헤더) */

/* 트레이서 초기화 콜백 */
static int my_tracer_init(struct trace_array *tr)
{
    /* ftrace가 제공하는 ring buffer를 사용 */
    tracing_start_cmdline_record();
    tracer_tracing_on(tr);
    return 0;
}

/* 트레이서 리셋 콜백 */
static void my_tracer_reset(struct trace_array *tr)
{
    tracing_stop_cmdline_record();
    tracer_tracing_off(tr);
}

/* 이벤트 기록 콜백 (function tracer hook) */
static void
my_trace_func(unsigned long ip,
              unsigned long parent_ip,
              struct ftrace_ops *ops,
              struct ftrace_regs *fregs)
{
    struct trace_array *tr = ops->private;
    struct ring_buffer_event *event;
    struct ftrace_entry *entry;
    int pc = preempt_count();

    /* ftrace 전용 ring buffer에 이벤트 기록 */
    event = trace_buffer_lock_reserve(
        tr->array_buffer.buffer,
        TRACE_FN, sizeof(*entry), pc);
    if (!event)
        return;

    entry = ring_buffer_event_data(event);
    entry->ip = ip;
    entry->parent_ip = parent_ip;

    ring_buffer_unlock_commit(
        tr->array_buffer.buffer);
}

/* 트레이서 정의 구조체 */
static struct tracer my_tracer __read_mostly = {
    .name       = "my_tracer",
    .init       = my_tracer_init,
    .reset      = my_tracer_reset,
    /* .print_line: trace 출력 형식 커스터마이즈 */
    /* .flags: 트레이서 옵션 플래그 */
};

/* 트레이서 등록 (부팅 시 또는 모듈 init) */
register_tracer(&my_tracer);

/*
 * 등록 후 사용:
 * echo my_tracer > /sys/kernel/tracing/current_tracer
 * cat /sys/kernel/tracing/trace
 */

TRACE_EVENT 매크로(Macro)를 통한 이벤트 정의

가장 일반적인 ring buffer 활용 방법은 TRACE_EVENT 매크로를 사용하여 구조화된 이벤트를 정의하는 것입니다. 이 매크로는 ring buffer 기록 코드, 출력 포맷, eBPF 연동 코드를 자동 생성합니다.

/* TRACE_EVENT 매크로로 커스텀 이벤트 정의
 * 파일: include/trace/events/my_subsystem.h */

#undef TRACE_SYSTEM
#define TRACE_SYSTEM my_subsystem

#if !defined(_TRACE_MY_SUBSYSTEM_H) || \
    defined(TRACE_HEADER_MULTI_READ)
#define _TRACE_MY_SUBSYSTEM_H

#include <linux/tracepoint.h>

/* 이벤트 정의 */
TRACE_EVENT(my_event,

    /* TP_PROTO: 트레이스포인트 프로토타입 */
    TP_PROTO(const char *name, int value,
             unsigned long addr),

    /* TP_ARGS: 인자 이름 */
    TP_ARGS(name, value, addr),

    /* TP_STRUCT__entry: ring buffer에 저장할 필드
     * → trace_event_raw_my_event 구조체 자동 생성 */
    TP_STRUCT__entry(
        __string(name_field, name)
        __field(int, value)
        __field(unsigned long, addr)
    ),

    /* TP_fast_assign: 이벤트 필드 채우기
     * → ring buffer에 데이터 기록하는 코드 생성 */
    TP_fast_assign(
        __assign_str(name_field, name);
        __entry->value = value;
        __entry->addr = addr;
    ),

    /* TP_printk: trace 파일 출력 형식 */
    TP_printk("name=%s value=%d addr=0x%lx",
              __get_str(name_field),
              __entry->value,
              __entry->addr)
);

#endif /* _TRACE_MY_SUBSYSTEM_H */

#include <trace/define_trace.h>

/*
 * 사용법 (커널 코드 내):
 *
 * #include <trace/events/my_subsystem.h>
 *
 * void my_function(void) {
 *     trace_my_event("test", 42, 0xdeadbeef);
 *     // → ring buffer에 자동으로 이벤트 기록
 * }
 *
 * tracefs에서 사용:
 * echo 1 > /sys/kernel/tracing/events/my_subsystem/my_event/enable
 * cat /sys/kernel/tracing/trace
 * # my_func-1234 [001] .... 1234.567890: my_event: name=test value=42 addr=0xdeadbeef
 */

ftrace 핵심 트레이서의 ring buffer 활용 패턴

# 실전 활용: ring buffer 상태를 확인하며 커스텀 이벤트 수집

# 1. 커스텀 이벤트 활성화
echo 1 > /sys/kernel/tracing/events/my_subsystem/my_event/enable

# 2. 버퍼 상태 모니터링 (오버런 확인)
while true; do
    entries=$(awk '/^entries:/{print $2}' \
        /sys/kernel/tracing/per_cpu/cpu0/stats)
    overrun=$(awk '/^overrun:/{print $2}' \
        /sys/kernel/tracing/per_cpu/cpu0/stats)
    echo "entries=$entries overrun=$overrun"
    sleep 1
done &

# 3. 실시간 이벤트 스트림 (소비 모드)
cat /sys/kernel/tracing/trace_pipe | grep my_event

# 4. 이벤트 필터 설정
echo 'value > 100' > \
    /sys/kernel/tracing/events/my_subsystem/my_event/filter

# 5. 히스토그램으로 집계
echo 'hist:keys=value:sort=hitcount' > \
    /sys/kernel/tracing/events/my_subsystem/my_event/trigger
cat /sys/kernel/tracing/events/my_subsystem/my_event/hist

Linux 6.12 ~ 6.17 Ring Buffer 동향

ftrace ring buffer는 Persistent(재부팅 생존) ring buffer 병합(6.12)을 기점으로 커널 관측성 영역에서 가장 활발하게 진화하는 자료구조 중 하나가 되었습니다. 이후 릴리스는 sub-buffer 크기 런타임 조정, mmap() 기반 zero-copy 읽기, eBPF 교차 참조로 확장되었습니다.

관련 문서

참고 자료

• kernel/trace/ring_buffer.c — ftrace 링 버퍼 핵심 구현체입니다. per-CPU 페이지 관리, lockless 쓰기, NMI-safe 예약 등 모든 핵심 로직이 포함되어 있습니다.
• include/linux/ring_buffer.h — 링 버퍼 공개 API 헤더입니다. ring_buffer_event, ring_buffer_alloc(), ring_buffer_write() 등의 인터페이스를 정의합니다.
• kernel/trace/trace.c — ftrace 프레임워크의 메인 구현체로, 링 버퍼를 소비하여 tracefs 인터페이스를 제공합니다.
• kernel/trace/ring_buffer_benchmark.c — 링 버퍼 벤치마크 커널 모듈 소스입니다. 쓰기/읽기 성능 측정에 사용합니다.
• Documentation/trace/ring-buffer-design.rst — 커널 공식 링 버퍼 설계 문서입니다. lockless 알고리즘, 중첩 쓰기, 타임스탬프 설계를 상세히 설명합니다.
• Documentation/trace/ftrace.rst — ftrace 공식 사용자 가이드입니다. tracefs 인터페이스를 통한 링 버퍼 제어 방법을 다룹니다.
• Documentation/trace/events.rst — trace event 시스템 문서입니다. 링 버퍼에 이벤트를 기록하는 메커니즘을 설명합니다.
• LWN: A look at ftrace (2009) — ftrace의 초기 설계와 링 버퍼 아키텍처를 소개하는 LWN 기사입니다.
• LWN: Debugging the kernel using ftrace - part 1 (2009) — ftrace를 활용한 커널 디버깅 기법을 다루며, 링 버퍼의 실전 활용 사례를 제공합니다.
• LWN: Using the TRACE_EVENT() macro (2010) — TRACE_EVENT 매크로를 통해 링 버퍼에 구조화된 이벤트를 기록하는 방법을 설명합니다.
• LWN: User-space ring buffer for tracing (2023) — 사용자 공간 링 버퍼 제안을 다루는 기사로, 커널 링 버퍼와의 비교 관점을 제공합니다.
• Steven Rostedt, "Internals of the RT Patch" (2009) — 링 버퍼 설계자인 Steven Rostedt의 논문으로, lockless 링 버퍼의 설계 동기와 구현 원리를 설명합니다.
• Steven Rostedt, "ftrace for Embedded" (Linux Plumbers, 2009) — 임베디드 환경에서의 ftrace 활용과 링 버퍼 최적화 전략을 다룬 발표 자료입니다.
• Steven Rostedt, "Understanding the Linux Kernel via Ftrace" (Kernel Recipes, 2017) — ftrace를 활용한 커널 내부 분석 기법을 다루는 발표로, 링 버퍼의 동작 원리를 시각적으로 설명합니다.
• tools/tracing/ — 커널 소스 트리에 포함된 트레이싱 보조 도구 모음입니다.