objpool (Lock-Free Per-CPU Object Pool)

리눅스 커널의 lock-free per-CPU 오브젝트 풀(Object Pool)인 objpool을 심층 분석합니다. include/linux/objpool.h와 lib/objpool.c에 정의된 이 메커니즘은 Linux 6.4에서 도입되었으며, NMI(Non-Maskable Interrupt)를 포함한 모든 인터럽트(Interrupt) 컨텍스트에서 안전하게 오브젝트를 할당/반환할 수 있도록 설계되었습니다. kprobe, fprobe, rethook의 핵심 인프라로 사용되며, per-CPU LIFO 스택 기반의 lock-free 알고리즘으로 높은 성능을 달성합니다.

단계별 이해

objpool의 동작 원리를 일상적인 비유로 단계별로 이해해 봅니다.

볼링장 신발 대여 시스템

볼링장에 4개의 레인(CPU 0~3)이 있고, 각 레인에 전용 신발 선반이 있습니다.

핵심 포인트는 다음과 같습니다.

• per-CPU 격리: 각 CPU가 자기만의 슬롯을 사용하므로 다른 CPU와 경합이 없습니다. 캐시 라인(Cache Line) 바운싱도 발생하지 않습니다.
• 사전 할당: 모든 오브젝트는 초기화 시점에 미리 만들어 놓으므로 런타임에 kmalloc 호출이 없습니다. 이로 인해 NMI/인터럽트 컨텍스트에서도 안전합니다.
• 고갈 시 NULL 반환: 오브젝트가 고갈되면 NULL을 반환할 뿐, 블로킹(Blocking)하지 않습니다. 호출자가 이 경우를 처리해야 합니다.
• Cross-CPU 반환 허용: CPU A에서 꺼낸 오브젝트를 CPU B에서 돌려놓을 수 있습니다. 이 경우 CPU B의 선반에 추가됩니다.

개요

왜 objpool이 필요한가

커널에서 오브젝트를 할당하는 일반적인 방법은 kmalloc/kmem_cache_alloc입니다. 그러나 이러한 할당자는 다음과 같은 상황에서 사용할 수 없습니다.

• NMI 컨텍스트 -- NMI 핸들러는 어떤 락도 획득할 수 없으므로, 내부적으로 락을 사용하는 SLUB/SLAB 할당자를 호출하면 데드락(Deadlock)이 발생합니다.
• 인터럽트 컨텍스트에서의 지연 -- 하드 IRQ(Hard IRQ) 핸들러에서 GFP_ATOMIC으로 할당하더라도 실패 가능성이 있고, 지연이 예측 불가능합니다.
• kretprobe의 재진입 -- kretprobe 핸들러가 다른 kretprobe를 트리거하거나 NMI에 의해 중단될 수 있어, 풀링 메커니즘이 재진입 안전(Reentrant-safe)해야 합니다.

기존 방식의 한계

Linux 6.4 이전에는 kretprobe가 freelist(llist 기반)를 사용했습니다. 이 방식에는 다음과 같은 문제가 있었습니다.

objpool은 이러한 문제를 per-CPU 슬롯 분리를 통해 근본적으로 해결합니다. 각 CPU는 자신의 슬롯에서만 pop/push하므로 다른 CPU와의 동기화가 필요 없습니다.

구조체 분석

struct objpool_head

objpool의 최상위 관리 구조체입니다. include/linux/objpool.h에 정의되어 있습니다.

/* include/linux/objpool.h */
struct objpool_head {
    int                         obj_size;    /* 오브젝트 크기 (바이트) */
    int                         nr_objs;     /* 전체 오브젝트 수 */
    int                         nr_cpus;     /* num_possible_cpus() */
    int                         capacity;    /* 슬롯당 엔트리 수 (nr_objs/nr_cpus + nr_objs) */
    gfp_t                       gfp;         /* 할당 플래그 */
    refcount_t                  ref;         /* 참조 카운트 */
    unsigned long               flags;       /* OBJPOOL_NR_OBJECT_BIT 등 */
    struct objpool_slot __percpu *slots;      /* per-CPU 슬롯 배열 */
    objpool_fini_cb             release;     /* 해제 콜백 */
    void                        *context;    /* 콜백에 전달할 컨텍스트 */
};

struct objpool_slot

각 CPU에 할당되는 per-CPU 슬롯 구조체입니다. LIFO 스택으로 동작합니다.

/* lib/objpool.c */
struct objpool_slot {
    uint32_t        head;       /* 다음 pop 위치 (소비자 인덱스) */
    uint32_t        tail;       /* 다음 push 위치 (생산자 인덱스) */
    uint32_t        last;       /* entries[] 배열 크기 마스크 */
    uint32_t        mask;       /* capacity - 1 (2의 거듭제곱 마스크) */
    void            *entries[]; /* 오브젝트 포인터 배열 (flexible array) */
};

슬롯 용량 계산

각 per-CPU 슬롯의 capacity는 전체 오브젝트 수보다 크게 설정됩니다. 이는 오브젝트가 하나의 CPU에서 pop되고 다른 CPU에서 push될 수 있기 때문입니다.

/* lib/objpool.c - objpool_init_percpu_slots() */
/* 슬롯 용량 = nr_objs / nr_cpus + nr_objs (넉넉하게) */
int capacity = roundup_pow_of_two(pool->nr_objs / pool->nr_cpus + pool->nr_objs);
/* 최소 capacity = roundup_pow_of_two(nr_objs + nr_objs / nr_cpus) */
/* 예: nr_objs=25, nr_cpus=4 → capacity = roundup_pow_of_two(25/4 + 25) = 32 */

Lock-Free 알고리즘

per-CPU LIFO 스택

각 per-CPU 슬롯은 순환 배열(Circular Array)을 사용하는 LIFO(Last-In, First-Out) 스택입니다. head와 tail 인덱스가 오브젝트의 유효 범위를 나타냅니다.

• pop (할당): head를 1 감소시키고, entries[head & mask]에서 오브젝트를 꺼냅니다.
• push (반환): entries[head & mask]에 오브젝트를 넣고, head를 1 증가시킵니다.
• 사용 가능 수: head - tail이 0보다 크면 오브젝트가 있습니다.

NMI-Safe 설계 원리

objpool이 NMI에서도 안전한 이유는 다음과 같습니다.

1. per-CPU 격리: 각 CPU는 자기 슬롯에서만 pop/push하므로 다른 CPU와의 경합이 없습니다.
2. 단조 인덱스(Monotonic Index): head는 pop 시 감소, push 시 증가합니다. NMI가 중간에 끼어들어도 서로 다른 entries[] 위치를 접근합니다.
3. 비파괴적 읽기: pop 시 head를 먼저 감소시킨 후 entries[head & mask]를 읽습니다. NMI가 추가로 pop하면 head가 더 감소할 뿐, 이미 감소된 위치의 데이터는 보존됩니다.
4. 락 미사용: 어떤 종류의 락이나 cmpxchg 루프도 사용하지 않으므로, NMI에 의한 교착(Deadlock)이 원리적으로 불가능합니다.

Cross-CPU 반환

오브젝트가 CPU 0에서 pop되고 CPU 1에서 push될 수 있습니다. 이 경우 CPU 1의 head가 증가하고, CPU 0의 오브젝트 수는 변하지 않습니다. 시간이 지나면 특정 CPU에 오브젝트가 편중될 수 있지만, capacity가 충분히 크게 설정되어 있으므로 오버플로는 발생하지 않습니다.

핵심 API

objpool_init

오브젝트 풀을 초기화하고 모든 오브젝트를 사전 할당합니다.

/* include/linux/objpool.h */
int objpool_init(struct objpool_head *pool,
                 int nr_objs, int obj_size,
                 gfp_t gfp, void *context,
                 objpool_init_obj_cb objinit,
                 objpool_fini_cb release);

objpool_init의 내부 동작을 단계별로 살펴봅니다.

/* lib/objpool.c - objpool_init() 내부 */

/* 1. 기본 필드 초기화 */
pool->obj_size = obj_size;
pool->nr_objs  = nr_objs;
pool->nr_cpus  = num_possible_cpus();
pool->gfp      = gfp & ~__GFP_ZERO;
pool->context  = context;
pool->release  = release;
refcount_set(&pool->ref, nr_objs + 1);

/* 2. per-CPU 슬롯 할당 */
rc = objpool_init_percpu_slots(pool, nr_objs, context, objinit);

/* 3. 오브젝트를 각 CPU 슬롯에 라운드 로빈으로 분배 */
for (i = 0; i < nr_objs; i++) {
    int cpu = i % pool->nr_cpus;
    struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);
    void *obj = objpool_alloc_object(pool);
    if (objinit)
        objinit(obj, context);
    slot->entries[slot->head & slot->mask] = obj;
    slot->head++;
}

objpool_pop (objpool_alloc)

풀에서 오브젝트를 하나 꺼냅니다. NMI를 포함한 모든 컨텍스트에서 호출할 수 있습니다.

/* include/linux/objpool.h */
static inline void *objpool_pop(struct objpool_head *pool)
{
    void *obj = NULL;
    unsigned long flags;
    int i, cpu;

    /* 선점(Preemption) 비활성화하여 CPU 고정 */
    cpu = raw_smp_processor_id();

    for (i = 0; i < pool->nr_cpus; i++) {
        int c = (cpu + i) % pool->nr_cpus;
        struct objpool_slot *slot = per_cpu_ptr(pool->slots, c);
        obj = objpool_try_get_slot(slot);
        if (obj)
            return obj;
    }
    return NULL;
}

objpool_push (objpool_free)

오브젝트를 풀에 반환합니다. 어떤 CPU에서든 호출할 수 있으며, 반환 시점의 현재 CPU 슬롯에 push됩니다.

/* include/linux/objpool.h */
static inline int objpool_push(void *obj, struct objpool_head *pool)
{
    int cpu = raw_smp_processor_id();
    struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);

    /* LIFO push: entries[head] = obj; head++ */
    slot->entries[slot->head & slot->mask] = obj;
    smp_wmb();  /* entries 쓰기가 head 증가 전에 가시적이도록 */
    slot->head++;

    /* 참조 카운트 감소: 모든 오브젝트 반환 + fini 시 release 콜백 호출 */
    if (refcount_dec_and_test(&pool->ref))
        pool->release(pool, pool->context);

    return 0;
}

objpool_fini / objpool_drop

풀을 해제합니다. 두 가지 변형이 있습니다.

/* 정상 종료: 참조 카운트 감소 → 0이면 release 콜백 */
void objpool_fini(struct objpool_head *pool);

/* 비상 해제: 즉시 모든 리소스 해제 (초기화 실패 시) */
void objpool_drop(struct objpool_head *pool);

API 요약 표

kprobe/fprobe 활용

kretprobe_instance 풀링

kretprobe는 함수의 리턴(Return) 시점에 핸들러를 실행하는 메커니즘입니다. 함수 진입 시 kretprobe_instance를 할당하고, 리턴 시 해제합니다. 함수가 재귀적으로 호출되거나 NMI에 의해 중단될 수 있으므로, 풀링 메커니즘이 NMI-safe해야 합니다.

/* kernel/kprobes.c - kretprobe 등록 시 objpool 초기화 */
static int kretprobe_init_inst_pool(struct kretprobe *rp)
{
    int num = rp->maxactive;
    if (num <= 0)
        num = max_t(int, 10, 2 * num_possible_cpus());

    return objpool_init(&rp->rph->pool,
                        num,
                        rp->data_size + sizeof(struct kretprobe_instance),
                        GFP_KERNEL, rp,
                        kretprobe_init_inst, kretprobe_fini_pool);
}

