Bottom Half 심화 (Workqueue · Softirq · Tasklet)

softirq 내부 구조, tasklet 상태 머신, CMWQ workqueue 아키텍처, worker pool, Bottom Half 선택 가이드를 심층적으로 다룹니다.

Softirq 심화

softirq는 커널에 정적으로 컴파일되는 Bottom Half 메커니즘으로, 10개의 고정 타입이 존재합니다. 네트워킹, 블록 I/O, 타이머 등 성능이 극도로 중요한 서브시스템에서 사용됩니다. 새로운 softirq를 추가하는 것은 커널 서브시스템 수준의 결정이며, 드라이버에서는 사용하지 않습니다.

open_softirq() 등록 API

softirq 핸들러는 커널 초기화 시 open_softirq()로 등록됩니다. 각 softirq 타입에 대해 하나의 핸들러만 존재하며, 런타임에 변경할 수 없습니다:

/* kernel/softirq.c */
static struct softirq_action softirq_vec[NR_SOFTIRQS];

void open_softirq(int nr, void (*action)(struct softirq_action *))
{
    softirq_vec[nr].action = action;
}

/* 각 서브시스템의 초기화 코드에서 등록 */
open_softirq(NET_TX_SOFTIRQ, net_tx_action);   /* net/core/dev.c */
open_softirq(NET_RX_SOFTIRQ, net_rx_action);   /* net/core/dev.c */
open_softirq(TASKLET_SOFTIRQ, tasklet_action);  /* kernel/softirq.c */
open_softirq(HI_SOFTIRQ, tasklet_hi_action);    /* kernel/softirq.c */
open_softirq(TIMER_SOFTIRQ, run_timer_softirq); /* kernel/time/timer.c */
open_softirq(SCHED_SOFTIRQ, run_rebalance_domains); /* kernel/sched/fair.c */
open_softirq(RCU_SOFTIRQ, rcu_core_si);         /* kernel/rcu/tree.c */

raise_softirq() vs raise_softirq_irqoff()

softirq를 트리거하려면 pending 비트를 설정해야 합니다. 두 가지 API가 존재합니다:

/* 일반적인 사용: 인터럽트 상태를 자동으로 저장/복원 */
void raise_softirq(unsigned int nr)
{
    unsigned long flags;
    local_irq_save(flags);       /* IRQ 비활성화 + 플래그 저장 */
    raise_softirq_irqoff(nr);
    local_irq_restore(flags);    /* 이전 IRQ 상태 복원 */
}

/* 이미 IRQ가 비활성화된 상태에서 사용 (top half 내부 등) */
void raise_softirq_irqoff(unsigned int nr)
{
    __raise_softirq_irqoff(nr);  /* or_softirq_pending(1 << nr) */
    /* 인터럽트 컨텍스트가 아니면 ksoftirqd를 깨움 */
    if (!in_interrupt())
        wakeup_softirqd();
}

/* 사용 예: hard IRQ 핸들러 내부 (이미 IRQ 비활성 상태) */
static irqreturn_t my_handler(int irq, void *dev)
{
    /* ... 최소 작업 ... */
    raise_softirq_irqoff(NET_RX_SOFTIRQ);  /* 효율적 */
    return IRQ_HANDLED;
}

__do_softirq() 내부

__do_softirq()는 softirq의 핵심 실행 루프입니다. pending 비트맵을 순회하며 등록된 핸들러를 호출하되, starvation 방지를 위한 제한이 존재합니다:

/* kernel/softirq.c - 핵심 실행 루프 (간략화) */
#define MAX_SOFTIRQ_TIME    msecs_to_jiffies(2)   /* 최대 2ms */
#define MAX_SOFTIRQ_RESTART 10                    /* 최대 10회 재시작 */

asmlinkage __visible void __do_softirq(void)
{
    unsigned long end = jiffies + MAX_SOFTIRQ_TIME;
    int max_restart = MAX_SOFTIRQ_RESTART;
    struct softirq_action *h;
    __u32 pending;

    pending = local_softirq_pending();   /* Per-CPU pending 비트 읽기 */

    __local_bh_disable_ip(_RET_IP_, SOFTIRQ_OFFSET);

restart:
    set_softirq_pending(0);              /* pending 클리어 */
    local_irq_enable();                  /* IRQ 재활성화 */

    h = softirq_vec;
    while (pending) {
        if (pending & 1) {
            h->action(h);               /* softirq 핸들러 호출 */
        }
        h++;
        pending >>= 1;
    }

    local_irq_disable();
    pending = local_softirq_pending();   /* 새로 발생한 softirq 확인 */

    /* 재시작 조건: pending 있고, 횟수/시간 제한 이내 */
    if (pending && --max_restart &&
        time_before(jiffies, end))
        goto restart;

    /* 제한 초과: ksoftirqd에 위임 */
    if (pending)
        wakeup_softirqd();

    __local_bh_enable(SOFTIRQ_OFFSET);
}

ksoftirqd 생명주기

ksoftirqd는 Per-CPU 커널 스레드로, softirq 처리의 폴백 경로를 담당합니다:

/* kernel/softirq.c - ksoftirqd 메인 루프 (간략화) */
static int ksoftirqd_should_run(unsigned int cpu)
{
    return local_softirq_pending();
}

static void run_ksoftirqd(unsigned int cpu)
{
    local_irq_disable();
    if (local_softirq_pending()) {
        __do_softirq();
        local_irq_enable();
        cond_resched();         /* 다른 태스크에 CPU 양보 */
        return;
    }
    local_irq_enable();
}

/*
 * ksoftirqd 특성:
 * - Per-CPU: ksoftirqd/0, ksoftirqd/1, ...
 * - 스케줄링 정책: SCHED_NORMAL (nice 0)
 * - 깨어나는 조건:
 *   1. __do_softirq()에서 시간/횟수 제한 초과
 *   2. 인터럽트 비활성 상태에서 raise_softirq() 호출
 *   3. local_bh_enable()에서 pending softirq 발견
 * - ksoftirqd는 일반 프로세스와 동일한 우선순위로 스케줄링됨
 *   → 높은 부하에서 softirq 지연 발생 가능 (의도된 동작)
 */

