Bottom Half 선택 가이드와 실전 패턴

Bottom Half 메커니즘(softirq, tasklet, workqueue, threaded IRQ) 중 올바른 선택 기준, 실전 패턴, 성능 최적화, 디버깅 가이드를 제공합니다.

이 페이지는 "슬립 필요 여부, 처리량, 순서 보장, 격리 수준, PREEMPT_RT 호환성" 기준으로 메커니즘을 고르는 실전 의사결정 표준을 제시합니다. 각 메커니즘의 심화 내용은 아래 전용 페이지를 참고하세요.

메커니즘별 심화 문서

각 Bottom Half 메커니즘의 내부 구현, API 상세, 디버깅 기법은 전용 페이지에서 다룹니다.

아래부터는 메커니즘 선택 기준과 공통 패턴을 다룹니다.

Bottom Half 선택 가이드

결정 매트릭스

결정 흐름도

고급 선택 결정 트리

위 흐름도보다 더 세부적인 결정 기준을 포함한 고급 결정 트리입니다. 지연 허용 범위, 순서 보장, 메모리 할당 모드, CPU 바인딩 여부까지 고려합니다.

PREEMPT_RT 영향

성능 특성 비교

실행 컨텍스트 비교 다이어그램

각 Bottom Half 메커니즘이 실행되는 컨텍스트를 시각적으로 비교합니다. 인터럽트 컨텍스트(softirq/tasklet)와 프로세스 컨텍스트(workqueue/threaded IRQ)의 차이를 이해하는 것이 올바른 선택의 핵심입니다.

지연시간/처리량 시각화

각 메커니즘의 호출 오버헤드와 처리 지연시간을 시각적 막대 그래프로 비교합니다. 실제 측정값은 하드웨어와 커널 설정에 따라 다르지만, 상대적인 크기를 이해하는 데 유용합니다.

관련 커널 설정

PREEMPT_RT에서의 Bottom Half 변환 상세

PREEMPT_RT(Real-Time) 패치가 적용되면 각 Bottom Half 메커니즘의 동작이 크게 변합니다. 아래 다이어그램은 일반 커널과 RT 커널에서 각 메커니즘의 실행 경로 변환을 보여줍니다.

PREEMPT_RT 마이그레이션 코드 패턴

tasklet에서 threaded IRQ로의 전환 과정을 단계별로 보여줍니다.

/* ===== 변환 전: tasklet 기반 드라이버 ===== */
struct my_device {
    struct tasklet_struct rx_tasklet;
    spinlock_t lock;
    void __iomem *regs;
};

static irqreturn_t my_hardirq(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    u32 status = readl(dev->regs + IRQ_STATUS);

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

    writel(status, dev->regs + IRQ_ACK);
    tasklet_schedule(&dev->rx_tasklet);  /* PREEMPT_RT 비호환! */
    return IRQ_HANDLED;
}

static void my_tasklet_func(struct tasklet_struct *t)
{
    struct my_device *dev = from_tasklet(dev, t, rx_tasklet);
    unsigned long flags;

    spin_lock_irqsave(&dev->lock, flags);
    process_rx_data(dev);
    spin_unlock_irqrestore(&dev->lock, flags);
}

/* ===== 변환 후: threaded IRQ 기반 ===== */
struct my_device {
    /* tasklet 제거 */
    spinlock_t lock;
    void __iomem *regs;
};

static irqreturn_t my_hardirq(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;
    u32 status = readl(dev->regs + IRQ_STATUS);

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

    writel(status, dev->regs + IRQ_ACK);
    return IRQ_WAKE_THREAD;  /* threaded handler 호출 */
}

static irqreturn_t my_thread_fn(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;

    spin_lock(&dev->lock);  /* RT에서 rt_mutex로 안전하게 변환됨 */
    process_rx_data(dev);
    spin_unlock(&dev->lock);

    return IRQ_HANDLED;
}

/* probe에서 등록 */
request_threaded_irq(irq, my_hardirq, my_thread_fn,
                    IRQF_ONESHOT, "my_device", dev);

일반적인 실수와 올바른 패턴

Bottom Half 사용 시 자주 발생하는 실수와 올바른 접근 방법을 비교합니다.

❌ 실수 1: softirq/tasklet에서 슬립 시도

/* 잘못된 예: softirq에서 블로킹 함수 호출 */
static void my_softirq_action(struct softirq_action *h)
{
    struct data *d;

    mutex_lock(&my_mutex);  /* ❌ softirq는 atomic context! */
    d = process_data();
    mutex_unlock(&my_mutex);
}