kretprobe 핸들러가 실행되는 흐름은 다음과 같습니다.

fprobe 데이터 전달

fprobe는 ftrace 기반의 함수 프로브(Function Probe)입니다. 함수 진입/종료 핸들러 간에 데이터를 전달할 때 objpool을 사용합니다.

/* kernel/trace/fprobe.c - fprobe의 objpool 초기화 */
static int fprobe_init_rethook(struct fprobe *fp, int num)
{
    struct fprobe_rethook_node *fnh;
    int size;

    if (num <= 0)
        num = max_t(int, 10, 2 * num_possible_cpus());

    size = sizeof(*fnh) + fp->entry_data_size;

    /* rethook이 내부적으로 objpool을 사용 */
    fp->rethook = rethook_alloc(fp->ops.func, fprobe_rethook_handler,
                                 size, num);
    return fp->rethook ? 0 : -ENOMEM;
}

rethook 활용

rethook은 함수 리턴 후킹(Return Hooking)의 공통 프레임워크입니다. kretprobe와 fprobe 모두 rethook을 통해 objpool을 사용합니다.

rethook_node 풀링

/* kernel/trace/rethook.c */
struct rethook *rethook_alloc(void *data,
                              rethook_handler_t handler,
                              int size, int num)
{
    struct rethook *rh;

    rh = kzalloc(sizeof(struct rethook), GFP_KERNEL);
    if (!rh)
        return NULL;

    rh->data    = data;
    rh->handler = handler;
    refcount_set(&rh->ref, 1);

    /* objpool 초기화: rethook_node를 num개 사전 할당 */
    if (objpool_init(&rh->pool, num, size,
                     GFP_KERNEL, rh,
                     rethook_init_node, rethook_free_pool)) {
        kfree(rh);
        return NULL;
    }

    return rh;
}

rethook_node 할당/반환

/* kernel/trace/rethook.c - 리턴 후킹 시점 */
struct rethook_node *rethook_try_get(struct rethook *rh)
{
    struct rethook_node *node;

    if (unlikely(!refcount_inc_not_zero(&rh->ref)))
        return NULL;

    /* objpool에서 노드 할당 */
    node = (struct rethook_node *)objpool_pop(&rh->pool);
    if (node)
        node->rethook = rh;
    else
        refcount_dec(&rh->ref);

    return node;
}

void rethook_recycle(struct rethook_node *node)
{
    struct rethook *rh = node->rethook;

    /* objpool에 노드 반환 */
    objpool_push(node, &rh->pool);

    if (refcount_dec_and_test(&rh->ref))
        rethook_free(rh);
}

성능 특성

Lock-Free 장점

per-CPU 캐시 친화성

메모리 오버헤드

objpool의 메모리 사용량을 계산해 봅니다.

/* 메모리 계산 예시: nr_objs=25, obj_size=128, nr_cpus=4 */

/* per-CPU 슬롯 하나의 크기 */
int capacity = roundup_pow_of_two(25/4 + 25);  /* = 32 */
size_t slot_size = sizeof(struct objpool_slot) + capacity * sizeof(void *);
/* = 16 + 32 * 8 = 272 바이트 */

/* 전체 슬롯 메모리 = nr_cpus * slot_size */
/* = 4 * 272 = 1,088 바이트 */

/* 오브젝트 메모리 = nr_objs * obj_size */
/* = 25 * 128 = 3,200 바이트 */

/* 총 메모리 = 슬롯 + 오브젝트 + objpool_head */
/* ≈ 1,088 + 3,200 + 64 = 4,352 바이트 */

슬롯 배열의 용량이 roundup_pow_of_two로 올림되므로, 실제 필요한 것보다 슬롯 메모리가 더 사용될 수 있습니다. 그러나 이 오버헤드는 lock-free 성능 이점에 비하면 미미합니다.

확장성 분석

objpool의 확장성을 CPU 수에 따라 분석합니다. 각 시나리오에서의 예상 지연 시간을 표로 정리합니다.

최선의 경우(현재 CPU 슬롯에 오브젝트 존재) 지연은 CPU 수에 무관합니다. 최악의 경우(모든 CPU 슬롯을 순회)에도 락 경합이 없으므로 다른 CPU의 성능에 영향을 주지 않습니다. 이는 spinlock 기반 mempool과의 가장 큰 차이점입니다.

Cross-CPU 반환의 영향

오브젝트가 CPU A에서 pop되고 CPU B에서 push되면, 시간이 지남에 따라 오브젝트가 특정 CPU에 편중될 수 있습니다. 이 현상의 영향을 분석합니다.

• 양성 시나리오: kretprobe에서 함수 진입/종료가 같은 CPU에서 일어나는 경우가 대부분이므로, cross-CPU 반환은 드뭅니다.
• 악성 시나리오: IRQ 핸들러에서 pop하고 softirq(다른 CPU 가능)에서 push하는 경우, 특정 CPU에 오브젝트가 집중될 수 있습니다.
• 완화 메커니즘: pop 시 로컬 슬롯이 비어 있으면 다른 CPU 슬롯에서 가져오므로, 자연스러운 재분배가 일어납니다.
• 용량 보장: capacity가 nr_objs + nr_objs/nr_cpus로 설정되어 있어, 모든 오브젝트가 한 CPU에 몰려도 오버플로가 발생하지 않습니다.

벤치마크 비교

Wu Qiang의 원본 패치에서 보고된 벤치마크 결과입니다 (8 CPU 시스템, kretprobe). 동일한 함수에 kretprobe를 설치하고 초당 처리량을 측정한 것입니다.

per-CPU 분리에 의한 캐시 라인 경합 제거가 약 80%의 성능 향상을 가져왔습니다. CPU 수가 많을수록 이 격차는 더 벌어집니다. 이는 전역 llist_head에 대한 CAS 경합이 CPU 수에 비례하여 증가하기 때문이며, objpool은 CPU 수와 무관하게 일정한 성능을 유지합니다.

특히 NMI 컨텍스트에서의 안정성 향상이 가장 중요한 기여입니다. 기존 freelist 방식에서는 NMI가 CAS 루프 중간에 발생하면 ABA 문제로 인한 데이터 손상이 이론적으로 가능했지만, objpool은 이를 원천적으로 차단합니다.

디버깅

objpool 고갈 진단

모든 오브젝트가 사용 중이면 objpool_pop이 NULL을 반환합니다. kretprobe에서 이 상황이 발생하면 해당 함수 호출의 리턴 후킹이 누락됩니다.

/* kretprobe에서 풀 고갈 시 nmissed 카운터 증가 */
static int pre_handler_kretprobe(struct kprobe *p,
                                   struct pt_regs *regs)
{
    struct kretprobe_instance *ri;

    ri = objpool_pop(&rp->rph->pool);
    if (!ri) {
        /* 풀 고갈: 이 리턴은 추적되지 않음 */
        rp->nmissed++;
        return 0;
    }
    /* ... 정상 처리 ... */
}

per-CPU 분포 확인

오브젝트가 특정 CPU에 편중되었는지 확인하려면 각 슬롯의 head - tail 값을 조회합니다.

/* 디버깅용: per-CPU 슬롯 상태 출력 */
static void objpool_debug_dump(struct objpool_head *pool)
{
    int cpu;
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);
        pr_info("objpool CPU%d: head=%u tail=%u avail=%u capacity=%u\n",
                cpu, slot->head, slot->tail,
                slot->head - slot->tail,
                slot->mask + 1);
    }
}

nmissed 모니터링

kretprobe의 풀 고갈 빈도를 확인합니다.

# kretprobe nmissed 확인 (debugfs)
cat /sys/kernel/debug/kprobes/list
# 출력 예시:
# ffffffff81234560 k do_sys_openat2+0x0 [OPTIMIZED]
# ffffffff81234560 r do_sys_openat2+0x0 nmissed=3

# nmissed가 높으면 maxactive 증가 필요
# echo 'r:myprobe do_sys_openat2 maxactive=100' > kprobe_events

maxactive 적정값 결정

kretprobe의 maxactive 값은 동시에 활성 상태인 인스턴스의 최대 수를 결정합니다. 이 값이 너무 작으면 nmissed가 증가하고, 너무 크면 메모리가 낭비됩니다.

maxactive 결정 가이드

기본값: max(10, 2 * num_possible_cpus()) — 4 CPU 시스템에서는 10, 128 CPU 시스템에서는 256입니다.

고주파 함수 (do_sys_openat2 등)

기본값의 4~8배: 40~80 (4 CPU 기준)

재귀적 함수 (lookup_slow 등)

재귀 깊이 × num_cpus: 재귀 깊이 8이면 32 (4 CPU 기준)

NMI 컨텍스트에서 호출되는 함수

기본값 + num_cpus: 14 (4 CPU 기준). NMI는 CPU당 1개 추가 인스턴스만 필요합니다.

일반적인 문제와 해결

ftrace를 이용한 추적

# objpool_pop/push 호출 추적
echo 1 > /sys/kernel/debug/tracing/events/kprobes/enable
echo 'p:objpool_pop_trace objpool_pop' >> /sys/kernel/debug/tracing/kprobe_events
echo 'p:objpool_push_trace objpool_push' >> /sys/kernel/debug/tracing/kprobe_events
echo 1 > /sys/kernel/debug/tracing/tracing_on
cat /sys/kernel/debug/tracing/trace_pipe

BPF를 이용한 고급 추적

# bpftrace로 objpool pop/push 빈도 모니터링
bpftrace -e '
kprobe:objpool_try_get_slot {
    @pop_count[cpu] = count();
}
kprobe:objpool_try_add_slot {
    @push_count[cpu] = count();
}
interval:s:5 {
    print(@pop_count);
    print(@push_count);
    clear(@pop_count);
    clear(@push_count);
}
'

# bpftrace로 objpool 고갈 이벤트 추적
bpftrace -e '
kretprobe:objpool_pop /retval == 0/ {
    @null_returns[cpu] = count();
    @null_stack[kstack] = count();
}
interval:s:10 {
    print(@null_returns);
    print(@null_stack);
}
'

crash 도구를 이용한 사후 분석

시스템 크래시 덤프에서 objpool 상태를 분석하는 방법입니다.

# crash 도구에서 objpool_head 구조체 확인
crash> struct objpool_head 0xffff888100123400
struct objpool_head {
  obj_size = 192,
  nr_objs = 25,
  nr_cpus = 4,
  capacity = 32,
  gfp = 3264,
  ref = {
    refs = {
      counter = 18
    }
  },
  slots = 0x00000000deadbeef,
  release = 0xffffffff81234560,
  context = 0xffff888100456000
}

# per-CPU 슬롯 상태 확인
crash> p/d ((struct objpool_slot *)per_cpu(slots, 0))->head
$1 = 15
crash> p/d ((struct objpool_slot *)per_cpu(slots, 0))->tail
$2 = 8
# → CPU 0에 7개 오브젝트 보유 (head - tail = 7)

소스 코드 워크스루

objpool_try_get_slot 내부 추적

objpool_pop이 호출하는 objpool_try_get_slot의 실제 구현을 분석합니다.