Per-CPU 동시성 모델

softirq의 가장 중요한 특성은 같은 softirq 타입이 여러 CPU에서 동시에 실행될 수 있다는 점입니다. 이는 높은 성능을 제공하지만, 공유 데이터에 대한 동기화가 필수입니다:

/*
 * Softirq 동시성 규칙:
 *
 * 1. 같은 softirq가 여러 CPU에서 동시 실행 가능
 *    → NET_RX_SOFTIRQ: CPU0과 CPU1에서 동시 실행 가능
 *    → Per-CPU 데이터 사용으로 락 최소화
 *
 * 2. 같은 CPU에서는 softirq가 중첩되지 않음
 *    → softirq 실행 중 동일 CPU의 다른 softirq는 대기
 *
 * 3. Hard IRQ만 softirq를 선점 가능
 *    → softirq 실행 중 동일 CPU의 프로세스는 실행 불가
 *
 * 4. 동기화 패턴:
 *    - softirq 간: spin_lock() (preemption은 이미 비활성)
 *    - softirq + 프로세스: spin_lock_bh() (프로세스 쪽)
 *    - softirq + hard IRQ: spin_lock_irq() (softirq 쪽)
 */

/* 예: 네트워크 수신 softirq의 Per-CPU 데이터 활용 */
DEFINE_PER_CPU(struct softnet_data, softnet_data);

static void net_rx_action(struct softirq_action *h)
{
    /* Per-CPU 데이터 접근 → 락 불필요 */
    struct softnet_data *sd = this_cpu_ptr(&softnet_data);
    struct list_head *list = &sd->poll_list;

    while (!list_empty(list)) {
        /* NAPI polling ... */
    }
}

선점 모드별 동작

Tasklet 심화

tasklet은 softirq(TASKLET_SOFTIRQ, HI_SOFTIRQ) 위에 구축된 동적 Bottom Half입니다. softirq와 달리 런타임에 동적으로 생성/삭제가 가능하며, 같은 tasklet은 절대 병렬 실행되지 않는 직렬화를 보장합니다.

Per-CPU Tasklet 리스트

tasklet은 Per-CPU 단일 연결 리스트에 저장됩니다. 일반 tasklet과 고우선순위 tasklet은 별도 리스트를 사용합니다:

/* kernel/softirq.c */
struct tasklet_head {
    struct tasklet_struct *head;
    struct tasklet_struct **tail;
};

/* Per-CPU 리스트: 일반 tasklet (TASKLET_SOFTIRQ, 우선순위 6) */
static DEFINE_PER_CPU(struct tasklet_head, tasklet_vec);

/* Per-CPU 리스트: 고우선순위 tasklet (HI_SOFTIRQ, 우선순위 0) */
static DEFINE_PER_CPU(struct tasklet_head, tasklet_hi_vec);

/*
 * 리스트 구조 (Per-CPU):
 *
 *  tasklet_vec.head ──▶ [tasklet_A] ──▶ [tasklet_B] ──▶ [tasklet_C] ──▶ NULL
 *                              ↑                                    ↑
 *                         먼저 스케줄됨                       tasklet_vec.tail
 *
 * 특징:
 * - FIFO 순서: 먼저 schedule된 tasklet이 먼저 실행
 * - tail 포인터로 O(1) 추가 보장
 * - tasklet_schedule() 시 현재 CPU의 리스트에 추가
 * - 실행도 같은 CPU에서 (Per-CPU 바인딩)
 */

HI_SOFTIRQ vs TASKLET_SOFTIRQ

/* softirq 우선순위 순서 (낮은 인덱스 = 먼저 실행) */
/*  0: HI_SOFTIRQ        ← tasklet_hi_schedule()로 등록된 tasklet */
/*  1: TIMER_SOFTIRQ     */
/*  2: NET_TX_SOFTIRQ    */
/*  3: NET_RX_SOFTIRQ    */
/*  4: BLOCK_SOFTIRQ     */
/*  5: IRQ_POLL_SOFTIRQ  */
/*  6: TASKLET_SOFTIRQ   ← tasklet_schedule()로 등록된 tasklet */
/*  7: SCHED_SOFTIRQ     */
/*  8: HRTIMER_SOFTIRQ   */
/*  9: RCU_SOFTIRQ       */

/* 고우선순위 tasklet 사용 예 (커널 사운드 서브시스템) */
struct snd_pcm_substream {
    struct tasklet_struct tasklet;  /* 오디오 버퍼 처리 */
    /* ... */
};

/* 오디오는 지연 민감 → HI_SOFTIRQ 사용 */
tasklet_hi_schedule(&substream->tasklet);

tasklet_struct 필드

/* include/linux/interrupt.h */
struct tasklet_struct {
    struct tasklet_struct *next;  /* Per-CPU 리스트의 다음 tasklet */
    unsigned long state;          /* TASKLET_STATE_SCHED | TASKLET_STATE_RUN */
    atomic_t count;               /* 0이면 활성, >0이면 비활성 (disable 카운트) */
    bool use_callback;            /* 새 API (callback) vs 레거시 (func) */
    union {
        void (*callback)(struct tasklet_struct *t); /* 새 API */
        void (*func)(unsigned long data);           /* 레거시 */
    };
    unsigned long data;           /* 레거시 API용 인자 */
};

tasklet_setup() vs tasklet_init() API

커널 5.9에서 tasklet_setup()이 도입되어 기존 tasklet_init()을 대체합니다. 새 API는 from_tasklet() 매크로와 함께 사용하여 타입 안전한 컨테이너 접근을 제공합니다:

/* ====== 새 API (커널 5.9+, 권장) ====== */
#include <linux/interrupt.h>

struct my_device {
    struct tasklet_struct tasklet;
    u32 pending_data;
};