/* 잘못된 예: tasklet에서 msleep */
static void my_tasklet_func(struct tasklet_struct *t)
{
    msleep(100);  /* ❌ atomic context에서 슬립 불가 */
    process_data();
}

/* 올바른 예: workqueue 사용 */
static void my_work_func(struct work_struct *work)
{
    struct data *d;

    mutex_lock(&my_mutex);  /* ✓ workqueue는 프로세스 컨텍스트 */
    d = process_data();
    mutex_unlock(&my_mutex);

    msleep(100);  /* ✓ 슬립 가능 */
}

❌ 실수 2: tasklet 재진입 가정

/* 잘못된 예: tasklet이 동시 실행될 것으로 가정 */
static atomic_t counter = ATOMIC_INIT(0);

static void my_tasklet(struct tasklet_struct *t)
{
    /* ❌ 불필요한 atomic 연산 - 같은 tasklet은 직렬화됨 */
    atomic_inc(&counter);
}

/* 올바른 예: tasklet 직렬화 보장 활용 */
struct my_driver_data {
    struct tasklet_struct tasklet;
    int counter;  /* ✓ atomic 불필요 */
};

static void my_tasklet(struct tasklet_struct *t)
{
    struct my_driver_data *data = from_tasklet(data, t, tasklet);
    data->counter++;  /* ✓ 같은 tasklet은 직렬화되므로 안전 */
}

❌ 실수 3: workqueue를 atomic context로 가정

/* 잘못된 예: workqueue에서 spinlock_irqsave 남용 */
static void my_work(struct work_struct *work)
{
    unsigned long flags;

    spin_lock_irqsave(&my_lock, flags);  /* ❌ 불필요 - 이미 인터럽트 활성화 */
    process_data();
    spin_unlock_irqrestore(&my_lock, flags);
}

/* 올바른 예: 적절한 락 사용 */
static void my_work(struct work_struct *work)
{
    spin_lock(&my_lock);  /* ✓ workqueue는 인터럽트 활성화 상태 */
    process_data();
    spin_unlock(&my_lock);

    /* 또는 mutex 사용 가능 */
    mutex_lock(&my_mutex);  /* ✓ 슬립 가능 */
    slow_operation();
    mutex_unlock(&my_mutex);
}

❌ 실수 4: raise_softirq() 사용 시 인터럽트 상태 오인

/* 잘못된 예: 인터럽트 활성화 상태에서 raise_softirq_irqoff 호출 */
void my_function(void)
{
    local_irq_disable();
    do_something();
    local_irq_enable();

    raise_softirq_irqoff(NET_RX_SOFTIRQ);  /* ❌ IRQ 활성화됨! */
}

/* 올바른 예 1: 인터럽트 비활성화 상태 확인 */
void my_function(void)
{
    unsigned long flags;

    local_irq_save(flags);
    do_something();
    raise_softirq_irqoff(NET_RX_SOFTIRQ);  /* ✓ IRQ 비활성화 상태 */
    local_irq_restore(flags);
}

/* 올바른 예 2: 안전한 raise_softirq 사용 */
void my_function(void)
{
    do_something();
    raise_softirq(NET_RX_SOFTIRQ);  /* ✓ 내부에서 IRQ 제어 */
}

❌ 실수 5: workqueue flush 시 데드락

/* 잘못된 예: work 내부에서 자신을 flush */
static void my_work(struct work_struct *work)
{
    do_something();
    flush_work(work);  /* ❌ 데드락! 자기 자신을 기다림 */
}

/* 잘못된 예: 같은 workqueue에서 flush_workqueue */
static void my_work(struct work_struct *work)
{
    flush_workqueue(my_wq);  /* ❌ 같은 wq에서 실행 중 */
}

/* 올바른 예: 별도 컨텍스트에서 flush */
void cleanup_driver(void)
{
    cancel_work_sync(&my_work);  /* ✓ 외부에서 취소 대기 */
    flush_workqueue(my_wq);       /* ✓ 모든 work 완료 대기 */
}

✅ 모범 사례 체크리스트

성능 최적화 가이드

Bottom Half는 시스템 전체 성능에 큰 영향을 미칩니다. 효율적인 사용 방법을 소개합니다.

softirq 실행 시간 제한

/* __do_softirq() 내부 - 최대 2ms 또는 10번 재시작 제한 */
unsigned long end = jiffies + MAX_SOFTIRQ_TIME;  /* 2ms */
int max_restart = MAX_SOFTIRQ_RESTART;  /* 10 */