/* lib/objpool.c */
static inline void *objpool_try_get_slot(struct objpool_slot *slot)
{
    void *obj;
    uint32_t head, tail;

    /* head와 tail을 읽기 */
    head = READ_ONCE(slot->head);
    tail = READ_ONCE(slot->tail);

    /* 비어있으면 NULL 반환 */
    if (head == tail)
        return NULL;

    /* head를 먼저 감소 (LIFO: 가장 최근에 push된 것을 pop) */
    head = head - 1;
    WRITE_ONCE(slot->head, head);

    /* 오브젝트를 읽기 전에 배리어 */
    smp_rmb();

    /* entries에서 오브젝트 포인터 획득 */
    obj = slot->entries[head & slot->mask];

    /* 해당 위치를 NULL로 클리어 (디버깅 용도) */
    slot->entries[head & slot->mask] = NULL;

    return obj;
}

objpool_try_add_slot 내부 추적

/* lib/objpool.c */
static inline int objpool_try_add_slot(void *obj,
                                        struct objpool_slot *slot)
{
    uint32_t head, tail;

    head = READ_ONCE(slot->head);
    tail = READ_ONCE(slot->tail);

    /* 슬롯이 꽉 찼는지 확인 */
    if (head - tail >= slot->last)
        return -ENOSPC;

    /* 오브젝트를 현재 head 위치에 저장 */
    slot->entries[head & slot->mask] = obj;

    /* 쓰기 배리어: entries 쓰기 후 head 증가 */
    smp_wmb();

    /* head 증가 */
    WRITE_ONCE(slot->head, head + 1);

    return 0;
}

per-CPU 슬롯 초기화 워크스루

/* lib/objpool.c - objpool_init_percpu_slots() */
static int objpool_init_percpu_slots(struct objpool_head *pool,
                                      int nr_objs,
                                      void *context,
                                      objpool_init_obj_cb objinit)
{
    int i, cpu;

    /* capacity 계산: 2의 거듭제곱으로 올림 */
    pool->capacity = roundup_pow_of_two(nr_objs / pool->nr_cpus + nr_objs);

    /* per-CPU 슬롯 메모리 할당 */
    pool->slots = __alloc_percpu(
        sizeof(struct objpool_slot) + pool->capacity * sizeof(void *),
        sizeof(void *));
    if (!pool->slots)
        return -ENOMEM;

    /* 각 CPU 슬롯 초기화 */
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);
        slot->head = 0;
        slot->tail = 0;
        slot->last = pool->capacity;
        slot->mask = pool->capacity - 1;
    }

    /* 오브젝트 할당 및 라운드 로빈 분배 */
    for (i = 0; i < nr_objs; i++) {
        void *obj;
        cpu = i % pool->nr_cpus;

        obj = kvmalloc_node(pool->obj_size, pool->gfp,
                             cpu_to_node(cpu));
        if (!obj)
            return -ENOMEM;

        if (objinit)
            objinit(obj, context);

        /* 해당 CPU 슬롯에 push */
        objpool_add_scattered(obj, pool, cpu);
    }
    return 0;
}

objpool_fini 내부 추적

/* lib/objpool.c */
void objpool_fini(struct objpool_head *pool)
{
    int count, cpu;

    /* 풀에 남아 있는 오브젝트 수 세기 */
    count = 0;
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);
        count += slot->head - slot->tail;
    }

    /* 풀에 있는 오브젝트 수만큼 refcount 감소 */
    if (count > 0)
        refcount_sub(count, &pool->ref);

    /* 자체 참조(-1) 해제 */
    if (refcount_dec_and_test(&pool->ref))
        objpool_free(pool);
}

참조 카운팅 흐름 정리

objpool의 참조 카운팅 메커니즘은 안전한 지연 해제(Deferred Free)를 보장합니다.

이 설계의 핵심은 objpool_fini 호출 후에도 아직 사용 중인 오브젝트를 안전하게 반환할 수 있는 점입니다. 마지막 오브젝트가 push될 때 비로소 release 콜백이 실행되어 모든 리소스가 해제됩니다.

실전 사용 예제

커널 모듈에서 objpool을 직접 사용하는 예제입니다.

#include <linux/module.h>
#include <linux/objpool.h>

struct my_context {
    struct objpool_head pool;
    /* 추가 필드 ... */
};

struct my_object {
    unsigned long timestamp;
    int           cpu;
    char          data[64];
};

/* 오브젝트 초기화 콜백 */
static int my_obj_init(void *obj, void *context)
{
    struct my_object *my = obj;
    memset(my, 0, sizeof(*my));
    return 0;
}

/* 풀 해제 콜백 */
static void my_pool_release(struct objpool_head *pool, void *context)
{
    struct my_context *ctx = context;
    pr_info("objpool released, all objects returned\n");
    kfree(ctx);
}

/* 모듈 초기화 */
static int __init my_init(void)
{
    struct my_context *ctx;
    int ret;

    ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
    if (!ctx)
        return -ENOMEM;

    /* 20개의 my_object를 사전 할당 */
    ret = objpool_init(&ctx->pool,
                        20,                    /* nr_objs */
                        sizeof(struct my_object), /* obj_size */
                        GFP_KERNEL,             /* gfp */
                        ctx,                    /* context */
                        my_obj_init,            /* objinit */
                        my_pool_release);       /* release */
    if (ret) {
        kfree(ctx);
        return ret;
    }

    /* 오브젝트 사용 예시 */
    {
        struct my_object *obj = objpool_pop(&ctx->pool);
        if (obj) {
            obj->timestamp = ktime_get_ns();
            obj->cpu = raw_smp_processor_id();
            snprintf(obj->data, sizeof(obj->data), "hello");

            /* 사용 후 반환 */
            objpool_push(obj, &ctx->pool);
        }
    }

    return 0;
}

/* 모듈 종료 */
static void __exit my_exit(void)
{
    /* objpool_fini 호출 → 모든 오브젝트 반환 후 release 콜백 */
    /* ctx는 release 콜백에서 kfree됨 */
    objpool_fini(&global_ctx->pool);
}

module_init(my_init);
module_exit(my_exit);
MODULE_LICENSE("GPL");

에러 처리 패턴

objpool 초기화가 중간에 실패할 수 있습니다. 이 경우 objpool_drop을 사용하여 이미 할당된 리소스를 정리합니다.

/* 초기화 실패 처리 */
int ret = objpool_init(&pool, nr_objs, obj_size,
                      GFP_KERNEL, ctx, init_cb, release_cb);
if (ret) {
    /* objpool_init이 실패하면 내부적으로 objpool_drop을 호출하여 */
    /* 이미 할당된 슬롯과 오브젝트를 해제합니다. */
    /* 호출자는 pool 구조체를 포함한 상위 구조체만 해제하면 됩니다. */
    kfree(ctx);
    return ret;
}

/* 주의: objpool_init 실패 시 release 콜백은 호출되지 않습니다. */
/* release 콜백에서 ctx를 kfree하더라도, 실패 경로에서는 직접 해제해야 합니다. */

관련 메커니즘 비교

사용 시나리오 정리

• objpool을 사용할 때: NMI/인터럽트 컨텍스트에서 안전한 할당이 필요하고, 최대 오브젝트 수를 미리 알 수 있으며, 고갈 시 NULL 반환이 허용될 때 사용합니다.
• mempool을 사용할 때: 메모리 부족 상황에서도 반드시 할당이 성공해야 할 때 사용합니다 (대기 가능한 컨텍스트).
• kmem_cache를 사용할 때: 오브젝트 수가 동적으로 변하고, 프로세스 컨텍스트에서 사용할 때 적합합니다.

상세 동작 비교

mempool과의 차이

mempool은 include/linux/mempool.h에 정의되어 있으며, 메모리 압박(Memory Pressure) 상황에서도 할당을 보장하는 비상 풀입니다.

/* mempool의 할당 흐름 (비교용) */
void *mempool_alloc(struct mempool *pool, gfp_t gfp)
{
    /* 1. 일반 할당 시도 */
    element = pool->alloc(gfp, pool->pool_data);
    if (element)
        return element;

    /* 2. 실패하면 spinlock 획득 후 비상 풀에서 할당 */
    spin_lock_irqsave(&pool->lock, flags);  /* ← NMI에서 교착! */
    if (pool->curr_nr) {
        element = pool->elements[--pool->curr_nr];
        spin_unlock_irqrestore(&pool->lock, flags);
        return element;
    }
    spin_unlock_irqrestore(&pool->lock, flags);

    /* 3. 비상 풀도 비면 대기 (블로킹) */
    prepare_to_wait(&pool->wait, ...);
    schedule();  /* ← 인터럽트 컨텍스트에서 불가! */
}

kmem_cache (SLUB)와의 차이

/* SLUB의 할당 흐름 (비교용) */
void *kmem_cache_alloc(struct kmem_cache *s, gfp_t flags)
{
    /* Fast path: per-CPU freelist에서 즉시 할당 (lock-free) */
    obj = __slab_alloc_node(...);  /* cmpxchg 사용 */
    if (obj)
        return obj;

    /* Slow path: 새 slab 페이지 할당 (GFP 플래그 필요) */
    page = alloc_pages(flags, ...);  /* ← NMI에서 불가! */
    /* ... slab 초기화 ... */
}

SLUB의 fast path는 per-CPU freelist를 사용하여 lock-free이지만, freelist가 비면 새 slab 페이지를 할당해야 하고, 이 과정에서 buddy allocator의 zone lock이 필요합니다. 따라서 NMI 컨텍스트에서는 사용할 수 없습니다.

llist (lockless list)와의 차이

/* llist의 pop 동작 (비교용) */
struct llist_node *llist_del_first(struct llist_head *head)
{
    struct llist_node *entry, *old_entry, *next;

    entry = smp_load_acquire(&head->first);
    for (;;) {
        if (entry == NULL)
            return NULL;
        old_entry = entry;
        next = READ_ONCE(entry->next);
        /* CAS 루프: 다른 CPU가 head를 변경하면 재시도 */
        entry = cmpxchg(&head->first, old_entry, next);
        if (entry == old_entry)   /* 성공 */
            return entry;
        /* 실패: 다시 읽고 재시도 (ABA 문제 가능) */
    }
}

설계 트레이드오프(Trade-off)

향후 발전 방향

• 동적 크기 조정: 현재 objpool은 초기화 후 크기를 변경할 수 없습니다. 향후 런타임에 오브젝트를 추가/제거하는 기능이 논의되고 있습니다.
• 범용화: kprobe/fprobe/rethook 외에도 BPF 프로그램이나 네트워크 스택에서의 활용이 검토되고 있습니다.
• NUMA 최적화: 현재 cpu_to_node로 로컬 메모리에서 할당하지만, cross-node 반환 시 추가 최적화 여지가 있습니다.
• 통계 인터페이스: per-CPU 슬롯 상태를 debugfs/sysfs로 노출하는 패치가 논의되고 있습니다.

커널 버전별 변화

자주 하는 실수