/* 콜백: tasklet_struct 포인터를 직접 받음 */
static void my_tasklet_cb(struct tasklet_struct *t)
{
    /* from_tasklet(): container_of 래퍼 매크로 */
    struct my_device *dev = from_tasklet(dev, t, tasklet);
    process_data(dev->pending_data);
}

/* 초기화: 내부적으로 use_callback = true 설정 */
tasklet_setup(&dev->tasklet, my_tasklet_cb);

/* 정적 선언 매크로 */
DECLARE_TASKLET(name, callback);  /* count=0 (활성) */
DECLARE_TASKLET_DISABLED(name, callback); /* count=1 (비활성) */


/* ====== 레거시 API (deprecated) ====== */

/* 콜백: unsigned long data 인자 사용 → 타입 불안전 */
static void old_tasklet_func(unsigned long data)
{
    struct my_device *dev = (struct my_device *)data;  /* 캐스트 필요 */
    process_data(dev->pending_data);
}

/* 초기화: func + data 방식 (use_callback = false) */
tasklet_init(&dev->tasklet, old_tasklet_func,
            (unsigned long)dev);  /* 캐스트 필요 */

상태 머신

tasklet은 두 개의 상태 비트로 스케줄링과 실행을 제어합니다:

/* 상태 비트 정의 */
enum {
    TASKLET_STATE_SCHED,  /* bit 0: 실행 대기열에 등록됨 */
    TASKLET_STATE_RUN,    /* bit 1: 현재 실행 중 (SMP에서만 의미) */
};

/*
 * 상태 전이:
 *
 * [Idle]  ──tasklet_schedule()──▶ [Scheduled]
 *   state: 0                       state: SCHED
 *   ▲                                   │
 *   │                                   │ softirq 실행 시
 *   │                                   ▼
 *   │                            [Running]
 *   │                              state: SCHED | RUN
 *   │                                   │
 *   └──────── 완료 후 클리어 ◀──────────┘
 *              SCHED, RUN 모두 클리어
 *
 * 핵심 규칙:
 * - SCHED가 이미 set이면 tasklet_schedule()은 no-op
 *   → 같은 tasklet을 여러 번 schedule해도 한 번만 실행
 * - RUN이 set이면 다른 CPU에서 실행 시도 시 건너뜀
 *   → 같은 tasklet의 병렬 실행 방지
 */

tasklet_schedule() 내부 구현

tasklet_schedule()은 tasklet을 Per-CPU 리스트에 추가하고 TASKLET_SOFTIRQ를 트리거합니다. 핵심은 TASKLET_STATE_SCHED 비트의 원자적 test-and-set입니다:

/* kernel/softirq.c - tasklet_schedule() 구현 */
static inline void tasklet_schedule(struct tasklet_struct *t)
{
    /* SCHED 비트가 이미 set이면 false 반환 → 아무것도 안 함 */
    if (!test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
        __tasklet_schedule(t);
}

void __tasklet_schedule(struct tasklet_struct *t)
{
    unsigned long flags;

    local_irq_save(flags);          /* IRQ 비활성화 (Per-CPU 리스트 보호) */

    /* 현재 CPU의 tasklet_vec 리스트 tail에 추가 */
    t->next = NULL;
    *__this_cpu_read(tasklet_vec.tail) = t;
    __this_cpu_write(tasklet_vec.tail, &t->next);

    raise_softirq_irqoff(TASKLET_SOFTIRQ);  /* softirq 트리거 */
    local_irq_restore(flags);
}

/*
 * 호출 컨텍스트별 동작:
 *
 * 1. Hard IRQ 핸들러 (Top Half) 내부:
 *    - 가장 일반적인 사용 패턴
 *    - IRQ가 이미 비활성 상태이므로 local_irq_save는 no-op에 가까움
 *    - irq_exit() 시점에 TASKLET_SOFTIRQ가 실행됨
 *
 * 2. softirq 핸들러 내부:
 *    - 현재 softirq 루프의 다음 반복에서 실행
 *    - __do_softirq()가 pending 비트를 재확인하므로
 *
 * 3. 프로세스 컨텍스트:
 *    - raise_softirq_irqoff() → wakeup_softirqd() 호출
 *    - ksoftirqd에서 실행됨
 *
 * 4. 중복 호출:
 *    - SCHED 비트가 이미 set → test_and_set_bit 실패 → 즉시 반환
 *    - 이미 스케줄된 tasklet은 두 번 큐에 들어가지 않음
 */

직렬화 보장

/*
 * tasklet 직렬화 규칙:
 *
 * 1. 같은 tasklet: 절대 병렬 실행 불가
 *    → TASKLET_STATE_RUN 비트로 보장
 *    → CPU A에서 실행 중이면 CPU B에서는 스킵 후 재스케줄
 *
 * 2. 다른 tasklet: 다른 CPU에서 병렬 실행 가능
 *    → tasklet_A는 CPU0, tasklet_B는 CPU1에서 동시 실행 가능
 *
 * 3. 같은 CPU: tasklet 간 중첩 불가
 *    → softirq 핸들러 내에서 순차 실행
 *
 * 비교:
 * - softirq: 같은 타입도 여러 CPU에서 동시 실행 → 락 필요
 * - tasklet: 같은 인스턴스는 직렬화 → 락 불필요 (해당 데이터에 대해)
 * - workqueue: work item 단위 직렬화
 */

tasklet_disable() / tasklet_enable()

/* count 기반 비활성화/활성화 */
void tasklet_disable(struct tasklet_struct *t)
{
    tasklet_disable_nosync(t);   /* atomic_inc(&t->count) */
    tasklet_unlock_wait(t);      /* RUN 비트 해제 대기 (SMP) */
    smp_mb();
}

/* count > 0이면 tasklet 실행이 보류됨 */
/* 중첩 가능: disable 2번 → enable 2번 필요 */

void tasklet_enable(struct tasklet_struct *t)
{
    smp_mb__before_atomic();
    atomic_dec(&t->count);       /* count가 0이 되면 실행 가능 */
}

/* 사용 패턴: 임시로 tasklet 실행 중지 */
tasklet_disable(&my_tasklet);
/* 이 구간에서 tasklet과 공유하는 데이터를 안전하게 수정 */
/* tasklet이 실행 중이었다면 완료를 기다림 */
tasklet_enable(&my_tasklet);

내부 실행 경로

/* kernel/softirq.c - tasklet_action() 간략화 */
static void tasklet_action(struct softirq_action *a)
{
    struct tasklet_struct *list;

    /* Per-CPU tasklet 리스트를 원자적으로 가져옴 */
    list = __this_cpu_read(tasklet_vec.head);
    __this_cpu_write(tasklet_vec.head, NULL);

    while (list) {
        struct tasklet_struct *t = list;
        list = list->next;

        /* 다른 CPU에서 실행 중인지 확인 */
        if (tasklet_trylock(t)) {        /* test_and_set_bit(RUN, &t->state) */
            if (!atomic_read(&t->count)) { /* disable되지 않았으면 */
                clear_bit(TASKLET_STATE_SCHED, &t->state);
                t->callback(t);              /* tasklet 핸들러 호출 */
                tasklet_unlock(t);           /* clear_bit(RUN, ...) */
                continue;
            }
            tasklet_unlock(t);
        }
        /* 실행 불가: 리스트에 다시 추가하고 재스케줄 */
        tasklet_schedule(t);
    }
}

/* tasklet_hi_action()도 동일한 로직이지만 HI_SOFTIRQ(우선순위 0)에서 실행 */

tasklet_kill() 내부 동작

tasklet_kill()은 tasklet을 안전하게 비활성화하고 해제하는 API입니다. 드라이버 언로드나 디바이스 제거 시 반드시 호출해야 합니다:

/* kernel/softirq.c - tasklet_kill() 구현 (간략화) */
void tasklet_kill(struct tasklet_struct *t)
{
    if (in_interrupt())
        pr_notice("Attempt to kill tasklet from interrupt\n");

    /* 1단계: SCHED 비트가 클리어될 때까지 대기 */
    /*    → 현재 스케줄된 tasklet의 실행 완료를 보장 */
    while (test_and_set_bit(TASKLET_STATE_SCHED, &t->state))
        wait_var_event(&t->state,
                       !test_bit(TASKLET_STATE_SCHED, &t->state));

    /* 2단계: RUN 비트가 클리어될 때까지 대기 (SMP) */
    /*    → 다른 CPU에서 실행 중인 tasklet 완료를 보장 */
    tasklet_unlock_wait(t);

    /* 3단계: SCHED 비트를 set 상태로 유지 */
    /*    → 이후 tasklet_schedule() 호출이 no-op이 됨 */
    /*    → tasklet이 다시 큐에 들어가는 것을 방지 */
}

/*
 * tasklet_kill() 사용 규칙:
 *
 * 1. 프로세스 컨텍스트에서만 호출 (슬립 가능해야 함)
 *    → 인터럽트/softirq 컨텍스트에서 호출하면 데드락!
 *
 * 2. tasklet_kill() 후에는 tasklet_schedule()이 무시됨
 *    → SCHED 비트가 영구적으로 set 상태이므로
 *    → 재사용하려면 tasklet_setup()으로 재초기화 필요
 *
 * 3. 드라이버에서의 올바른 해제 순서:
 */

static void my_remove(struct pci_dev *pdev)
{
    struct my_device *dev = pci_get_drvdata(pdev);

    /* 1. 새 인터럽트 발생 방지 */
    free_irq(dev->irq, dev);

    /* 2. 이미 스케줄된 tasklet 완료 대기 + 비활성화 */
    tasklet_kill(&dev->tasklet);

    /* 3. 이제 안전하게 리소스 해제 */
    kfree(dev->buffer);
    pci_release_regions(pdev);
}

PREEMPT_RT에서의 Tasklet

PREEMPT_RT(실시간) 커널에서는 tasklet의 동작이 크게 변경됩니다. softirq가 더 이상 인터럽트 컨텍스트에서 직접 실행되지 않으므로, tasklet도 영향을 받습니다:

/*
 * PREEMPT_RT에서의 tasklet 변환:
 *
 * 일반 커널:
 *   tasklet → TASKLET_SOFTIRQ → __do_softirq() (인터럽트 컨텍스트)
 *
 * PREEMPT_RT:
 *   tasklet → TASKLET_SOFTIRQ → ksoftirqd 스레드 (프로세스 컨텍스트)
 *
 * 영향:
 * 1. tasklet이 선점 가능해짐
 *    → 일반 커널에서 불가능한 우선순위 역전 발생 가능
 *    → RT 태스크가 tasklet에 의해 지연될 수 있음
 *
 * 2. ksoftirqd 우선순위에 의존
 *    → ksoftirqd는 SCHED_NORMAL(nice 0)
 *    → RT 태스크보다 항상 낮은 우선순위
 *    → 결정적 지연시간 보장이 어려움
 *
 * 3. RT 커널에서의 권장 대안:
 *    → threaded IRQ: 스레드별 우선순위 개별 설정 가능
 *    → workqueue: CMWQ의 유연한 스케줄링 활용
 */

/* PREEMPT_RT에서 tasklet 대신 threaded IRQ 사용 예시 */
/* Before: tasklet 기반 */
request_irq(irq, my_top_half, 0, "dev", priv);
/* top_half 내부에서: tasklet_schedule(&priv->tasklet); */

/* After: threaded IRQ (RT 호환) */
request_threaded_irq(irq, my_top_half, my_threaded_bottom,
                     IRQF_ONESHOT, "dev", priv);
/* top_half: return IRQ_WAKE_THREAD; */
/* my_threaded_bottom: 전용 커널 스레드에서 실행 */
/* → chrt으로 스레드 우선순위 조정 가능:
 *   chrt -f -p 50 $(pgrep irq/XX-dev)
 */

실제 커널 드라이버 사용 예제

실제 커널 소스에서 tasklet을 사용하는 대표적인 패턴입니다:

/* 예시: 네트워크 드라이버의 tasklet 기반 수신 처리 */
struct my_nic {
    struct net_device *netdev;
    struct tasklet_struct rx_tasklet;
    spinlock_t rx_lock;
    struct sk_buff_head rx_queue;  /* 수신 패킷 큐 */
    void __iomem *regs;
};

/* Top Half: 하드웨어 인터럽트 핸들러 */
static irqreturn_t my_nic_irq(int irq, void *dev_id)
{
    struct my_nic *nic = dev_id;
    u32 status = ioread32(nic->regs + IRQ_STATUS);

    if (!(status & RX_IRQ_BIT))
        return IRQ_NONE;

    /* 인터럽트 ACK + 추가 인터럽트 마스킹 */
    iowrite32(RX_IRQ_BIT, nic->regs + IRQ_ACK);
    iowrite32(0, nic->regs + IRQ_MASK);

    /* Bottom Half로 위임 */
    tasklet_schedule(&nic->rx_tasklet);
    return IRQ_HANDLED;
}

/* Bottom Half: tasklet 핸들러 */
static void my_nic_rx_tasklet(struct tasklet_struct *t)
{
    struct my_nic *nic = from_tasklet(nic, t, rx_tasklet);
    struct sk_buff *skb;
    int budget = 64;  /* 한 번에 최대 64개 패킷 처리 */

    spin_lock(&nic->rx_lock);
    while (budget-- > 0) {
        u32 desc_status = ioread32(nic->regs + RX_DESC);
        if (!(desc_status & DESC_READY))
            break;

        skb = netdev_alloc_skb(nic->netdev, desc_status & DESC_LEN_MASK);
        if (!skb) {
            nic->netdev->stats.rx_dropped++;
            continue;
        }

        /* DMA 버퍼에서 skb로 복사 (인터럽트 컨텍스트이므로 GFP_ATOMIC) */
        memcpy(skb_put(skb, desc_status & DESC_LEN_MASK),
               nic->rx_buf, desc_status & DESC_LEN_MASK);

        skb->protocol = eth_type_trans(skb, nic->netdev);
        netif_rx(skb);
        nic->netdev->stats.rx_packets++;
    }
    spin_unlock(&nic->rx_lock);

    /* 인터럽트 재활성화 */
    iowrite32(RX_IRQ_BIT, nic->regs + IRQ_MASK);
}

/* 프로브: 초기화 */
static int my_nic_probe(struct pci_dev *pdev, /* ... */)
{
    struct my_nic *nic;
    /* ... 할당, 매핑 ... */

    tasklet_setup(&nic->rx_tasklet, my_nic_rx_tasklet);
    spin_lock_init(&nic->rx_lock);

    request_irq(pdev->irq, my_nic_irq, IRQF_SHARED, "my_nic", nic);
    return 0;
}

/* 제거: 정리 */
static void my_nic_remove(struct pci_dev *pdev)
{
    struct my_nic *nic = pci_get_drvdata(pdev);

    free_irq(pdev->irq, nic);      /* 1. IRQ 핸들러 해제 */
    tasklet_kill(&nic->rx_tasklet); /* 2. tasklet 완료 대기 + 비활성화 */
    /* 3. 나머지 리소스 해제 ... */
}

Tasklet 디버깅

/*
 * Tasklet 관련 디버깅 기법:
 *
 * 1. /proc/softirqs - tasklet 실행 횟수 확인
 *    $ cat /proc/softirqs
 *                    CPU0       CPU1       CPU2       CPU3
 *          HI:          0          0          0          0    ← 고우선순위 tasklet
 *       TIMER:     123456      98765      87654      76543
 *      NET_TX:        100         50         30         20
 *      NET_RX:      45678      34567      23456      12345
 *       BLOCK:       5678       4567       3456       2345
 *    IRQ_POLL:          0          0          0          0
 *     TASKLET:      12345       8765       6543       4321   ← 일반 tasklet
 *       SCHED:      34567      23456      12345       9876
 *     HRTIMER:         10          8          6          4
 *         RCU:      67890      56789      45678      34567
 *
 * 2. CPU 편향 확인
 *    - TASKLET 카운트가 특정 CPU에 집중되면 부하 불균형
 *    - tasklet은 schedule한 CPU에서만 실행되므로
 *    - Top Half(IRQ)의 affinity가 원인 → /proc/irq/XX/smp_affinity
 *
 * 3. ftrace로 tasklet 실행 추적
 *    $ echo 1 > /sys/kernel/debug/tracing/events/irq/tasklet_entry/enable
 *    $ echo 1 > /sys/kernel/debug/tracing/events/irq/tasklet_exit/enable
 *    $ cat /sys/kernel/debug/tracing/trace
 *    → tasklet 핸들러 진입/종료 시점, 실행 시간 확인
 *
 * 4. lockdep 경고
 *    - CONFIG_PROVE_LOCKING 활성화 시
 *    - tasklet 핸들러에서 mutex 사용 → 경고 발생
 *    - 인터럽트 컨텍스트에서 슬립 시도 감지
 *
 * 5. WARN/BUG 트리거 조건
 *    - tasklet_kill()을 인터럽트 컨텍스트에서 호출
 *    - disable된 tasklet을 kill하지 않고 모듈 언로드
 *    - use-after-free: tasklet_kill() 없이 구조체 해제
 */

Deprecation 이유와 대안

tasklet은 다음과 같은 이유로 deprecated 추세이며, 새 코드에서는 사용하지 않아야 합니다:

Tasklet → Workqueue 마이그레이션

/* ====== Before: tasklet 사용 ====== */
struct my_device {
    struct tasklet_struct tasklet;
    /* ... */
};

static void my_tasklet_handler(struct tasklet_struct *t)
{
    struct my_device *dev = from_tasklet(dev, t, tasklet);
    /* 슬립 불가! spin_lock만 사용 가능 */
    spin_lock(&dev->lock);
    process_data(dev);
    spin_unlock(&dev->lock);
}

/* 초기화 */
tasklet_setup(&dev->tasklet, my_tasklet_handler);

/* IRQ 핸들러에서 */
tasklet_schedule(&dev->tasklet);

/* 해제 */
tasklet_kill(&dev->tasklet);


/* ====== After: workqueue 사용 ====== */
struct my_device {
    struct work_struct work;
    /* ... */
};

static void my_work_handler(struct work_struct *work)
{
    struct my_device *dev = container_of(work, struct my_device, work);
    /* 슬립 가능! mutex, GFP_KERNEL 사용 가능 */
    mutex_lock(&dev->mutex);
    process_data(dev);
    mutex_unlock(&dev->mutex);
}

/* 초기화 */
INIT_WORK(&dev->work, my_work_handler);

/* IRQ 핸들러에서 */
schedule_work(&dev->work);

/* 해제 */
cancel_work_sync(&dev->work);

Tasklet → Threaded IRQ 마이그레이션

특히 하드웨어 인터럽트와 1:1 대응하는 tasklet은 threaded IRQ로 전환하는 것이 더 자연스럽습니다:

/* ====== Before: request_irq + tasklet ====== */
static irqreturn_t my_top_half(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    iowrite32(IRQ_ACK, dev->regs + IRQ_STATUS);
    tasklet_schedule(&dev->tasklet);
    return IRQ_HANDLED;
}

static void my_tasklet_handler(struct tasklet_struct *t)
{
    struct my_device *dev = from_tasklet(dev, t, tasklet);
    spin_lock(&dev->lock);
    process_data(dev);
    spin_unlock(&dev->lock);
}

/* 등록 */
tasklet_setup(&dev->tasklet, my_tasklet_handler);
request_irq(irq, my_top_half, IRQF_SHARED, "dev", dev);


/* ====== After: request_threaded_irq ====== */
static irqreturn_t my_top_half(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    iowrite32(IRQ_ACK, dev->regs + IRQ_STATUS);
    return IRQ_WAKE_THREAD;  /* tasklet 대신 스레드 깨움 */
}

static irqreturn_t my_threaded_handler(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    /* 프로세스 컨텍스트! mutex, GFP_KERNEL 사용 가능 */
    mutex_lock(&dev->mutex);
    process_data(dev);
    mutex_unlock(&dev->mutex);
    return IRQ_HANDLED;
}

/* 등록: tasklet_setup() + request_irq() 대신 단일 호출 */
request_threaded_irq(irq, my_top_half, my_threaded_handler,
                     IRQF_SHARED | IRQF_ONESHOT, "dev", dev);
/* tasklet_kill() 대신 free_irq()로 정리 */

Workqueue 심화 (CMWQ)

Concurrency Managed Workqueue (CMWQ)는 Linux 2.6.36에서 도입된 현대적 workqueue 아키텍처입니다. 기존의 singlethread/multithread workqueue를 대체하여, 커널이 worker pool을 중앙 관리하고 동시성을 자동 제어합니다.

CMWQ 아키텍처 개요

CMWQ의 핵심 설계 원칙:

• 공유 worker pool: 모든 workqueue가 worker pool을 공유하여 커널 스레드 수 최소화
• 자동 동시성 관리: worker가 블록되면 새 worker를 자동 생성하여 CPU 유휴 방지
• Bound/Unbound 분리: CPU-bound 작업과 NUMA-aware unbound 작업 구분
• 속성 기반 매핑: workqueue 플래그에 따라 적절한 worker pool에 자동 매핑

/*
 * CMWQ 계층 구조:
 *
 *   workqueue_struct          ← 사용자가 생성/사용하는 인터페이스
 *       │
 *       ├── pool_workqueue    ← workqueue와 worker_pool의 연결 (Per-CPU 또는 Per-NUMA)
 *       │       │
 *       │       └── worker_pool  ← kworker 스레드들의 풀 (공유 자원)
 *       │               │
 *       │               ├── worker (kworker/0:0)
 *       │               ├── worker (kworker/0:1)
 *       │               └── worker (kworker/0:2H)   ← highpri
 *       │
 *       └── pool_workqueue    ← 다른 CPU/NUMA 노드
 *               └── worker_pool
 *                       └── ...
 */

Worker Pool

/*
 * Worker Pool 유형:
 *
 * 1. Bound (Per-CPU) Pool:
 *    - 각 CPU에 2개: normal (nice=0) + highpri (nice=-20)
 *    - kworker/CPU:ID 또는 kworker/CPU:IDH (highpri)
 *    - 해당 CPU에서만 work 실행 → 캐시 친화적
 *
 * 2. Unbound Pool:
 *    - NUMA 노드별 생성, 속성(nice, cpumask)으로 관리
 *    - kworker/uPOOL:ID
 *    - 어떤 CPU에서든 실행 가능 → 부하 분산
 *    - long-running 또는 CPU-intensive 작업에 적합
 *
 * 동시성 관리:
 *    - 풀의 running worker가 모두 블록되면 새 worker 생성
 *    - idle worker는 일정 시간 후 소멸 (IDLE_WORKER_TIMEOUT: 300초)
 *    - manager worker가 pool을 관리 (worker 생성/소멸)
 */

/* kernel/workqueue.c 주요 구조체 (간략화) */
struct worker_pool {
    spinlock_t lock;
    int cpu;                      /* bound pool의 CPU, unbound는 -1 */
    int node;                     /* NUMA 노드 */
    int id;
    unsigned int flags;
    struct list_head worklist;    /* pending work items */
    int nr_workers;               /* 총 worker 수 */
    int nr_idle;                  /* idle worker 수 */
    int nr_running;               /* 실행 중인 worker 수 (atomic) */
    struct list_head idle_list;   /* idle worker 리스트 */
    struct timer_list idle_timer;
    struct timer_list mayday_timer;
};

alloc_workqueue() API

/* workqueue 생성 */
struct workqueue_struct *alloc_workqueue(
    const char *fmt,          /* 이름 형식 (printf 스타일) */
    unsigned int flags,       /* WQ_* 플래그 조합 */
    int max_active,           /* Per-CPU 최대 동시 실행 work 수 */
    ...                       /* fmt 인자 */
);

/* 예제 */
struct workqueue_struct *my_wq;
my_wq = alloc_workqueue("my_driver_wq",
    WQ_UNBOUND | WQ_MEM_RECLAIM, 0);

max_active 동시성 제어

/*
 * max_active: Per-CPU 또는 Per-NUMA 동시 실행 work item 수 제한
 *
 * - 0: 기본값 (WQ_DFL_ACTIVE = 256)
 * - 1: 순차 실행 (alloc_ordered_workqueue)
 * - N: 최대 N개 동시 실행
 *
 * Bound (Per-CPU) workqueue:
 *   max_active=4 → 각 CPU에서 최대 4개 work 동시 실행
 *
 * Unbound workqueue:
 *   max_active=4 → 각 NUMA 노드에서 최대 4개 work 동시 실행
 *
 * 주의: max_active는 실행 중인 work만 제한
 *       pending(대기 중) work 수는 무제한
 */

/* 순차 실행이 필요한 경우 */
struct workqueue_struct *ordered_wq;
ordered_wq = alloc_ordered_workqueue("my_ordered", 0);
/* = alloc_workqueue("my_ordered", __WQ_ORDERED, 1) */

시스템 Workqueue

커널은 미리 생성된 시스템 workqueue를 제공합니다. 대부분의 경우 전용 workqueue를 만들 필요 없이 시스템 workqueue를 사용합니다:

Work Item 생명주기

/*
 * Work Item 상태 전이:
 *
 * [Idle]                        work이 어떤 workqueue에도 없는 상태
 *   │
 *   │ queue_work() / schedule_work()
 *   ▼
 * [Pending]                     worklist에 대기 중
 *   WORK_STRUCT_PENDING 비트 set
 *   │
 *   │ kworker가 dequeue
 *   ▼
 * [Running]                     worker가 콜백 실행 중
 *   PENDING 클리어, current_work = this
 *   │
 *   │ 콜백 완료
 *   ▼
 * [Idle]                        다시 큐잉 가능
 *
 * 핵심 규칙:
 * - PENDING인 work를 다시 queue하면 no-op (중복 방지)
 * - Running 중에 queue하면 PENDING이 set되어 완료 후 재실행
 * - 서로 다른 workqueue에 같은 work를 queue할 수 없음
 */

/* work_struct 내부 */
struct work_struct {
    atomic_long_t data;        /* flags + pool_workqueue 포인터 */
    struct list_head entry;     /* worklist 연결 */
    work_func_t func;           /* 콜백 함수 */
};

queue_work() vs schedule_work()

/* schedule_work(): system_wq에 큐잉 (편의 함수) */
static inline bool schedule_work(struct work_struct *work)
{
    return queue_work(system_wq, work);
}

/* queue_work(): 특정 workqueue에 큐잉 */
bool queue_work(struct workqueue_struct *wq,
               struct work_struct *work);

/* queue_work_on(): 특정 CPU에 큐잉 */
bool queue_work_on(int cpu, struct workqueue_struct *wq,
                   struct work_struct *work);

/* 반환값: true = 새로 큐잉됨, false = 이미 pending */

/* 전용 workqueue 사용 vs system_wq 기준:
 *
 * system_wq 사용 (schedule_work):
 *   - 짧은 작업, 다른 work와 간섭 적음
 *   - 대부분의 드라이버에서 적합
 *
 * 전용 workqueue 생성:
 *   - flush/cancel 시 다른 서브시스템에 영향 없어야 할 때
 *   - 특수 플래그 필요 (WQ_UNBOUND, WQ_MEM_RECLAIM 등)
 *   - max_active 제어 필요
 *   - /sys/kernel/debug/workqueue에서 독립 모니터링 필요
 */

Ordered Workqueue

/* Ordered Workqueue: 큐잉 순서대로 하나씩 실행 */
struct workqueue_struct *owq;
owq = alloc_ordered_workqueue("my_ordered", 0);

/*
 * 특성:
 * - max_active = 1 → 동시에 하나의 work만 실행
 * - 큐잉 순서 보장 (FIFO)
 * - 내부적으로 __WQ_ORDERED 플래그 + unbound
 *
 * 사용 시나리오:
 * - 상태 머신 이벤트 처리 (순서 중요)
 * - 파일시스템 로그/저널 쓰기
 * - 하드웨어 초기화 시퀀스
 *
 * 주의: ordered wq는 WQ_UNBOUND를 암시적으로 포함
 *       → CPU 마이그레이션 가능 (특정 CPU 고정 아님)
 */

Delayed Work

/* Delayed Work: 지정 시간 후 실행 */
struct delayed_work {
    struct work_struct work;
    struct timer_list timer;
    struct workqueue_struct *wq;
    int cpu;
};

/* 초기화 */
INIT_DELAYED_WORK(&dev->dwork, my_delayed_handler);

/* 큐잉: delay jiffies 후 실행 */
queue_delayed_work(my_wq, &dev->dwork,
    msecs_to_jiffies(100));    /* 100ms 후 */

/* 시스템 workqueue에 큐잉 */
schedule_delayed_work(&dev->dwork,
    msecs_to_jiffies(500));    /* 500ms 후 */

/* mod_delayed_work(): 이미 pending인 delayed work의 타이머 변경 */
mod_delayed_work(my_wq, &dev->dwork,
    msecs_to_jiffies(200));    /* 기존 타이머 취소 + 200ms로 재설정 */
/* 반환값: true = 기존 pending work를 변경, false = 새로 큐잉 */

/* 즉시 실행으로 변경 */
mod_delayed_work(my_wq, &dev->dwork, 0);
/* delay=0이면 가능한 빨리 실행 */

취소 패턴

/* Work 취소: 동기적으로 완료 대기 */
bool cancel_work_sync(struct work_struct *work);
/*
 * - pending이면: dequeue 후 반환 (true)
 * - running이면: 완료를 기다린 후 반환 (false)
 * - idle이면: 즉시 반환 (false)
 * 주의: 슬립 가능! 인터럽트/atomic 컨텍스트에서 호출 불가
 */

/* Delayed Work 취소 */
bool cancel_delayed_work(struct delayed_work *dwork);
/* 비동기: 타이머만 취소, 이미 실행 중이면 대기 안 함 */

bool cancel_delayed_work_sync(struct delayed_work *dwork);
/* 동기: 타이머 취소 + 실행 중인 콜백 완료 대기 */

/* 안전한 드라이버 해제 패턴 */
static void my_remove(struct platform_device *pdev)
{
    struct my_device *dev = platform_get_drvdata(pdev);

    /* 1. 더 이상 새 work가 큐잉되지 않도록 플래그 설정 */
    dev->shutting_down = true;

    /* 2. 모든 work 취소 (실행 중이면 완료 대기) */
    cancel_work_sync(&dev->work);
    cancel_delayed_work_sync(&dev->dwork);

    /* 3. 이 시점에서 work 콜백이 실행되지 않음을 보장 */
}

Flush 패턴

/* flush_work(): 특정 work의 완료 대기 */
bool flush_work(struct work_struct *work);
/* pending/running work 완료를 기다림 */

/* flush_workqueue(): workqueue의 모든 pending work 완료 대기 */
void flush_workqueue(struct workqueue_struct *wq);
/* 호출 시점에 pending인 모든 work의 완료를 기다림 */
/* flush 후에 새로 큐잉된 work는 포함하지 않음 */

/* flush_scheduled_work(): system_wq flush */
void flush_scheduled_work(void);
/* 모듈 해제 시 system_wq에 큐잉된 work 정리용 */

Workqueue 디버깅

# workqueue 상태 확인
cat /sys/kernel/debug/workqueue

# 출력 예시:
# workqueue           CPU  POOL  ACTIVE/MAX  WORKERS  FLAGS
# events               0     0      0/256      3
# events               1     2      0/256      2
# events_highpri       0     1      0/256      2     highpri
# my_driver_wq        -1    16      2/4        3     unbound

# kworker 스레드 확인
ps aux | grep kworker
# kworker/0:0   - CPU 0 bound worker
# kworker/0:0H  - CPU 0 highpri bound worker
# kworker/u8:0  - unbound worker (pool id=8)

# WQ_SYSFS가 설정된 workqueue의 런타임 설정
ls /sys/devices/virtual/workqueue/
# cpumask  max_active  nice

# wq_watchdog: 정체된 work 탐지
# CONFIG_WQ_WATCHDOG=y + wq_watchdog_thresh_ms (기본 30초)
echo 10000 > /sys/module/workqueue/parameters/watchdog_thresh
# 10초 이상 실행 중인 work 경고

# workqueue tracepoint
echo 1 > /sys/kernel/debug/tracing/events/workqueue/workqueue_queue_work/enable
echo 1 > /sys/kernel/debug/tracing/events/workqueue/workqueue_execute_start/enable
echo 1 > /sys/kernel/debug/tracing/events/workqueue/workqueue_execute_end/enable
cat /sys/kernel/debug/tracing/trace

Best Practices

/*
 * Workqueue Best Practices
 *
 * 1. 대부분의 경우 system_wq 사용 (schedule_work)
 *    → 전용 workqueue는 정말 필요할 때만 생성
 *
 * 2. 메모리 회수 경로에서 사용되면 WQ_MEM_RECLAIM 필수
 *    → 파일시스템, 블록 I/O, 스왑 관련 work
 *
 * 3. long-running 작업은 WQ_UNBOUND 사용
 *    → bound pool의 동시성 관리에 간섭 방지
 *
 * 4. CPU-intensive 작업은 WQ_CPU_INTENSIVE 사용
 *    → 불필요한 worker 생성 방지
 *
 * 5. 드라이버 해제 시 반드시 cancel_*_sync() 호출
 *    → use-after-free 방지
 *
 * 6. work 콜백에서 자신의 workqueue flush 금지
 *    → 데드락
 *
 * 7. IRQ 핸들러에서 work 큐잉 가능
 *    → queue_work()는 IRQ-safe
 *    → 하지만 cancel_work_sync()는 불가 (슬립)
 *
 * 8. 같은 work_struct를 여러 workqueue에 큐잉하지 말 것
 *    → 하나의 work는 하나의 workqueue에만 속할 수 있음
 */

Bottom Half 선택 가이드

결정 매트릭스

결정 흐름도

PREEMPT_RT 영향

/*
 * PREEMPT_RT에서의 Bottom Half 변화:
 *
 * 1. Softirq:
 *    - 모든 softirq가 ksoftirqd에서 실행 (선점 가능)
 *    - irq_exit()에서 직접 실행하지 않음
 *    - local_bh_disable()이 preempt_disable()로 변경되지 않음
 *      → RT 뮤텍스 기반으로 변경
 *
 * 2. Tasklet:
 *    - PREEMPT_RT에서 문제 유발
 *    - 인터럽트 컨텍스트 가정 코드가 호환되지 않음
 *    - 커널 커뮤니티에서 제거 진행 중
 *
 * 3. Workqueue:
 *    - 정상 동작 (이미 프로세스 컨텍스트)
 *    - RT 우선순위 설정 가능
 *
 * 4. Threaded IRQ:
 *    - 정상 동작 (이미 스레드 기반)
 *    - SCHED_FIFO 우선순위로 실행
 *    - chrt 명령으로 IRQ 스레드 우선순위 조정 가능
 *
 * 5. Spinlock:
 *    - spin_lock()이 rt_mutex로 변경 (슬립 가능!)
 *    - raw_spin_lock()만 진짜 스핀 (사용 최소화)
 *    - spin_lock_irqsave() → sleeping lock + local_irq_save
 */

성능 특성 비교

관련 커널 설정