/* 성능 최적화: softirq 핸들러는 2ms 이내에 완료해야 함 */
static void my_softirq_action(struct softirq_action *h)
{
    struct sk_buff *skb;
    int budget = 64;  /* 한 번에 처리할 패킷 수 제한 */

    while ((skb = dequeue_packet()) && --budget > 0) {
        process_packet(skb);
    }

    /* 남은 작업이 있으면 다시 스케줄 */
    if (has_pending_packets())
        raise_softirq(NET_RX_SOFTIRQ);
}

tasklet 병합으로 오버헤드 감소

/* ❌ 비효율적: 매 인터럽트마다 tasklet 스케줄 */
irqreturn_t my_irq_handler(int irq, void *dev_id)
{
    tasklet_schedule(&my_tasklet);  /* 매번 스케줄 */
    return IRQ_HANDLED;
}

/* ✅ 효율적: pending 상태면 스케줄 생략 */
irqreturn_t my_irq_handler(int irq, void *dev_id)
{
    struct my_device *dev = dev_id;

    atomic_inc(&dev->pending_count);

    /* 이미 스케줄되어 있으면 중복 스케줄 안 함 */
    if (!test_and_set_bit(TASKLET_STATE_SCHED, &my_tasklet.state))
        tasklet_schedule(&my_tasklet);

    return IRQ_HANDLED;
}

static void my_tasklet_func(struct tasklet_struct *t)
{
    struct my_device *dev = from_tasklet(dev, t, tasklet);
    int count = atomic_xchg(&dev->pending_count, 0);

    /* 여러 인터럽트를 한 번에 처리 */
    process_batch(dev, count);
}

workqueue 동시성 튜닝

/* ❌ 기본 workqueue - 제한된 동시성 */
schedule_work(&my_work);  /* system_wq 사용 */

/* ✅ 커스텀 workqueue - 동시성 제어 */
struct workqueue_struct *my_wq;

/* CPU-bound 작업 - UNBOUND로 CPU 간 이동 허용 */
my_wq = alloc_workqueue("my_wq",
                        WQ_UNBOUND | WQ_HIGHPRI,
                        4);  /* max_active = 4 */

/* I/O-bound 작업 - 높은 동시성 허용 */
my_wq = alloc_workqueue("io_wq",
                        WQ_MEM_RECLAIM,
                        256);  /* 많은 동시 실행 허용 */

/* 성능 측정 */
u64 start = ktime_get_ns();
flush_workqueue(my_wq);
u64 elapsed = ktime_get_ns() - start;
pr_info("Workqueue flush took %llu ns\n", elapsed);

NAPI와의 통합 최적화

/* 네트워크 드라이버 최적화 패턴 */
static int my_napi_poll(struct napi_struct *napi, int budget)
{
    struct my_device *dev = container_of(napi, struct my_device, napi);
    int work_done = 0;

    /* budget만큼만 처리 (softirq 시간 제한 준수) */
    while (work_done < budget) {
        struct sk_buff *skb = receive_packet(dev);
        if (!skb)
            break;

        netif_receive_skb(skb);
        work_done++;
    }

    /* 모든 패킷 처리했으면 NAPI 종료 */
    if (work_done < budget) {
        napi_complete(napi);
        enable_interrupts(dev);
    }

    return work_done;
}

성능 측정 도구

# softirq 통계 확인
cat /proc/softirqs
#                    CPU0       CPU1       CPU2       CPU3
#          HI:          0          0          0          0
#       TIMER:    5123456    4987654    5234567    5123456
#      NET_TX:      12345       9876      11234      10123
#      NET_RX:    9876543    8765432    9123456    8987654

# workqueue 통계 (debugfs)
cat /sys/kernel/debug/workqueue/workqueues
cat /sys/kernel/debug/workqueue/pool_workqueues

# ftrace로 softirq 지연 측정
echo 1 > /sys/kernel/tracing/events/irq/softirq_entry/enable
echo 1 > /sys/kernel/tracing/events/irq/softirq_exit/enable
cat /sys/kernel/tracing/trace

# perf로 softirq 시간 분석
perf record -e irq:softirq_entry,irq:softirq_exit -a sleep 10
perf script

실전 케이스 스터디

실제 드라이버와 서브시스템에서 Bottom Half를 활용하는 패턴을 분석합니다.

케이스 1: 네트워크 드라이버 (NAPI + softirq)

시나리오: 고성능 네트워크 카드의 RX 처리

/* drivers/net/ethernet/intel/e1000e/netdev.c 패턴 */