• pop 후 push를 잊음: 에러 경로(Error Path)에서 objpool_push를 호출하지 않으면 오브젝트가 영구적으로 손실되며, 풀이 점차 고갈됩니다. 모든 objpool_pop 호출에는 대응하는 objpool_push가 있어야 합니다.
• fini 후 push 호출 안전성 오해: objpool_fini 호출 후에도 사용 중인 오브젝트를 push할 수 있습니다. 참조 카운팅 메커니즘이 이를 안전하게 처리합니다.
• 객체 크기 불일치: objpool_init에 전달한 obj_size와 실제로 사용하는 오브젝트의 크기가 다르면 메모리 손상이 발생합니다.
• release 콜백에서 pool 접근: release 콜백이 호출될 때 pool은 이미 해제 중이므로, pool 멤버에 접근하면 안 됩니다. context 포인터만 사용해야 합니다.
• objpool_drop 오용: objpool_drop은 초기화 실패 경로 전용입니다. 정상 해제에는 반드시 objpool_fini를 사용해야 합니다.
• nr_objs 과소 설정: 동시 활성 오브젝트 수보다 nr_objs가 작으면 빈번한 NULL 반환이 발생합니다. kretprobe의 경우 maxactive를 2 * num_possible_cpus() 이상으로 설정해야 합니다.
• GFP 플래그 오류: objpool_init은 프로세스 컨텍스트에서만 호출할 수 있으므로 GFP_KERNEL을 사용합니다. GFP_ATOMIC은 필요하지 않으며, 런타임 pop/push는 할당 자체를 하지 않으므로 GFP 플래그와 무관합니다.

Per-CPU 슬롯 내부 구조 심층

objpool의 핵심 자료 구조는 struct objpool_slot입니다. 각 CPU마다 하나씩 할당되며, 내부에 entries[] 순환 배열과 head/tail 인덱스를 사용하여 LIFO 스택을 구현합니다. 이 섹션에서는 entries[] 배열의 물리적 배치, head/tail 인덱스의 의미, 그리고 push/pop 연산의 5단계 과정을 상세히 분석합니다.

entries[] 배열의 메모리 배치

objpool_slot은 가변 길이 구조체입니다. 헤더 뒤에 entries[] 배열이 이어지며, 배열의 크기는 capacity 필드가 결정합니다. entries[]의 각 원소는 오브젝트 포인터(void *)가 아니라, 오브젝트 풀 전체 배열 내에서의 인덱스입니다. 실제 오브젝트 주소는 objpool_head.obj_arr[index]로 변환합니다.

/* include/linux/objpool.h - 슬롯 구조체 */
struct objpool_slot {
    uint32_t head;      /* pop 위치 (소비자 인덱스) */
    uint32_t tail;      /* push 위치 (생산자 인덱스) */
    uint32_t last;      /* 마지막 push 위치 (tail 캐시) */
    uint32_t mask;      /* capacity - 1 (2의 거듭제곱 마스크) */
    void *entries[];    /* 오브젝트 포인터 순환 배열 */
};

/* capacity는 항상 2의 거듭제곱 → mask = capacity - 1
 * 인덱스 계산: slot->entries[idx & slot->mask]
 * 이렇게 하면 모듈로 연산 없이 비트 AND로 순환 가능 */

head와 tail은 단조 증가(monotonically increasing)하는 32비트 정수입니다. 실제 배열 인덱스는 head & mask 또는 tail & mask로 계산합니다. 이 설계 덕분에 head와 tail이 같은 값을 가질 때 빈 슬롯, tail - head가 capacity와 같을 때 가득 찬 슬롯임을 O(1)에 판별할 수 있습니다.

/* 슬롯 상태 판별 매크로 (개념적) */
#define SLOT_EMPTY(s)  ((s)->head == (s)->tail)
#define SLOT_FULL(s)   ((s)->tail - (s)->head == (s)->mask + 1)
#define SLOT_COUNT(s)  ((s)->tail - (s)->head)

/* head/tail 단조 증가 예시:
 * capacity=4, mask=3일 때
 * head=5 → 실제 인덱스 = 5 & 3 = 1
 * tail=8 → 실제 인덱스 = 8 & 3 = 0
 * 현재 오브젝트 수 = 8 - 5 = 3 */

LIFO push/pop 5단계

objpool의 push와 pop은 각각 5단계로 구성됩니다. 락 없이 동작하기 위해 head와 tail 인덱스를 원자적으로 조작하며, try_cmpxchg_release/acquire 시맨틱을 활용합니다.

pop 연산 (오브젝트 획득)

1. 현재 CPU 슬롯 조회: this_cpu_ptr(pool->cpu_slots)로 현재 CPU의 슬롯 포인터를 가져옵니다.
2. head 읽기: READ_ONCE(slot->head)로 현재 head 값을 가져옵니다.
3. 비어있는지 확인: head == READ_ONCE(slot->tail)이면 이 슬롯은 비어 있으므로 다른 CPU 슬롯에서 시도합니다(steal).
4. head 전진(CAS): try_cmpxchg_release(&slot->head, &head, head + 1)로 head를 원자적으로 1 증가시킵니다. 실패하면 2단계로 돌아갑니다.
5. 오브젝트 반환: slot->entries[head & slot->mask]에서 오브젝트 포인터를 꺼내 반환합니다.

push 연산 (오브젝트 반환)

1. 현재 CPU 슬롯 조회: this_cpu_ptr(pool->cpu_slots)로 현재 CPU의 슬롯을 가져옵니다.
2. tail 읽기: READ_ONCE(slot->tail)로 현재 tail 값을 가져옵니다.
3. 오브젝트 저장: slot->entries[tail & slot->mask] = obj로 오브젝트 포인터를 배열에 기록합니다.
4. 쓰기 배리어: smp_wmb()로 오브젝트 포인터가 entries[]에 완전히 기록된 후에 tail이 전진하도록 보장합니다.
5. tail 전진: WRITE_ONCE(slot->tail, tail + 1)로 tail을 1 증가시켜 새 오브젝트가 가시적이 되도록 합니다.

/* objpool_pop 실제 구현 (include/linux/objpool.h, 단순화) */
static inline void *objpool_pop(struct objpool_head *pool)
{
    struct objpool_slot *slot;
    uint32_t head, tail;
    int cpu, i;

    /* 1단계: 현재 CPU 슬롯 조회 */
    cpu = raw_smp_processor_id();
    for (i = 0; i < num_possible_cpus(); i++) {
        slot = per_cpu_ptr(pool->cpu_slots,
                           (cpu + i) % num_possible_cpus());
        /* 2단계: head 읽기 */
        head = READ_ONCE(slot->head);
        do {
            /* 3단계: 비어있는지 확인 */
            tail = READ_ONCE(slot->tail);
            if (head == tail)
                break;  /* 이 슬롯은 비어 있음 → 다음 CPU 시도 */
            /* 4단계: head 전진 (CAS) */
        } while (!try_cmpxchg_release(&slot->head,
                                       &head, head + 1));
        if (head != tail) {
            /* 5단계: 오브젝트 반환 */
            return slot->entries[head & slot->mask];
        }
    }
    return NULL;  /* 모든 슬롯이 비어 있음 */
}

capacity 계산과 2의 거듭제곱 정렬

각 per-CPU 슬롯의 capacity는 nr_objs를 CPU 수로 나눈 값을 2의 거듭제곱으로 올림한 값입니다. 이 정렬은 모듈로 연산을 비트 AND로 대체하여 성능을 최적화합니다.

/* lib/objpool.c - capacity 계산 (단순화) */
static int objpool_init_percpu_slots(struct objpool_head *pool,
                                     int nr_objs, int obj_size,
                                     gfp_t gfp)
{
    int nr_cpus = num_possible_cpus();
    /* CPU당 최소 오브젝트 수: nr_objs / nr_cpus, 최소 4 */
    int capacity = max(roundup_pow_of_two(nr_objs / nr_cpus + 1), 4UL);
    int slot_size;
    int cpu;

    /* 슬롯 크기: 헤더 + entries[] 배열 */
    slot_size = struct_size_t(struct objpool_slot, entries, capacity);

    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot;
        /* per-CPU 슬롯 할당 (NUMA-aware) */
        slot = kvzalloc_node(slot_size, gfp, cpu_to_node(cpu));
        if (!slot)
            return -ENOMEM;
        slot->mask = capacity - 1;
        /* per-CPU 포인터에 저장 */
        *per_cpu_ptr(pool->cpu_slots, cpu) = slot;
    }

    pool->capacity = capacity;
    return 0;
}

/* 예시: nr_objs=20, nr_cpus=4
 * → 20/4 + 1 = 6 → roundup_pow_of_two(6) = 8
 * → capacity=8, mask=7
 * → 총 슬롯 메모리: 4 * (16 + 8*8) = 4 * 80 = 320 바이트 */

steal 알고리즘 상세

로컬 슬롯이 비어있을 때 실행되는 steal 알고리즘은 라운드 로빈 방식으로 다른 CPU의 슬롯을 순회합니다. steal은 원격 캐시라인 접근을 유발하므로 성능 비용이 높지만, NULL을 반환하는 것보다 낫습니다.

• 순회 순서: (current_cpu + 1) % nr_cpus부터 시작하여 모든 CPU를 한 바퀴 돕니다.
• 공정성: 매번 현재 CPU 다음부터 시작하므로 특정 CPU에 편중되지 않습니다.
• 캐시 비용: 원격 슬롯의 head/tail과 entries[]가 다른 CPU의 L1/L2 캐시에 있으므로 MESI Invalidate가 발생합니다. 이 비용은 50~200ns 정도로, 로컬 pop의 15ns에 비해 상당히 높습니다.
• 빈도 최소화: nr_objs를 충분히 크게 설정하면 steal이 거의 발생하지 않습니다. kretprobe의 기본 maxactive = 2 * num_possible_cpus()는 CPU당 2개씩 확보하여 steal 빈도를 낮춥니다.

NMI-Safe 설계 원리

NMI(Non-Maskable Interrupt)는 커널에서 가장 제한적인 인터럽트 컨텍스트입니다. kmalloc, mutex, spinlock_t(irqsave 포함) 등 거의 모든 동기화/할당 메커니즘이 NMI 내에서 사용할 수 없습니다. objpool은 이 극한 환경에서도 안전하게 동작하도록 설계되었습니다.

NMI에서 일반 할당이 불가능한 이유

NMI는 다른 모든 인터럽트(하드웨어 인터럽트, 소프트 인터럽트 포함)를 선점할 수 있으며, 비활성화할 수 없습니다. 이로 인해 다음과 같은 문제가 발생합니다:

• 스핀락 데드락: CPU A가 슬랩 할당자의 스핀락을 잡고 있는 상태에서 NMI가 발생하면, NMI 핸들러가 같은 스핀락을 잡으려 할 때 자기 자신과 데드락이 발생합니다. spin_lock_irqsave도 NMI를 막지 못합니다.
• mempool 불가: mempool은 내부적으로 spin_lock_irqsave를 사용하므로 NMI에서 동일한 데드락 위험이 있습니다.
• llist(Lock-less List) 한계: Linux 6.4 이전의 kretprobe는 llist를 사용했지만, llist는 per-CPU가 아니므로 캐시라인 경합이 심했고, 전역 리스트의 cmpxchg 충돌로 성능이 저하되었습니다.

per-CPU + lock-free로 NMI-safe 달성

objpool이 NMI-safe를 달성하는 핵심 원리는 다음 세 가지입니다:

1. per-CPU 분리: 각 CPU가 독립적인 슬롯을 가지므로 CPU 간 경합이 없습니다. NMI가 현재 CPU의 슬롯에 접근해도 다른 CPU의 슬롯과 충돌하지 않습니다.
2. 락 미사용: try_cmpxchg와 WRITE_ONCE/READ_ONCE만 사용합니다. CAS 실패 시 데드락이 아니라 단순히 재시도합니다.
3. 사전 할당: 모든 오브젝트가 objpool_init 시점에 할당되므로 런타임에 메모리 할당자를 호출할 필요가 없습니다.

/* NMI에서 objpool 사용이 안전한 이유 - 시나리오 분석 */

/* 시나리오: CPU 0에서 kretprobe 핸들러 실행 중 NMI 발생 */
void kretprobe_handler(void) {
    /* ① 프로세스 컨텍스트에서 pop (head=3→4) */
    struct kretprobe_instance *ri = objpool_pop(&rp->rph->pool);
    /* ... ri 사용 중 ... */

    /* ★ 여기서 NMI 발생! */
    /* NMI 핸들러 내부의 다른 kretprobe도 같은 풀에서 pop 시도 */
    /* → head=4→5 (CAS 성공, 다른 슬롯 인덱스이므로 충돌 없음) */

    /* ② NMI 핸들러 복귀 후 계속 진행 */
    objpool_push(ri, &rp->rph->pool);  /* tail 전진 */
}

/* 핵심: head의 CAS는 NMI 핸들러와 일반 핸들러가 서로 다른 값으로
 * 경합하므로 한쪽은 성공하고 다른 쪽은 재시도합니다.
 * 데드락이 발생할 수 있는 지점이 존재하지 않습니다. */

NMI 중첩 시나리오

x86 아키텍처에서 NMI는 이론적으로 중첩될 수 있습니다(NMI 핸들러 실행 중 다시 NMI 발생). 이 극단적인 경우에도 objpool은 안전합니다:

NMI 중첩 시나리오

같은 CPU에서 head를 두 번 전진시키는 경우입니다. 초기 슬롯 상태: head=10, tail=14 (4개 오브젝트)

1. [NMI-1 진입] pop 시작: head=10을 읽고, CAS(&head, 10, 11) 성공 → entries[10 & mask]를 반환합니다. NMI-1 처리 중...
2. [NMI-2 진입] pop 시작 (NMI 중첩): head=11을 읽고 (NMI-1이 이미 전진시킴), CAS(&head, 11, 12) 성공 → entries[11 & mask]를 반환합니다. NMI-2 처리 후 push → tail 전진합니다.
3. [NMI-2 복귀]
4. NMI-1이 push → tail 전진합니다.
5. [NMI-1 복귀]

결과: 모든 오브젝트가 정상적으로 반환됩니다. CAS의 원자성이 동일 CPU 내 중첩도 처리합니다.

메모리 순서 보장 분석

objpool의 NMI-safe 동작은 올바른 메모리 순서(Memory Ordering)에 의존합니다. 각 연산에서 사용되는 배리어(Barrier)를 분석합니다:

이 배리어 조합은 x86에서는 대부분 no-op입니다. x86의 TSO(Total Store Ordering) 모델에서 smp_wmb()는 컴파일러 배리어로만 동작하고, try_cmpxchg는 LOCK CMPXCHG 명령어로 컴파일되어 암묵적 풀 배리어를 제공합니다. 따라서 x86에서 objpool의 배리어 오버헤드는 사실상 0입니다. ARM64에서는 smp_wmb()가 DMB ISHST로 변환되어 약간의 비용이 발생합니다.

kretprobe 활용 상세

kretprobe는 함수 반환 시점을 후킹하는 메커니즘입니다. 함수 진입 시 kretprobe_instance를 할당하고, 반환 시 핸들러를 호출한 뒤 인스턴스를 반환합니다. Linux 6.4부터 이 인스턴스 풀링에 objpool이 사용됩니다.

kretprobe_instance 생명주기

kretprobe_instance는 함수 호출 하나당 하나씩 소비되며, 함수 반환 시 재활용됩니다. 이 생명주기에서 objpool의 pop/push가 정확히 한 번씩 호출됩니다.

/* kernel/kprobes.c - kretprobe_instance 정의 */
struct kretprobe_instance {
    union {
        struct freelist_node freelist;  /* 구 버전 호환 */
    };
    struct rcu_head rcu;
    char data[];  /* 사용자 정의 데이터 (kretprobe.data_size) */
};

/* kretprobe 등록: objpool 초기화 */
int register_kretprobe(struct kretprobe *rp)
{
    int ret;
    /* maxactive가 0이면 기본값 = 2 * num_possible_cpus() */
    if (rp->maxactive <= 0)
        rp->maxactive = max_t(int, 10, 2 * num_possible_cpus());

    /* objpool 초기화: maxactive개의 kretprobe_instance 사전 할당 */
    ret = objpool_init(&rp->rph->pool,
                       rp->maxactive,       /* nr_objs */
                       sizeof(struct kretprobe_instance) + rp->data_size,
                       GFP_KERNEL,
                       rp->rph,             /* context */
                       kretprobe_init_inst,  /* objpool_init_obj_cb */
                       kretprobe_fini_pool); /* objpool_fini_cb */
    return ret;
}

pre_handler에서 pop, post에서 push

kretprobe의 동작은 pre_handler_kretprobe(함수 진입)와 __kretprobe_trampoline_handler(함수 반환)에서 이루어집니다:

/* kernel/kprobes.c - 함수 진입 시 (pre_handler) */
static int pre_handler_kretprobe(struct kprobe *p, struct pt_regs *regs)
{
    struct kretprobe *rp = container_of(p, struct kretprobe, kp);
    struct kretprobe_instance *ri;

    /* ★ objpool에서 인스턴스 pop (NMI-safe) */
    ri = objpool_pop(&rp->rph->pool);
    if (!ri) {
        /* 모든 인스턴스가 사용 중 → 이 호출은 추적하지 않음 */
        rp->nmissed++;
        return 0;
    }

    /* 반환 주소를 kretprobe_trampoline으로 교체 */
    arch_prepare_kretprobe(ri, regs);
    /* 현재 태스크의 ri 목록에 추가 */
    llist_add(&ri->llist, &current->kretprobe_instances);
    return 0;
}

/* kernel/kprobes.c - 함수 반환 시 */
static __used void *__kretprobe_trampoline_handler(struct pt_regs *regs)
{
    struct kretprobe_instance *ri;
    /* current->kretprobe_instances에서 ri 꺼내기 */
    ri = /* ... llist에서 pop ... */;

    /* 사용자 핸들러 호출 */
    if (rp->handler)
        rp->handler(ri, regs);

    /* ★ objpool에 인스턴스 push (반환) */
    objpool_push(ri, &rp->rph->pool);

    return (void *)ri->ret_addr;  /* 원래 반환 주소 복원 */
}

maxactive와 nmissed

maxactive는 objpool에 사전 할당할 kretprobe_instance 수를 결정합니다. 동시에 활성화될 수 있는 대상 함수 호출 수보다 커야 합니다. nmissed는 인스턴스 부족으로 추적하지 못한 호출 수입니다.

/* maxactive 설정 가이드라인 */

/* 기본값: 2 * num_possible_cpus()
 * - 각 CPU에서 프로세스 컨텍스트 1개 + NMI 1개 동시 사용 가능
 * - 대부분의 경우 충분하지만, 재귀 함수나 장시간 실행 함수는 부족할 수 있음 */

/* nmissed 모니터링 */
static void check_kretprobe_status(struct kretprobe *rp)
{
    /* /sys/kernel/debug/kprobes/list에서도 확인 가능 */
    if (rp->nmissed > 0) {
        pr_warn("kretprobe %pS: nmissed=%lu (maxactive=%d)\n",
                rp->kp.addr, rp->nmissed, rp->maxactive);
        /* nmissed가 0보다 크면 maxactive를 늘려야 합니다 */
    }
}

/* 재귀 함수의 경우 maxactive 계산:
 * maxactive = num_possible_cpus() * max_recursion_depth * 2
 * 예: 4코어, 최대 재귀 깊이 10 → maxactive = 4 * 10 * 2 = 80 */

kretprobe: freelist에서 objpool로의 전환

Linux 6.4 이전에는 kretprobe가 freelist(llist 기반)를 사용했습니다. objpool로 전환된 이유를 비교합니다:

핵심 개선점은 캐시라인 분리입니다. freelist의 전역 head 포인터는 모든 CPU가 CAS로 경합하므로 L1 캐시라인이 지속적으로 무효화(invalidate)됩니다. objpool은 각 CPU가 자신의 슬롯만 접근하므로 캐시라인이 안정적으로 유지됩니다.

fprobe/rethook 통합 상세

fprobe는 ftrace 기반의 고성능 함수 프로브입니다. 반환 값 추적이 필요할 때 내부적으로 rethook을 사용하며, rethook은 다시 objpool로 rethook_node를 풀링합니다. 이 3단 구조(fprobe → rethook → objpool)를 상세히 분석합니다.

fprobe → rethook → objpool 3단 구조

fprobe의 exit_handler가 설정되면 rethook이 활성화되고, rethook은 objpool을 통해 rethook_node를 관리합니다:

/* kernel/trace/fprobe.c - fprobe 등록 시 rethook 초기화 */
int register_fprobe(struct fprobe *fp, const char *filter, const char *notfilter)
{
    /* exit_handler가 있으면 rethook 필요 */
    if (fp->exit_handler) {
        int num = fp->nr_maxactive ? : 2 * num_possible_cpus();
        fp->rethook = rethook_alloc((void *)fp,
                                     fprobe_exit_handler,
                                     num,
                                     sizeof(struct fprobe_rethook_node));
        /* rethook_alloc 내부에서 objpool_init 호출 */
    }
    /* ... ftrace 등록 ... */
    return 0;
}

/* kernel/trace/rethook.c - rethook 초기화 */
struct rethook *rethook_alloc(void *data,
                              rethook_handler_t handler,
                              int num, int data_size)
{
    struct rethook *rh;
    rh = kzalloc(sizeof(*rh), GFP_KERNEL);
    rh->handler = handler;
    rh->data = data;

    /* ★ objpool 초기화: num개의 rethook_node 사전 할당 */
    if (objpool_init(&rh->pool, num,
                     sizeof(struct rethook_node) + data_size,
                     GFP_KERNEL, rh,
                     rethook_init_node, rethook_fini_pool))
        goto err;
    return rh;
}

rethook_try_get과 rethook_recycle

/* kernel/trace/rethook.c - rethook_node 획득/반환 */

/* 함수 진입 시: rethook_node 획득 */
struct rethook_node *rethook_try_get(struct rethook *rh)
{
    struct rethook_node *node;

    if (unlikely(!rethook_valid(rh)))
        return NULL;

    /* ★ objpool에서 pop */
    node = objpool_pop(&rh->pool);
    if (node)
        node->rethook = rh;
    return node;
}

/* 함수 반환 후: rethook_node 반환 */
void rethook_recycle(struct rethook_node *node)
{
    /* ★ objpool에 push */
    objpool_push(node, &node->rethook->pool);
}

/* fprobe entry에서 rethook 사용 */
static void fprobe_handler(unsigned long ip, unsigned long parent_ip,
                            struct ftrace_ops *ops, struct ftrace_regs *fregs)
{
    struct fprobe *fp = container_of(ops, struct fprobe, ops);
    struct rethook_node *rhn = NULL;

    if (fp->exit_handler) {
        rhn = rethook_try_get(fp->rethook);  /* objpool_pop */
        if (!rhn) {
            fp->nmissed++;
            return;
        }
    }
    /* entry_handler 호출 */
    if (fp->entry_handler)
        fp->entry_handler(fp, ip, parent_ip, fregs);

    if (rhn)
        rethook_hook(rhn, fregs, true);  /* 반환 주소 교체 */
}

fprobe와 kretprobe 선택 기준

fprobe와 kretprobe는 모두 objpool을 사용하지만, 사용 방식에 차이가 있습니다:

fprobe는 kretprobe의 상위 호환으로 설계되었으며, 새로운 트레이싱 코드에서는 fprobe 사용이 권장됩니다. 두 메커니즘 모두 내부적으로 objpool의 동일한 pop/push 경로를 사용하므로 오브젝트 풀링 성능은 동일합니다.

rethook_node 전체 생명주기

rethook_node의 생명주기를 시간 순서대로 정리합니다:

1. 초기화 (rethook_alloc): objpool_init으로 num개의 rethook_node를 사전 할당합니다. 각 노드의 rethook 필드는 NULL입니다.
2. 획득 (rethook_try_get): objpool_pop으로 노드를 꺼냅니다. node->rethook = rh로 소속 rethook을 기록합니다.
3. 설치 (rethook_hook): 현재 태스크의 반환 주소 스택에 노드를 삽입합니다. 원래 반환 주소를 node->ret_addr에 저장하고, 반환 주소를 rethook_trampoline으로 교체합니다.
4. 트리거 (rethook_trampoline_handler): 대상 함수 반환 시 트램펄린이 호출됩니다. rh->handler를 호출하여 사용자 핸들러(fprobe의 exit_handler)를 실행합니다.
5. 반환 (rethook_recycle): objpool_push로 노드를 풀에 되돌립니다. 다음 함수 호출에서 재사용됩니다.

성능 벤치마크와 대안 비교

objpool의 성능을 다른 커널 오브젝트 풀링/할당 메커니즘과 정량적으로 비교합니다. 벤치마크는 동일한 크기의 오브젝트를 반복적으로 할당/해제하는 시나리오에서 측정합니다.

벤치마크 방법론

다음 4가지 메커니즘을 비교합니다:

• objpool: per-CPU LIFO, lock-free, NMI-safe
• kmem_cache (SLUB): 범용 슬랩 할당자, per-CPU freelist 보유, NMI-unsafe
• mempool: 예약 풀 + 백업 할당자, spin_lock_irqsave 사용, NMI-unsafe
• per-CPU freelist (llist): Linux 6.4 이전 kretprobe 방식, CAS 기반 전역 리스트

/* 벤치마크 코드 개요 (간략화) */
#define BENCH_ITERATIONS  1000000
#define OBJ_SIZE          64

static void bench_objpool(struct objpool_head *pool)
{
    int i;
    ktime_t start = ktime_get();
    for (i = 0; i < BENCH_ITERATIONS; i++) {
        void *obj = objpool_pop(pool);
        if (obj)
            objpool_push(obj, pool);
    }
    pr_info("objpool: %lld ns/op\n",
            ktime_to_ns(ktime_sub(ktime_get(), start)) / BENCH_ITERATIONS);
}

/* 대표적인 결과 (x86_64, 4코어, 6.4 커널):
 *
 * 메커니즘           | 단일 CPU  | 4 CPU 병렬 | NMI-safe
 * ------------------|----------|-----------|--------
 * objpool           |   15 ns  |    18 ns  |   O
 * kmem_cache (SLUB) |   25 ns  |    30 ns  |   X
 * mempool           |   45 ns  |   120 ns  |   X
 * llist (freelist)  |   20 ns  |    85 ns  |   O
 *
 * objpool은 단일 CPU에서 가장 빠르고, 멀티 CPU에서도
 * per-CPU 분리 덕분에 거의 선형 스케일링을 달성합니다.
 * llist는 NMI-safe하지만 전역 리스트의 CAS 경합으로
 * 멀티 CPU에서 성능이 급격히 저하됩니다. */

objpool 선택 기준

커널 모듈 예제

objpool을 직접 사용하는 커널 모듈 예제입니다. /proc/objpool_demo 인터페이스를 통해 pop/push 연산을 테스트할 수 있습니다.

모듈 소스 코드

/* objpool_demo.c - objpool 사용 데모 커널 모듈 */
#define pr_fmt(fmt) KBUILD_MODNAME ": " fmt

#include <linux/module.h>
#include <linux/proc_fs.h>
#include <linux/seq_file.h>
#include <linux/objpool.h>
#include <linux/slab.h>

#define POOL_SIZE    32    /* 사전 할당할 오브젝트 수 */

struct demo_object {
    unsigned long id;       /* 오브젝트 식별 번호 */
    unsigned long pop_cpu;  /* 마지막으로 pop한 CPU */
    unsigned long pop_count;/* pop된 총 횟수 */
    ktime_t last_pop_time;  /* 마지막 pop 시각 */
};

static struct objpool_head demo_pool;
static atomic_long_t total_pops = ATOMIC_LONG_INIT(0);
static atomic_long_t total_pushes = ATOMIC_LONG_INIT(0);
static atomic_long_t total_misses = ATOMIC_LONG_INIT(0);

/* 오브젝트 초기화 콜백: objpool_init에서 각 오브젝트마다 호출 */
static int demo_obj_init(void *obj, void *context)
{
    struct demo_object *dobj = obj;
    static atomic_long_t next_id = ATOMIC_LONG_INIT(0);

    dobj->id = atomic_long_inc_return(&next_id);
    dobj->pop_cpu = -1;
    dobj->pop_count = 0;
    dobj->last_pop_time = 0;
    pr_debug("초기화: obj[%lu] at %px\n", dobj->id, dobj);
    return 0;
}

/* 풀 해제 콜백 */
static int demo_pool_fini(struct objpool_head *head, void *context)
{
    pr_info("풀 해제 완료 (total pops=%ld, pushes=%ld, misses=%ld)\n",
            atomic_long_read(&total_pops),
            atomic_long_read(&total_pushes),
            atomic_long_read(&total_misses));
    return 0;
}

/* pop + 작업 + push 테스트 */
static void demo_pop_push(void)
{
    struct demo_object *obj;

    obj = objpool_pop(&demo_pool);
    if (!obj) {
        atomic_long_inc(&total_misses);
        pr_warn("pop 실패: 풀 고갈 (CPU %d)\n",
                raw_smp_processor_id());
        return;
    }

    atomic_long_inc(&total_pops);

    /* 오브젝트 사용: 메타데이터 업데이트 */
    obj->pop_cpu = raw_smp_processor_id();
    obj->pop_count++;
    obj->last_pop_time = ktime_get();

    pr_debug("pop: obj[%lu] on CPU %lu (count=%lu)\n",
             obj->id, obj->pop_cpu, obj->pop_count);

    /* 오브젝트 반환 */
    objpool_push(obj, &demo_pool);
    atomic_long_inc(&total_pushes);
}

/* /proc/objpool_demo 읽기: 풀 상태 표시 */
static int demo_proc_show(struct seq_file *m, void *v)
{
    seq_printf(m, "=== objpool_demo 상태 ===\n");
    seq_printf(m, "풀 크기 (nr_objs):    %d\n", POOL_SIZE);
    seq_printf(m, "오브젝트 크기:         %zu bytes\n",
               sizeof(struct demo_object));
    seq_printf(m, "총 pop 횟수:          %ld\n",
               atomic_long_read(&total_pops));
    seq_printf(m, "총 push 횟수:         %ld\n",
               atomic_long_read(&total_pushes));
    seq_printf(m, "총 miss 횟수:         %ld\n",
               atomic_long_read(&total_misses));
    seq_printf(m, "num_possible_cpus:    %d\n",
               num_possible_cpus());
    return 0;
}

/* /proc/objpool_demo 쓰기: "pop" 명령으로 테스트 실행 */
static ssize_t demo_proc_write(struct file *file,
                                const char __user *buf,
                                size_t count, loff_t *ppos)
{
    char cmd[16];
    size_t len = min(count, sizeof(cmd) - 1);

    if (copy_from_user(cmd, buf, len))
        return -EFAULT;
    cmd[len] = '\0';

    if (strncmp(cmd, "pop", 3) == 0) {
        demo_pop_push();
    } else if (strncmp(cmd, "burst", 5) == 0) {
        int i;
        for (i = 0; i < 100; i++)
            demo_pop_push();
        pr_info("100회 burst pop/push 완료\n");
    }
    return count;
}

static int demo_proc_open(struct inode *inode, struct file *file)
{
    return single_open(file, demo_proc_show, NULL);
}

static const struct proc_ops demo_proc_ops = {
    .proc_open    = demo_proc_open,
    .proc_read    = seq_read,
    .proc_write   = demo_proc_write,
    .proc_lseek   = seq_lseek,
    .proc_release = single_release,
};

static struct proc_dir_entry *proc_entry;

static int __init demo_init(void)
{
    int ret;

    /* objpool 초기화 */
    ret = objpool_init(&demo_pool,
                       POOL_SIZE,
                       sizeof(struct demo_object),
                       GFP_KERNEL,
                       NULL,             /* context */
                       demo_obj_init,    /* init 콜백 */
                       demo_pool_fini);  /* fini 콜백 */
    if (ret) {
        pr_err("objpool_init 실패: %d\n", ret);
        return ret;
    }

    /* /proc 인터페이스 생성 */
    proc_entry = proc_create("objpool_demo", 0666, NULL,
                              &demo_proc_ops);
    if (!proc_entry) {
        objpool_fini(&demo_pool);
        return -ENOMEM;
    }

    pr_info("모듈 로드: %d개 오브젝트, 각 %zu 바이트\n",
            POOL_SIZE, sizeof(struct demo_object));
    return 0;
}

static void __exit demo_exit(void)
{
    proc_remove(proc_entry);
    objpool_fini(&demo_pool);
    pr_info("모듈 언로드\n");
}

module_init(demo_init);
module_exit(demo_exit);
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("objpool 데모 커널 모듈");
MODULE_AUTHOR("Linux Kernel Documentation");

Makefile

# Makefile - objpool_demo 빌드
obj-m += objpool_demo.o

# 커널 빌드 디렉터리 (환경에 맞게 수정)
KDIR ?= /lib/modules/$(shell uname -r)/build

all:
	$(MAKE) -C $(KDIR) M=$(PWD) modules

clean:
	$(MAKE) -C $(KDIR) M=$(PWD) clean

# 사용법:
#   make                              # 모듈 빌드
#   sudo insmod objpool_demo.ko       # 모듈 로드
#   cat /proc/objpool_demo            # 상태 확인
#   echo "pop" > /proc/objpool_demo   # 단일 pop/push 테스트
#   echo "burst" > /proc/objpool_demo # 100회 burst 테스트
#   cat /proc/objpool_demo            # 결과 확인
#   sudo rmmod objpool_demo           # 모듈 언로드
#   dmesg | tail -20                  # 커널 로그 확인

모듈 테스트 및 검증

모듈을 로드한 후 다음 명령으로 동작을 검증할 수 있습니다:

## 1. 모듈 빌드 및 로드
make
sudo insmod objpool_demo.ko

## 2. 초기 상태 확인
cat /proc/objpool_demo
# === objpool_demo 상태 ===
# 풀 크기 (nr_objs):    32
# 오브젝트 크기:         32 bytes
# 총 pop 횟수:          0
# 총 push 횟수:         0
# 총 miss 횟수:         0
# num_possible_cpus:    4

## 3. 단일 pop/push 테스트
echo "pop" > /proc/objpool_demo
cat /proc/objpool_demo
# 총 pop 횟수:          1
# 총 push 횟수:         1

## 4. burst 테스트 (100회 연속 pop/push)
echo "burst" > /proc/objpool_demo
cat /proc/objpool_demo
# 총 pop 횟수:          101
# 총 push 횟수:         101
# 총 miss 횟수:         0