/* 1. 인터럽트 핸들러 (Top Half) */
static irqreturn_t e1000_intr(int irq, void *data)
{
    struct net_device *netdev = data;
    struct e1000_adapter *adapter = netdev_priv(netdev);
    u32 icr = er32(ICR);

    if (!icr)
        return IRQ_NONE;  /* 우리 인터럽트 아님 */

    if (icr & E1000_ICR_RXT0) {  /* RX 인터럽트 */
        /* 인터럽트 비활성화 */
        ew32(IMC, ~0);

        /* NAPI 스케줄 (softirq로 위임) */
        if (napi_schedule_prep(&adapter->napi)) {
            __napi_schedule(&adapter->napi);
        }
    }

    return IRQ_HANDLED;
}

/* 2. NAPI poll (NET_RX_SOFTIRQ에서 실행) */
static int e1000_clean(struct napi_struct *napi, int budget)
{
    struct e1000_adapter *adapter =
        container_of(napi, struct e1000_adapter, napi);
    int work_done = 0;

    /* budget만큼만 처리 */
    e1000_clean_rx_irq(adapter, &work_done, budget);

    /* 모두 처리했으면 NAPI 완료 */
    if (work_done < budget) {
        napi_complete_done(napi, work_done);

        /* 인터럽트 재활성화 */
        e1000_irq_enable(adapter);
    }

    return work_done;
}

/* 3. 실제 패킷 처리 */
static bool e1000_clean_rx_irq(struct e1000_adapter *adapter,
                             int *work_done, int work_to_do)
{
    while (*work_done < work_to_do) {
        struct sk_buff *skb = e1000_receive_skb(adapter);
        if (!skb)
            break;

        /* 네트워크 스택으로 전달 */
        napi_gro_receive(&adapter->napi, skb);
        (*work_done)++;
    }

    return false;
}

핵심 포인트:

• Top Half: 최소한의 작업 (인터럽트 비활성화, NAPI 스케줄)
• Bottom Half: softirq(NET_RX)에서 실제 패킷 처리
• Budget 제한으로 softirq 시간 제한 준수

케이스 2: 블록 드라이버 (workqueue)

시나리오: NVMe 드라이버의 I/O 완료 처리

/* drivers/nvme/host/pci.c 패턴 */

/* 1. 인터럽트 핸들러 */
static irqreturn_t nvme_irq(int irq, void *data)
{
    struct nvme_queue *nvmeq = data;

    /* CQ에서 완료 엔트리 확인 */
    if (nvme_cqe_pending(nvmeq)) {
        /* workqueue로 완료 처리 위임 */
        queue_work(nvmeq->cq_wq, &nvmeq->cq_work);
    }

    return IRQ_HANDLED;
}

/* 2. Workqueue 핸들러 */
static void nvme_cq_work(struct work_struct *work)
{
    struct nvme_queue *nvmeq =
        container_of(work, struct nvme_queue, cq_work);

    /* 완료 큐 처리 (블로킹 가능) */
    nvme_process_cq(nvmeq);

    /* 블록 레이어에 완료 통지 */
    blk_mq_complete_request(...);
}

케이스 3: 타이머 + tasklet 조합

시나리오: 주기적 하드웨어 폴링

/* 실전 패턴: 폴링 기반 디바이스 */
struct my_device {
    struct timer_list poll_timer;
    struct tasklet_struct poll_tasklet;
    void __iomem *regs;
};

/* 1. 타이머 콜백 (TIMER_SOFTIRQ) */
static void poll_timer_func(struct timer_list *t)
{
    struct my_device *dev = from_timer(dev, t, poll_timer);

    /* 빠른 레지스터 읽기만 수행 */
    u32 status = readl(dev->regs + STATUS_REG);

    if (status & DATA_READY) {
        /* 실제 처리는 tasklet으로 위임 */
        tasklet_schedule(&dev->poll_tasklet);
    }

    /* 다음 폴링 예약 (100ms) */
    mod_timer(&dev->poll_timer, jiffies + HZ/10);
}

/* 2. Tasklet 핸들러 */
static void poll_tasklet_func(struct tasklet_struct *t)
{
    struct my_device *dev = from_tasklet(dev, t, poll_tasklet);

    /* 데이터 처리 (약간 더 복잡한 작업) */
    process_device_data(dev);
}

케이스 4: 지연된 리소스 정리

시나리오: RCU + workqueue로 안전한 메모리 해제

/* 실전 패턴: RCU와 workqueue 조합 */
struct my_object {
    struct rcu_head rcu;
    struct work_struct cleanup_work;
    void *large_buffer;
};

/* 1. 객체 삭제 요청 */
void delete_object(struct my_object *obj)
{
    /* RCU로 readers 보호 */
    call_rcu(&obj->rcu, object_rcu_callback);
}

/* 2. RCU 콜백 (softirq) */
static void object_rcu_callback(struct rcu_head *rcu)
{
    struct my_object *obj = container_of(rcu, struct my_object, rcu);

    /* 무거운 정리 작업은 workqueue로 위임 */
    INIT_WORK(&obj->cleanup_work, cleanup_work_func);
    schedule_work(&obj->cleanup_work);
}

/* 3. Workqueue에서 실제 정리 */
static void cleanup_work_func(struct work_struct *work)
{
    struct my_object *obj =
        container_of(work, struct my_object, cleanup_work);

    /* 슬립 가능한 정리 작업 */
    vfree(obj->large_buffer);  /* 시간이 걸릴 수 있음 */
    kfree(obj);
}

실전 네트워크 드라이버 BH 처리 경로

네트워크 드라이버에서 패킷이 수신되어 프로토콜 스택까지 전달되는 과정의 전체 Bottom Half 경로를 시각화합니다. NIC 인터럽트부터 소켓 버퍼 전달까지의 흐름입니다.

네트워크 드라이버 BH 핵심 코드 분석

/* net/core/dev.c - NET_RX softirq 핸들러 */
static void net_rx_action(struct softirq_action *h)
{
    struct softnet_data *sd = this_cpu_ptr(&softnet_data);
    unsigned long time_limit = jiffies +
        usecs_to_jiffies(READ_ONCE(netdev_budget_usecs));
    int budget = READ_ONCE(netdev_budget);  /* 기본값 300 */
    LIST_HEAD(list);
    LIST_HEAD(repoll);

    local_irq_disable();
    list_splice_init(&sd->poll_list, &list);
    local_irq_enable();

    for (;;) {
        struct napi_struct *n;

        skb_defer_free_flush(sd);

        if (list_empty(&list)) {
            if (!sd_has_rps_ipi_waiting(sd) &&
                list_empty(&repoll))
                return;
            break;
        }

        n = list_first_entry(&list, struct napi_struct, poll_list);
        budget -= napi_poll(n, &repoll);

        /* 시간 초과 또는 budget 소진 시 중단 */
        if (unlikely(budget <= 0 ||
                     time_after_eq(jiffies, time_limit))) {
            sd->time_squeeze++;
            break;
        }
    }
    /* 남은 NAPI가 있으면 softirq 재스케줄 */
    /* -> __raise_softirq_irqoff(NET_RX_SOFTIRQ) */
}

디버깅 도구와 기법

Bottom Half 관련 문제를 진단하기 위한 도구와 기법을 체계적으로 정리합니다.

/proc/softirqs 분석

# softirq 타입별, CPU별 누적 카운트 확인
cat /proc/softirqs
#                    CPU0       CPU1       CPU2       CPU3
#          HI:          2          0          0          1
#       TIMER:    5123456    4987654    5234567    5123456
#      NET_TX:      12345       9876      11234      10123
#      NET_RX:    9876543    8765432    9123456    8987654
#       BLOCK:     234567     198765     212345     207654
# IRQ_POLL:          0          0          0          0
#     TASKLET:       4567       3456       4321       3987
#       SCHED:    2345678    2234567    2456789    2345678
#     HRTIMER:      12345      11234      13456      12345
#         RCU:    3456789    3345678    3567890    3456789

# 실시간 모니터링 (1초 간격으로 변화량 관찰)
watch -d -n 1 'cat /proc/softirqs'

# CPU 간 불균형 확인 (NET_RX가 한 CPU에 몰리면 RSS 설정 필요)
# softnet_stat으로 추가 통계 확인
cat /proc/net/softnet_stat
# column 1: processed  column 2: dropped  column 3: time_squeeze

ftrace를 이용한 softirq/workqueue 추적

# ftrace 설정: softirq 이벤트 추적
cd /sys/kernel/tracing

# softirq 진입/종료 이벤트 활성화
echo 1 > events/irq/softirq_entry/enable
echo 1 > events/irq/softirq_exit/enable
echo 1 > events/irq/softirq_raise/enable

# workqueue 이벤트 추적
echo 1 > events/workqueue/workqueue_queue_work/enable
echo 1 > events/workqueue/workqueue_execute_start/enable
echo 1 > events/workqueue/workqueue_execute_end/enable

# 추적 시작
echo 1 > tracing_on
sleep 5
echo 0 > tracing_on