## 5. 커널 로그 확인
dmesg | grep objpool_demo
# objpool_demo: 모듈 로드: 32개 오브젝트, 각 32 바이트
# objpool_demo: 100회 burst pop/push 완료

## 6. 멀티 코어 테스트 (여러 터미널에서 동시 실행)
# 터미널 1:
for i in $(seq 1 100); do echo "burst" > /proc/objpool_demo; done
# 터미널 2:
for i in $(seq 1 100); do echo "burst" > /proc/objpool_demo; done

## 7. 결과 확인: miss가 0이면 per-CPU 분산이 잘 동작하는 것
cat /proc/objpool_demo
# 총 pop 횟수:          20101
# 총 push 횟수:         20101
# 총 miss 횟수:         0   ← 오브젝트 충분

## 8. 모듈 언로드
sudo rmmod objpool_demo
dmesg | tail -5
# objpool_demo: 풀 해제 완료 (total pops=20101, pushes=20101, misses=0)
# objpool_demo: 모듈 언로드

오류 처리 패턴

objpool을 사용하는 모듈에서 주의해야 할 오류 처리 패턴을 정리합니다:

• objpool_init 실패: 메모리 부족 시 -ENOMEM을 반환합니다. 이 경우 objpool_fini를 호출하면 안 됩니다 -- 부분적으로 할당된 슬롯은 objpool_init 내부에서 자동으로 정리됩니다.
• objpool_pop 실패: 모든 오브젝트가 사용 중이면 NULL을 반환합니다. 이는 오류가 아니라 정상적인 풀 고갈 상태입니다. nmissed 카운터를 증가시키고 graceful하게 처리해야 합니다.
• objpool_fini 순서: objpool_fini는 모든 pop된 오브젝트가 push로 반환된 후에 호출해야 합니다. 미반환 오브젝트가 있으면 메모리 누수가 발생합니다.
• 콜백에서의 오류: objpool_init_obj_cb가 0이 아닌 값을 반환하면 objpool_init이 실패합니다. 콜백 내부에서 추가 리소스를 할당하는 경우 롤백 로직이 필요합니다.

커널 소스 심층 분석

이 섹션에서는 objpool의 실제 커널 소스 코드를 단계별로 추적하며, 초기화부터 해제까지의 전체 흐름을 분석합니다. Linux 6.4+ 기준으로 include/linux/objpool.h와 lib/objpool.c의 핵심 구현을 다룹니다.

objpool_init: per-CPU 슬롯 메모리 배치

objpool_init은 per-CPU 슬롯을 할당하고 오브젝트를 라운드 로빈으로 분배합니다. 이 과정에서 NUMA 노드 인식 할당과 캐시 라인 정렬(Cache Line Alignment)이 이루어집니다.

/* lib/objpool.c - objpool_init 전체 흐름 (v6.4+ 기준) */
int objpool_init(struct objpool_head *pool,
                 int nr_objs, int object_size,
                 gfp_t gfp, void *context,
                 objpool_init_obj_cb objinit,
                 objpool_fini_cb release)
{
    int rc, cpu, i;

    /* 1단계: 메타데이터 초기화 */
    memset(pool, 0, sizeof(*pool));
    pool->nr_objs  = nr_objs;
    pool->obj_size = ALIGN(object_size, sizeof(void *));
    pool->nr_cpus  = num_possible_cpus();
    pool->gfp      = gfp & ~__GFP_ZERO;
    pool->context  = context;
    pool->release  = release;

    /* 2단계: capacity 계산 - 2의 거듭제곱으로 올림 */
    pool->capacity = roundup_pow_of_two(
        nr_objs / pool->nr_cpus + nr_objs);

    /* 3단계: per-CPU 슬롯 배열 할당
     * __alloc_percpu는 각 CPU에 독립적 메모리를 배치하며,
     * 캐시 라인 경계에 정렬됩니다 (일반적으로 64바이트) */
    pool->slots = __alloc_percpu(
        sizeof(struct objpool_slot) +
            pool->capacity * sizeof(void *),
        sizeof(void *));
    if (!pool->slots)
        return -ENOMEM;

    /* 4단계: 각 CPU 슬롯 초기화 */
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot =
            per_cpu_ptr(pool->slots, cpu);
        slot->head = 0;
        slot->tail = 0;
        slot->last = pool->capacity;
        slot->mask = pool->capacity - 1;
    }

    /* 5단계: 오브젝트 사전 할당 및 라운드 로빈 분배
     * NUMA 인식: 각 오브젝트를 해당 CPU의 NUMA 노드에서 할당 */
    for (i = 0; i < nr_objs; i++) {
        void *obj;
        cpu = i % pool->nr_cpus;

        /* kvmalloc_node: NUMA 로컬 메모리에서 할당 */
        obj = kvmalloc_node(pool->obj_size, pool->gfp,
                             cpu_to_node(cpu));
        if (!obj) {
            objpool_free(pool);
            return -ENOMEM;
        }

        /* 사용자 초기화 콜백 호출 */
        if (objinit) {
            rc = objinit(obj, context);
            if (rc) {
                kvfree(obj);
                objpool_free(pool);
                return rc;
            }
        }

        /* 해당 CPU 슬롯에 push */
        {
            struct objpool_slot *slot =
                per_cpu_ptr(pool->slots, cpu);
            slot->entries[slot->head & slot->mask] = obj;
            slot->head++;
        }
    }

    /* 6단계: 참조 카운트 설정
     * nr_objs + 1: 각 오브젝트에 1, 풀 자체에 1 */
    refcount_set(&pool->ref, nr_objs + 1);

    return 0;
}

objpool_pop: Fast Path vs Slow Path (Steal)

objpool_pop은 두 가지 경로를 가집니다. Fast Path는 현재 CPU의 로컬 슬롯에서 즉시 오브젝트를 꺼내는 것이고, Slow Path는 로컬 슬롯이 비어있을 때 다른 CPU의 슬롯에서 오브젝트를 가져오는(steal) 것입니다.

/* objpool_pop 내부 흐름 분석 */

/* Fast Path: 로컬 슬롯에 오브젝트 존재 (~15ns) */
void *objpool_pop(struct objpool_head *pool)
{
    int cpu = raw_smp_processor_id();
    struct objpool_slot *slot = per_cpu_ptr(pool->slots, cpu);
    uint32_t head = READ_ONCE(slot->head);
    uint32_t tail = READ_ONCE(slot->tail);

    /* Fast Path: 로컬 슬롯에 오브젝트가 있음 */
    if (likely(head != tail)) {
        /* CAS로 head 전진 (NMI 재진입 보호) */
        if (try_cmpxchg_release(&slot->head, &head, head + 1))
            return slot->entries[head & slot->mask];
        /* CAS 실패 = NMI가 먼저 pop → 재시도 */
    }

    /* Slow Path: 로컬 슬롯이 비어있거나 CAS 실패 */
    return objpool_pop_slow(pool, cpu);
}

/* Slow Path: 다른 CPU 슬롯에서 steal (~50~200ns) */
static noinline void *objpool_pop_slow(
    struct objpool_head *pool, int cpu)
{
    int i;
    /* 현재 CPU 다음부터 모든 CPU를 순회 */
    for (i = 1; i < pool->nr_cpus; i++) {
        int target = (cpu + i) % pool->nr_cpus;
        struct objpool_slot *slot =
            per_cpu_ptr(pool->slots, target);
        uint32_t head, tail;

        head = READ_ONCE(slot->head);
        do {
            tail = READ_ONCE(slot->tail);
            if (head == tail)
                break;  /* 이 슬롯도 비어있음 */
        } while (!try_cmpxchg_release(
                &slot->head, &head, head + 1));

        if (head != tail)
            return slot->entries[head & slot->mask];
    }
    return NULL;  /* 모든 CPU가 비어있음: 풀 고갈 */
}

objpool_fini: RCU와 참조 카운팅을 통한 안전한 해제

objpool_fini는 즉시 메모리를 해제하지 않습니다. 대신 참조 카운팅(Reference Counting)을 활용하여 모든 사용 중인 오브젝트가 반환된 후에 비로소 release 콜백을 호출합니다. 이 메커니즘은 RCU(Read-Copy-Update)의 지연 해제 패턴과 유사합니다.

/* lib/objpool.c - objpool_fini 상세 흐름 */
void objpool_fini(struct objpool_head *pool)
{
    int count, cpu;

    /* 1단계: 풀에 남아있는 (미사용) 오브젝트 수 계산 */
    count = 0;
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot =
            per_cpu_ptr(pool->slots, cpu);
        count += READ_ONCE(slot->head) -
                 READ_ONCE(slot->tail);
    }

    /* 2단계: 미사용 오브젝트의 참조를 한 번에 해제
     * 이 오브젝트들은 아무도 사용하지 않으므로 안전 */
    if (count > 0)
        refcount_sub(count, &pool->ref);

    /* 3단계: 풀 자체의 참조(-1) 해제
     * 사용 중인 오브젝트가 없으면 여기서 ref=0 → release */
    if (refcount_dec_and_test(&pool->ref))
        objpool_free(pool);

    /* 사용 중인 오브젝트가 있는 경우:
     * ref = (nr_objs + 1) - count - 1 = 사용중_오브젝트수
     * 각 오브젝트가 push될 때 refcount_dec_and_test가
     * 호출되며, 마지막 오브젝트 반환 시 ref=0 → release */
}

/* objpool_free: 실제 리소스 해제 */
static void objpool_free(struct objpool_head *pool)
{
    int cpu;

    if (!pool->slots)
        return;

    /* 각 CPU 슬롯의 오브젝트를 해제 */
    for_each_possible_cpu(cpu) {
        struct objpool_slot *slot =
            per_cpu_ptr(pool->slots, cpu);
        uint32_t h = slot->head, t = slot->tail;
        while (h != t) {
            void *obj = slot->entries[h & slot->mask];
            if (obj)
                kvfree(obj);
            h++;
        }
    }

    /* per-CPU 슬롯 메모리 해제 */
    free_percpu(pool->slots);
    pool->slots = NULL;

    /* release 콜백 호출 (호출자의 상위 구조체 해제) */
    if (pool->release)
        pool->release(pool, pool->context);
}

objpool_slot 구조체 내부: 순환 버퍼 설계

objpool_slot은 per-CPU 순환 버퍼(Circular Buffer)입니다. 이 구조체가 어떻게 LIFO 스택을 구현하면서도 NMI-safe를 유지하는지 상세히 분석합니다.

/* objpool_slot 메모리 레이아웃 상세
 *
 * 주소:  | offset 0  | offset 4  | offset 8  | offset 12 |
 *        |-----------|-----------|-----------|-----------|
 *        |   head    |   tail    |   last    |   mask    |
 *        |  (4바이트)  |  (4바이트)  |  (4바이트)  |  (4바이트)  |
 *
 * 주소:  | offset 16         | offset 24         | ...
 *        |-------------------|-------------------|---
 *        | entries[0] (8B)   | entries[1] (8B)   | ...
 *        | (void * 포인터)    | (void * 포인터)    |
 *
 * 총 크기: 16 + capacity * 8 바이트
 * capacity=16일 때: 16 + 128 = 144 바이트
 *
 * head와 tail의 관계:
 * - head >= tail 항상 성립 (head는 pop에서 증가)
 * - head == tail: 슬롯이 비어 있음
 * - head - tail == capacity: 슬롯이 가득 참 (이론상)
 * - 실제 entries[] 인덱스: value & mask (순환)
 *
 * 단조 증가 설계의 이점:
 * - head와 tail 각각 32비트 → 2^32까지 증가 가능
 * - ABA 문제 방지: 같은 인덱스로 돌아오려면 2^32회 순환 필요
 * - uint32_t 오버플로 시에도 차이 연산(head-tail)은 정확 */

struct objpool_slot {
    uint32_t head;      /* consumer index (pop 시 증가) */
    uint32_t tail;      /* producer index (push 시 증가) */
    uint32_t last;      /* capacity (최대 항목 수) */
    uint32_t mask;      /* capacity - 1 (비트 AND 마스크) */
    void    *entries[];  /* flexible array: 오브젝트 포인터 */
};

/* 핵심 불변식(Invariant):
 * 1) 0 <= (head - tail) <= last
 * 2) entries[head & mask] .. entries[(tail-1) & mask] 이 유효 범위
 * 3) head는 pop에서만 증가, tail은 push에서만 증가
 * 4) 동일 CPU에서 pop은 CAS, push는 WRITE_ONCE (비대칭)
 * 5) cross-CPU steal 시에만 원격 슬롯의 head를 CAS로 전진 */

mempool_t와의 상세 비교: 언제 무엇을 사용해야 하는가

mempool_t(include/linux/mempool.h)과 objpool은 모두 사전 할당 풀이지만, 설계 목표와 사용 컨텍스트가 근본적으로 다릅니다.

/* mempool_t 사용 예시 (비교용) */
static struct kmem_cache *my_cache;
static mempool_t *my_mempool;

/* 초기화: 16개 오브젝트를 사전 할당 */
my_cache = kmem_cache_create("my_obj", 128, 0, 0, NULL);
my_mempool = mempool_create_slab_pool(16, my_cache);

/* 할당: spinlock 획득 → NMI에서 불가 */
void *obj = mempool_alloc(my_mempool, GFP_KERNEL);
/* GFP_KERNEL: 메모리 부족 시 대기 (sleep) 가능
 * GFP_ATOMIC: 대기 불가, 풀 고갈 시 NULL */

/* 해제 */
mempool_free(obj, my_mempool);

/* 정리 */
mempool_destroy(my_mempool);
kmem_cache_destroy(my_cache);

/* 결론:
 * - NMI/하드IRQ에서 안전한 할당이 필요 → objpool
 * - OOM에서도 할당 보장이 필요 → mempool_t
 * - 두 요구사항이 동시에 필요한 경우는 없음
 *   (NMI에서는 대기 불가, 대기가 가능하면 NMI가 아님) */

구현 예제

kretprobe 오브젝트 풀 사용 예제

kretprobe를 등록하고 함수 반환 값을 추적하는 커널 모듈입니다. objpool이 내부적으로 어떻게 사용되는지 보여줍니다.

/* kretprobe_trace.c - kretprobe를 통한 함수 반환값 추적 */
#include <linux/module.h>
#include <linux/kprobes.h>
#include <linux/slab.h>

/* 사용자 정의 데이터: kretprobe_instance.data[]에 저장 */
struct trace_data {
    ktime_t entry_time;  /* 함수 진입 시각 */
    int     entry_cpu;   /* 진입 시 CPU 번호 */
};

/* 함수 진입 핸들러: objpool_pop이 내부에서 호출됨 */
static int entry_handler(struct kretprobe_instance *ri,
                          struct pt_regs *regs)
{
    struct trace_data *data = (struct trace_data *)ri->data;

    /* 이 시점에서 ri는 이미 objpool에서 pop된 상태입니다.
     * pre_handler_kretprobe → objpool_pop → 이 핸들러 */
    data->entry_time = ktime_get();
    data->entry_cpu  = raw_smp_processor_id();

    return 0;  /* 0: 정상 진행, 음수: 후킹 취소 */
}

/* 함수 반환 핸들러: 이후 objpool_push가 호출됨 */
static int ret_handler(struct kretprobe_instance *ri,
                        struct pt_regs *regs)
{
    struct trace_data *data = (struct trace_data *)ri->data;
    unsigned long retval = regs_return_value(regs);
    s64 delta_ns = ktime_to_ns(
        ktime_sub(ktime_get(), data->entry_time));

    pr_info("do_sys_openat2: ret=%ld, latency=%lldns, "
            "entry_cpu=%d, exit_cpu=%d\n",
            retval, delta_ns,
            data->entry_cpu, raw_smp_processor_id());

    /* 이 핸들러 반환 후 ri는 objpool_push로 풀에 반환됩니다.
     * cross-CPU 반환 가능: entry_cpu != exit_cpu일 수 있음 */
    return 0;
}

static struct kretprobe my_kretprobe = {
    .handler     = ret_handler,
    .entry_handler = entry_handler,
    .data_size   = sizeof(struct trace_data),
    .maxactive   = 0,  /* 0 = 기본값 max(10, 2*num_cpus) */
    .kp.symbol_name = "do_sys_openat2",
};

static int __init trace_init(void)
{
    int ret = register_kretprobe(&my_kretprobe);
    if (ret) {
        pr_err("kretprobe 등록 실패: %d\n", ret);
        return ret;
    }

    /* 이 시점에서 objpool이 초기화되어 있습니다:
     * - maxactive개의 kretprobe_instance가 사전 할당
     * - 각 instance 크기 = sizeof(kretprobe_instance) + sizeof(trace_data)
     * - per-CPU 슬롯에 라운드 로빈으로 분배 */
    pr_info("kretprobe 등록: %s (maxactive=%d)\n",
            my_kretprobe.kp.symbol_name,
            my_kretprobe.maxactive);
    return 0;
}

static void __exit trace_exit(void)
{
    unregister_kretprobe(&my_kretprobe);
    pr_info("kretprobe 해제: nmissed=%lu\n",
            my_kretprobe.nmissed);
    /* unregister_kretprobe 내부에서 objpool_fini가 호출됩니다.
     * 사용 중인 인스턴스가 있으면 반환 후 release 콜백 실행 */
}

module_init(trace_init);
module_exit(trace_exit);
MODULE_LICENSE("GPL");

커스텀 고빈도 할당 풀 예제

네트워크 패킷 처리와 같은 고빈도 할당 시나리오에서 objpool을 사용하는 예제입니다. 인터럽트 컨텍스트에서 안전하게 버퍼를 할당/해제합니다.

/* highfreq_pool.c - 고빈도 할당용 커스텀 objpool 예제 */
#include <linux/module.h>
#include <linux/objpool.h>
#include <linux/interrupt.h>
#include <linux/hrtimer.h>

#define POOL_OBJS     64   /* CPU당 16개 기준 (4 CPU) */
#define BUF_SIZE      256  /* 각 버퍼 크기 */

struct pool_context {
    struct objpool_head pool;
    struct hrtimer      timer;
    atomic_long_t      alloc_count;
    atomic_long_t      free_count;
    atomic_long_t      miss_count;
};

struct packet_buf {
    u64  timestamp;
    u16  len;
    u8   cpu;
    u8   flags;
    char data[BUF_SIZE - 12];  /* 패딩 조정 */
};

static struct pool_context *ctx;

/* 오브젝트 초기화 콜백 */
static int buf_init(void *obj, void *context)
{
    memset(obj, 0, sizeof(struct packet_buf));
    return 0;
}

/* 풀 해제 콜백: 컨텍스트 구조체 해제 */
static int pool_release(struct objpool_head *head, void *context)
{
    struct pool_context *pctx = context;
    pr_info("풀 해제: alloc=%ld, free=%ld, miss=%ld\n",
            atomic_long_read(&pctx->alloc_count),
            atomic_long_read(&pctx->free_count),
            atomic_long_read(&pctx->miss_count));
    kfree(pctx);
    return 0;
}

/* 고빈도 할당/해제 시뮬레이션 (hrtimer 콜백, IRQ 컨텍스트) */
static enum hrtimer_restart timer_callback(struct hrtimer *timer)
{
    struct pool_context *pctx =
        container_of(timer, struct pool_context, timer);
    struct packet_buf *buf;

    /* IRQ 컨텍스트에서 objpool_pop 호출: 안전 */
    buf = objpool_pop(&pctx->pool);
    if (!buf) {
        atomic_long_inc(&pctx->miss_count);
        goto restart;
    }

    atomic_long_inc(&pctx->alloc_count);

    /* 버퍼 사용: 타임스탬프와 CPU 기록 */
    buf->timestamp = ktime_get_ns();
    buf->cpu = raw_smp_processor_id();
    buf->len = 64;

    /* 즉시 반환 (실제로는 DMA 완료 후 반환 등) */
    objpool_push(buf, &pctx->pool);
    atomic_long_inc(&pctx->free_count);

restart:
    hrtimer_forward_now(timer, ns_to_ktime(100000));  /* 100us */
    return HRTIMER_RESTART;
}

static int __init highfreq_init(void)
{
    int ret;

    ctx = kzalloc(sizeof(*ctx), GFP_KERNEL);
    if (!ctx)
        return -ENOMEM;

    /* objpool 초기화: 64개의 packet_buf 사전 할당 */
    ret = objpool_init(&ctx->pool,
                       POOL_OBJS,
                       sizeof(struct packet_buf),
                       GFP_KERNEL,
                       ctx,
                       buf_init,
                       pool_release);
    if (ret) {
        kfree(ctx);
        return ret;
    }

    /* hrtimer 시작: 100us 간격으로 pop/push 반복 */
    hrtimer_init(&ctx->timer, CLOCK_MONOTONIC,
                 HRTIMER_MODE_REL);
    ctx->timer.function = timer_callback;
    hrtimer_start(&ctx->timer,
                  ns_to_ktime(100000),
                  HRTIMER_MODE_REL);

    pr_info("고빈도 풀 시작: %d개 오브젝트, %zuB/obj\n",
            POOL_OBJS,
            sizeof(struct packet_buf));
    return 0;
}

static void __exit highfreq_exit(void)
{
    hrtimer_cancel(&ctx->timer);
    /* objpool_fini: 모든 오브젝트 반환 후 pool_release 호출
     * pool_release에서 ctx를 kfree하므로 이후 ctx 접근 불가 */
    objpool_fini(&ctx->pool);
}

module_init(highfreq_init);
module_exit(highfreq_exit);
MODULE_LICENSE("GPL");

참고 자료

• include/linux/objpool.h -- objpool_head, objpool_pop/push 인라인 함수 정의
• lib/objpool.c -- objpool_init, objpool_fini, per-CPU 슬롯 관리 구현
• kernel/kprobes.c -- kretprobe의 objpool 활용 코드
• kernel/trace/rethook.c -- rethook의 objpool 활용 코드
• kernel/trace/fprobe.c -- fprobe의 rethook(objpool) 활용
• LKML: objpool 패치 시리즈 -- Wu Qiang의 objpool 도입 패치 (v6.4)
• LWN: A scalable, lock-free object pool -- objpool 설계 배경 기사

관련 문서

• 메모리 관리 개요 -- per-CPU 할당, NUMA 노드 개념
• Slab Allocator (SLUB) -- kmem_cache 기반 오브젝트 캐싱
• Per-CPU 변수 -- per-CPU 데이터 구조의 기본 원리
• 메모리 배리어 (Memory Barrier) -- smp_wmb/smp_rmb 동작 원리
• Atomic 연산 -- refcount_t, READ_ONCE/WRITE_ONCE
• Lockdep -- lock-free 메커니즘의 정확성 검증
• ftrace -- kprobe/fprobe/rethook 트레이싱 프레임워크
• 동기화 기법 -- lock-free 프로그래밍과 동기화 원리