# 결과 확인
cat trace | head -50
# <idle>-0  [001] ..s1  1234.567890: softirq_entry: vec=3 [action=NET_RX]
# <idle>-0  [001] ..s1  1234.567925: softirq_exit:  vec=3 [action=NET_RX]
# kworker/1:0-123 [001] ....  1234.568000: workqueue_execute_start: work=...

# softirq 실행 시간 히스토그램 (function_graph 트레이서)
echo function_graph > current_tracer
echo do_softirq > set_graph_function
echo 1 > tracing_on
sleep 2
echo 0 > tracing_on
cat trace

perf를 이용한 성능 분석

# softirq 핫스팟 분석
perf record -e irq:softirq_entry -e irq:softirq_exit -a -g sleep 10
perf report

# softirq 실행 시간 분포
perf script | awk '/softirq_entry/{start=$4} /softirq_exit/{print $4-start}'

# workqueue 실행 빈도
perf stat -e workqueue:workqueue_execute_start -a sleep 10

# 스케줄링 지연 측정 (wakeup latency)
perf sched record sleep 5
perf sched latency --sort max

# IRQ 비활성화 시간 측정 (irqsoff tracer)
echo irqsoff > /sys/kernel/tracing/current_tracer
echo 1 > /sys/kernel/tracing/tracing_on
sleep 5
echo 0 > /sys/kernel/tracing/tracing_on
cat /sys/kernel/tracing/trace

디버깅용 커널 설정

데드락과 우선순위 역전 패턴

Bottom Half 사용 시 발생하기 쉬운 데드락 및 우선순위 역전 패턴과 방지 방법입니다.

패턴 1: softirq와 프로세스 간 spinlock 데드락

/* ❌ 데드락 시나리오:
 * CPU 0에서 process_context()가 spin_lock(&lock) 획득
 * -> 인터럽트 발생 -> softirq에서 spin_lock(&lock) 시도
 * -> 같은 CPU에서 락 보유자를 선점할 수 없어 영원히 대기 */

spinlock_t my_lock;

/* 프로세스 컨텍스트 */
void process_context(void)
{
    spin_lock(&my_lock);     /* ❌ BH 비활성화 없이 락 획득 */
    do_something();
    spin_unlock(&my_lock);
}

/* softirq 컨텍스트 */
void my_softirq(struct softirq_action *h)
{
    spin_lock(&my_lock);     /* 데드락! */
    process_data();
    spin_unlock(&my_lock);
}

/* ✅ 올바른 패턴: spin_lock_bh() 사용 */
void process_context(void)
{
    spin_lock_bh(&my_lock);  /* BH 비활성화 + 락 획득 */
    do_something();
    spin_unlock_bh(&my_lock);
}

void my_softirq(struct softirq_action *h)
{
    spin_lock(&my_lock);     /* BH에서는 spin_lock만으로 충분 */
    process_data();
    spin_unlock(&my_lock);
}

패턴 2: workqueue 중첩 flush 데드락

/* ❌ 데드락 시나리오:
 * work_A가 system_wq에서 실행 중 flush_work(&work_B) 호출
 * work_B도 system_wq에 대기 중이지만 max_active=1이면
 * work_A가 완료될 때까지 work_B 실행 불가 -> 순환 대기 */

static void work_a_func(struct work_struct *work)
{
    do_part1();
    flush_work(&work_b);  /* ❌ 같은 WQ에서 다른 work 대기 */
    do_part2();
}

/* ✅ 올바른 패턴: 별도 workqueue 사용 또는 설계 변경 */
static void work_a_func(struct work_struct *work)
{
    do_part1();
    /* work_b를 다른 workqueue에 큐잉 */
    queue_work(separate_wq, &work_b);
    /* 또는 completion 사용 */
    wait_for_completion(&work_b_done);
    do_part2();
}

패턴 3: softirq 우선순위 역전 (starvation)

/* 우선순위 역전 시나리오:
 * 1. 고우선순위 RT 태스크가 CPU에서 실행 중
 * 2. 네트워크 패킷 도착 -> NET_RX softirq 발생
 * 3. RT 태스크가 선점을 허용하지 않아 softirq 지연
 * 4. 네트워크 패킷 처리가 밀려 드롭 발생 */

/* 해결 방법 1: IRQ affinity 분리 */
/* RT 태스크 CPU와 IRQ 처리 CPU를 분리 */
# echo 0-1 > /proc/irq/<NIC_IRQ>/smp_affinity_list
# taskset -c 2-3 ./rt_application

/* 해결 방법 2: PREEMPT_RT + threaded IRQ */
/* 네트워크 IRQ 스레드 우선순위를 RT 태스크보다 높게 설정 */
# chrt -f -p 90 $(pgrep -f "irq/.*eth0")

/* 해결 방법 3: ksoftirqd 우선순위 조정 */
# chrt -f -p 50 $(pgrep ksoftirqd/0)

추가 실전 케이스: USB와 블록 I/O

USB 드라이버 BH 패턴

USB 드라이버는 URB(USB Request Block) 완료 콜백이 인터럽트 컨텍스트에서 호출되므로, 무거운 처리는 workqueue로 위임해야 합니다.

/* USB 완료 콜백 (인터럽트 컨텍스트) */
static void usb_rx_complete(struct urb *urb)
{
    struct my_usb_dev *dev = urb->context;

    switch (urb->status) {
    case 0:  /* 성공 */
        /* 데이터를 버퍼에 복사 (빠른 작업만) */
        memcpy(dev->rx_buf + dev->rx_len,
               urb->transfer_buffer, urb->actual_length);
        dev->rx_len += urb->actual_length;

        /* 무거운 처리는 workqueue로 위임 */
        schedule_work(&dev->rx_work);
        break;
    case -ENOENT:
    case -ECONNRESET:
    case -ESHUTDOWN:
        return;  /* URB 취소됨 */
    default:
        dev_err(&dev->intf->dev, "URB error: %d\n", urb->status);
    }

    /* URB 재제출 (다음 데이터 수신) */
    usb_submit_urb(urb, GFP_ATOMIC);
}

/* workqueue 핸들러 (프로세스 컨텍스트) */
static void usb_rx_work(struct work_struct *work)
{
    struct my_usb_dev *dev = container_of(work,
                                struct my_usb_dev, rx_work);

    mutex_lock(&dev->data_mutex);  /* 슬립 가능 */
    parse_protocol(dev->rx_buf, dev->rx_len);
    deliver_to_userspace(dev);
    dev->rx_len = 0;
    mutex_unlock(&dev->data_mutex);
}

블록 I/O 완료 처리 패턴

/* 블록 I/O 완료 경로: softirq(BLOCK_SOFTIRQ) 또는 IRQ */
/* drivers/block/virtio_blk.c 패턴 */

/* 1. 인터럽트 핸들러 (Top Half) */
static irqreturn_t virtblk_irq(int irq, void *data)
{
    struct virtio_blk_vq *vq = data;

    /* 완료 처리를 위해 BLOCK softirq 스케줄 */
    blk_mq_complete_request(rq);
    return IRQ_HANDLED;
}

/* 2. blk_mq 완료 경로 선택 */
/* blk_mq_complete_request() 내부:
 *   - 같은 CPU에서 완료 가능하면 softirq로 직접 실행
 *   - 다른 CPU면 IPI 전송하여 해당 CPU의 softirq에서 완료
 *   - 목적: 캐시 친화성 최적화 */

/* 3. 완료 콜백 (softirq 또는 프로세스 컨텍스트) */
static void virtblk_done(struct request *rq)
{
    struct virtblk_req *vbr = blk_mq_rq_to_pdu(rq);

    /* 에러 처리 */
    blk_mq_end_request(rq, virtblk_result(vbr));
    /* -> I/O 완료 통지 -> 대기 중인 프로세스 깨움 */
}

문제 해결 FAQ

Bottom Half 사용 시 자주 발생하는 문제와 해결 방법입니다.

Q1: "ksoftirqd가 CPU를 100% 사용합니다"

증상: top에서 ksoftirqd/N이 높은 CPU 사용률

# CPU별 softirq 카운트 확인
watch -n1 'cat /proc/softirqs'

# 특정 softirq가 급증하는지 확인
# NET_RX가 높으면 네트워크 부하, BLOCK이 높으면 I/O 부하

원인: softirq 부하가 너무 높아 ksoftirqd로 처리 위임

해결:

# 1. 네트워크 인터럽트 분산 (RSS/RPS)
ethtool -L eth0 combined 4  # 4개 큐 사용

# 2. IRQ affinity 설정
echo 2 > /proc/irq/<IRQ번호>/smp_affinity_list  # CPU 2에 바인딩

# 3. NAPI budget 조정 (패킷 처리량 제한)
sysctl -w net.core.netdev_budget=300
sysctl -w net.core.netdev_budget_usecs=2000

Q2: tasklet이 실행되지 않습니다

디버깅:

# tasklet 상태 확인 (커널 코드)
printk("tasklet state: %lx\n", my_tasklet.state);
/* TASKLET_STATE_SCHED (0x01): 스케줄됨
   TASKLET_STATE_RUN   (0x02): 실행 중 */

# softirq 통계 확인
cat /proc/softirqs | grep -E 'HI|TASKLET'

가능한 원인:

1. tasklet_disable() 호출됨 - tasklet_enable() 확인
2. tasklet_kill() 호출 후 재스케줄 시도
3. Atomic counter 오류로 state 손상

해결:

/* enable/disable 쌍 확인 */
tasklet_disable(&my_tasklet);
do_something();
tasklet_enable(&my_tasklet);  /* 반드시 호출! */

/* kill 후 재초기화 */
tasklet_kill(&my_tasklet);
tasklet_setup(&my_tasklet, my_func);  /* 재초기화 */

Q3: workqueue가 예상보다 늦게 실행됩니다

증상: schedule_work() 호출 후 수초 지연

디버깅:

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

# 대기 중인 work 확인
cat /proc/PID/stack  # kworker PID

원인:

• 같은 workqueue에서 다른 work이 블로킹 중
• max_active 제한 도달
• CPU-bound workqueue에서 I/O 대기

해결:

/* 전용 workqueue 생성 */
struct workqueue_struct *my_wq;

my_wq = alloc_workqueue("my_fast_wq",
                        WQ_HIGHPRI | WQ_UNBOUND,
                        0);  /* 제한 없음 */

queue_work(my_wq, &my_work);  /* system_wq 대신 사용 */

Q4: "BUG: sleeping function called from invalid context" 에러

증상: softirq/tasklet에서 슬립 함수 호출

BUG: sleeping function called from invalid context at kernel/locking/mutex.c:...
in_atomic(): 1, irqs_disabled(): 0, non_block: 0, pid: 0, name: swapper/0

원인: atomic context에서 블로킹 함수 호출

해결:

/* ❌ tasklet에서 mutex */
static void my_tasklet(struct tasklet_struct *t)
{
    mutex_lock(&my_mutex);  /* 에러 발생 */
    do_something();
    mutex_unlock(&my_mutex);
}

/* ✅ workqueue로 변경 */
static void my_work(struct work_struct *work)
{
    mutex_lock(&my_mutex);  /* OK */
    do_something();
    mutex_unlock(&my_mutex);
}

Q5: PREEMPT_RT에서 지연시간이 증가했습니다

증상: CONFIG_PREEMPT_RT 활성화 후 응답 시간 저하

원인: tasklet이 ksoftirqd에서 실행되며 스케줄링 지연 발생

해결:

/* tasklet을 threaded IRQ로 변경 */
int ret = request_threaded_irq(
    irq,
    my_hardirq_handler,  /* Top Half */
    my_thread_fn,         /* Threaded Bottom Half */
    IRQF_ONESHOT,
    "my_device",
    dev);

/* 스레드 우선순위 조정 */
struct sched_param param = { .sched_priority = 50 };
sched_setscheduler(current, SCHED_FIFO, ¶m);

Q6: workqueue flush 시 데드락 발생

증상: flush_workqueue()에서 멈춤

# 스택 트레이스 확인
cat /proc/PID/stack
# [<ffffffff>] flush_workqueue+0x...
# [<ffffffff>] my_cleanup+0x...

원인: work 내부에서 자신이 속한 workqueue flush

해결:

/* ❌ 데드락 패턴 */
static void my_work(struct work_struct *work)
{
    flush_workqueue(system_wq);  /* 데드락! */
}

/* ✅ 올바른 정리 순서 */
void module_exit(void)
{
    /* 1. 새 work 스케줄 중단 */
    shutdown_flag = 1;
    smp_wmb();

    /* 2. pending work 취소 */
    cancel_work_sync(&my_work);

    /* 3. workqueue flush (외부에서) */
    flush_workqueue(my_wq);

    /* 4. workqueue 파괴 */
    destroy_workqueue(my_wq);
}

관련 문서

Bottom Half와 관련된 다른 주제를 더 깊이 이해하고 싶다면 다음 문서를 참고하세요.

• 인터럽트 — 인터럽트 Top Half, threaded IRQ
• Softirq & Hardirq — softirq 내부 구현 심화
• Tasklet 심화 — 상태 머신, 직렬화, 마이그레이션
• Workqueue (CMWQ) 심화 — CMWQ 아키텍처, worker pool
• 프로세스 스케줄러 — ksoftirqd/kworker 스레드의 스케줄링 정책
• 타이머 — 타이머 기반 지연 실행
• 디바이스 드라이버 — 드라이버에서의 Bottom Half 사용