NAPI (New API) — 네트워크 패킷(Packet) 처리

NAPI는 리눅스 커널의 핵심 수신 경로 최적화 메커니즘입니다. 인터럽트(Interrupt) 폭풍을 방지하고 고속 네트워크 환경에서 최대 처리량(Throughput)을 달성하기 위해 폴링(Polling) 기반 배치 처리를 활용합니다. napi_struct의 내부 구조부터 드라이버 구현 패턴, 멀티큐 스케일링, 스레드(Thread) NAPI, NAPI 메모리 관리(napi_alloc_skb, page_pool), 해시 테이블(Hash Table) 기반 소켓(Socket) 바인딩, IRQ 일시 중단(Suspension), 버지 폴링, XDP 연동까지 전 영역을 상세히 다룹니다.

NAPI 개요와 탄생 배경

인터럽트 기반 수신의 한계

초기 리눅스 네트워크 드라이버는 패킷이 도착할 때마다 하드웨어 인터럽트(IRQ)를 발생시키고, 각 인터럽트 핸들러(Handler)에서 패킷을 직접 처리했습니다. 100 Mbps 시대에는 이 방식으로 충분했으나, 1 Gbps 이상의 고속 네트워크에서는 치명적인 문제가 드러났습니다.

링크 속도별 초당 최대 패킷 수

pre-NAPI vs NAPI 코드 경로 비교

인터럽트 완화 전후 CPU 사용률 비교

라이브록(Livelock) 발생 시나리오

라이브록은 시스템이 실제로 유용한 작업을 처리하지 못하고 인터럽트 처리에만 매달리는 상태입니다. pre-NAPI 환경에서 다음과 같이 발생합니다:

1. 패킷 A 도착 → IRQ 핸들러 진입, 패킷 처리 시작
2. 패킷 처리 도중 패킷 B, C, D가 연속 도착 → 새 IRQ 발생
3. 현재 IRQ 핸들러 종료 즉시 다음 IRQ 처리 시작
4. IRQ 처리가 끊이지 않아 사용자 공간(User Space) 프로세스(Process), 타이머(Timer), TCP 재전송(Retransmission) 등이 실행 불가
5. 결과: 패킷은 도착하지만 소켓 버퍼(Buffer)에 전달되지 않아 TCP 타임아웃 발생

NAPI는 첫 번째 IRQ에서 이후 IRQ를 비활성화하고 softIRQ 컨텍스트에서 배치 처리함으로써 이 라이브록을 원천 차단합니다. netdev_budget과 netdev_budget_usecs 제한으로 softIRQ도 CPU를 독점하지 못하도록 합니다.

softnet_data 구조체(Struct)와 NAPI의 관계

softnet_data는 per-CPU 자료구조로, NAPI 폴링의 실제 큐 역할을 합니다. 커널 소스 include/linux/netdevice.h에 정의되어 있습니다.

/* include/linux/netdevice.h (Linux 6.x, 주요 필드만 발췌) */
struct softnet_data {
    /* poll_list: NAPI 인스턴스들이 등록되는 링크드 리스트.
       net_rx_action()이 이 리스트를 순회하며 각 napi_struct를 폴링함 */
    struct list_head    poll_list;

    /* output_queue: TX 완료 처리 대기 큐 */
    struct Qdisc       *output_queue;
    struct Qdisc      **output_queue_tailp;

    /* completion_queue: 해제 대기 중인 sk_buff 체인.
       softIRQ에서 일괄 해제하여 IRQ 컨텍스트에서의 해제 비용 감소 */
    struct sk_buff     *completion_queue;

    /* input_pkt_queue: RPS가 다른 CPU로 패킷을 전달할 때 사용하는 큐 */
    struct sk_buff_head input_pkt_queue;

    /* backlog: 단일 큐 NIC 또는 loopback에서 사용하는 기본 NAPI 인스턴스 */
    struct napi_struct  backlog;

    /* time_squeeze: 버짓/시간 초과로 인해 소프트IRQ가 조기 종료된 횟수.
       /proc/net/softnet_stat의 3번째 열 */
    unsigned int        time_squeeze;

    /* received_rps: RPS를 통해 이 CPU로 전달된 패킷 수 */
    unsigned int        received_rps;

    /* dropped: backlog 큐 초과로 드롭된 패킷 수.
       /proc/net/softnet_stat의 2번째 열 */
    unsigned int        dropped;

#ifdef CONFIG_RPS
    /* rps_ipi_list: RPS IPI 대기 리스트 */
    struct softnet_data *rps_ipi_list;
#endif
};

NAPI 커널 버전별 주요 개선 이력

NAPI의 주요 설계 원칙

NAPI 처리 흐름

핵심 자료구조: napi_struct

napi_struct는 NAPI의 핵심 자료구조로, 각 수신 큐(RX queue)마다 하나씩 존재합니다. 커널 소스의 include/linux/netdevice.h에 정의되어 있습니다.

/* include/linux/netdevice.h (Linux 6.x 기준, 일부 생략) */
struct napi_struct {
    /* poll_list: softirq의 NET_RX_SOFTIRQ가 순회하는 링크드 리스트 */
    struct list_head    poll_list;

    /* state: 원자적으로 조작되는 상태 비트맵 (NAPI_STATE_*) */
    unsigned long       state;

    /* weight: 한 번의 poll() 호출에서 처리할 최대 패킷 수 (버짓) */
    int                 weight;

    /* defer_hard_irqs_count: 지연된 하드 IRQ 재활성화 카운터 */
    int                 defer_hard_irqs_count;

    /* gro_bitmask: GRO 활성 버킷 비트마스크 (gro_hash 중 유효 버킷 표시) */
    unsigned long       gro_bitmask;

    /* poll: 드라이버가 등록하는 폴링 함수 포인터 */
    int                 (*poll)(struct napi_struct *, int);

#ifdef CONFIG_NETPOLL
    struct netpoll_info __rcu *napi_id_list;
#endif

    /* dev: 이 NAPI가 속한 net_device */
    struct net_device   *dev;

    /* gro_hash: GRO 병합 대기 중인 skb 체인 (GRO_HASH_BUCKETS=8 버킷) */
    struct gro_list     gro_hash[GRO_HASH_BUCKETS];

    /* skb: 현재 처리 중인 skb (GRO 경로에서 사용) */
    struct sk_buff      *skb;

    /* rx_list: 처리 완료된 skb들의 임시 큐 */
    struct list_head    rx_list;
    int                 rx_count;

    /* napi_id: NAPI 인스턴스 고유 ID (busy polling 식별용, MIN_NAPI_ID 이상) */
    unsigned int        napi_id;

    /* threaded: 스레드 NAPI 사용 여부 */
    u8                  threaded;

    /* thread: 스레드 NAPI 전용 커널 스레드 포인터 */
    struct task_struct  *thread;

    /* dev_list: net_device의 napi_list에 연결 */
    struct list_head    dev_list;

    /* poll_owner: 현재 poll()을 실행 중인 CPU (-1이면 유휴) */
    int                 poll_owner;
};

gro_hash 버킷 구조 상세

GRO는 napi_struct 내의 gro_hash 배열에 패킷을 버킷 단위로 보관합니다. 버킷 수는 GRO_HASH_BUCKETS(8)이며, 해시(Hash) 키로 플로우를 분산시킵니다.

/* net/core/dev.c (Linux 6.x) */
#define GRO_HASH_BUCKETS  8

struct gro_list {
    /* list: 동일 해시 버킷의 GRO 대기 skb 체인 */
    struct list_head  list;
    /* count: 버킷 내 skb 개수 (MAX_GRO_SKBS=8 초과 시 flush) */
    int               count;
};

/* NAPI_GRO_CB: skb->cb 영역에 GRO 전용 메타데이터 저장 */
struct napi_gro_cb {
    /* data_offset: skb->data에서 헤더 시작까지의 오프셋 */
    unsigned int   data_offset;

    /* flush: 즉시 flush 필요 여부 (순서 역전, 헤더 불일치 등) */
    u8             flush;

    /* flush_id: 병합 후 flush할 패킷 식별 ID */
    u16            flush_id;

    /* count: 이 GRO skb에 병합된 패킷 수 */
    u16            count;

    /* same_flow: 동일 플로우로 판별된 경우 true */
    u8             same_flow;

    /* ip_fixedid: IP ID가 고정(incrementing)인지 여부 */
    u8             ip_fixedid;

    /* encap_mark: 터널 헤더 처리 중임을 표시 */
    u8             encap_mark;

    /* csum_valid: 체크섬 이미 검증됨 */
    u8             csum_valid;

    /* is_atomic: 원자적 GRO (단편화 없음) */
    u8             is_atomic;

    /* tot_len: 병합된 전체 페이로드 길이 */
    unsigned int   tot_len;
};

#define NAPI_GRO_CB(skb) ((struct napi_gro_cb *)(skb)->cb)

napi_struct와 net_device의 연결 관계

/* include/linux/netdevice.h, net/core/dev.c — 간략화 */
/* net_device는 등록된 모든 NAPI 인스턴스를 napi_list로 추적 */
struct net_device {
    /* ... */
    /* napi_list: 이 디바이스의 모든 napi_struct 링크드 리스트 */
    struct list_head    napi_list;
    /* ... */
};

/* netif_napi_add() 내부에서 dev->napi_list에 추가 */
void netif_napi_add(struct net_device *dev, struct napi_struct *napi, ...)
{
    /* napi_id 할당: napi_gen_id()로 전역 카운터에서 증가 */
    napi->napi_id = napi_gen_id();   /* >= MIN_NAPI_ID (0x10000) */

    /* napi_hash에 등록 (busy polling 조회용 해시 테이블) */
    napi_hash_add(napi);

    /* dev->napi_list에 연결 */
    list_add_rcu(&napi->dev_list, &dev->napi_list);
    set_bit(NAPI_STATE_LISTED, &napi->state);
}

/* 디바이스의 모든 NAPI 인스턴스 순회 예 */
static void mynic_enable_all_napi(struct net_device *dev)
{
    struct napi_struct *napi;
    list_for_each_entry(napi, &dev->napi_list, dev_list)
        napi_enable(napi);
}

napi_id 할당 메커니즘

napi_id는 버지 폴링에서 특정 NAPI 인스턴스를 찾기 위한 키입니다. 전역 해시 테이블 napi_hash에 등록되어 napi_by_id()로 조회됩니다.

/* net/core/dev.c — 간략화 */
/* MIN_NAPI_ID: 소켓 식별자(sk_napi_id)와 구분을 위한 최솟값 */
#define MIN_NAPI_ID     ((unsigned int)(NR_CPUS + 1))

/* 전역 NAPI ID 카운터 (원자적 증가) */
static atomic_t napi_gen_id_counter = ATOMIC_INIT(MIN_NAPI_ID);

static unsigned int napi_gen_id(void)
{
    unsigned int id;
    do {
        id = atomic_inc_return(&napi_gen_id_counter);
        if (id < MIN_NAPI_ID)
            id = atomic_inc_return(&napi_gen_id_counter);
    } while (napi_by_id(id));  /* 충돌 시 재시도 */
    return id;
}

NAPI_STATE_* 플래그 테이블

NAPI API와 동작 원리

초기화/해제 API

/* include/linux/netdevice.h, net/core/dev.c — 간략화 */
/* NIC의 RX 큐에 NAPI 인스턴스 등록 */
void netif_napi_add(struct net_device *dev,
                    struct napi_struct *napi,
                    int (*poll)(struct napi_struct *, int),
                    int weight);

/* 멀티큐 드라이버: 큐 인덱스와 함께 등록 (Linux 6.1+) */
void netif_napi_add_config(struct net_device *dev,
                           struct napi_struct *napi,
                           int (*poll)(struct napi_struct *, int),
                           int weight, int napi_id);

/* tx 경로 전용 NAPI (weight = NAPI_POLL_WEIGHT) */
void netif_napi_add_tx(struct net_device *dev,
                       struct napi_struct *napi,
                       int (*poll)(struct napi_struct *, int));

/* NAPI 인스턴스 해제 (napi_list에서 제거, napi_hash에서 제거) */
void netif_napi_del(struct napi_struct *napi);

/* NAPI 활성화: NAPI_STATE_DISABLE 클리어 → 스케줄 허용 */
void napi_enable(struct napi_struct *napi);

/* NAPI 비활성화: 진행 중인 poll() 완료 대기 후 DISABLE 비트 설정 */
void napi_disable(struct napi_struct *napi);

/* NAPI poll이 완전히 중단될 때까지 대기 (RCU 동기화 포함) */
void napi_synchronize(const struct napi_struct *napi);

napi_disable() 내부 구현

/* net/core/dev.c */
void napi_disable(struct napi_struct *napi)
{
    unsigned long val, new;

    might_sleep();
    set_bit(NAPI_STATE_DISABLE, &napi->state);

    /* SCHED 또는 SCHED_THREADED 비트가 클리어될 때까지 대기
       (진행 중인 poll()이 완료될 때까지 폴링 대기) */
    do {
        val = READ_ONCE(napi->state);
        if (!(val & (NAPIF_STATE_SCHED | NAPIF_STATE_SCHED_THREADED)))
            break;
        /* 짧게 슬립하여 CPU 낭비 방지 (usleep_range: 200~500μs) */
        usleep_range(200, 500);
    } while (1);

    /* IN_BUSY_POLL도 해소될 때까지 대기 */
    do {
        val = READ_ONCE(napi->state);
        if (!(val & NAPIF_STATE_IN_BUSY_POLL))
            break;
        usleep_range(200, 500);
    } while (1);

    clear_bit(NAPI_STATE_DISABLE, &napi->state);
}

napi_synchronize() vs napi_disable() 차이점

스케줄링 API와 __napi_schedule() 내부 구현

/* include/linux/netdevice.h, net/core/dev.c — 간략화 */
/* IRQ 핸들러에서 NAPI 스케줄 (인터럽트 컨텍스트에서 안전) */
void napi_schedule(struct napi_struct *napi);

/* IRQ 비활성화된 상태에서 스케줄 (local_irq_save 생략으로 더 빠름) */
void napi_schedule_irqoff(struct napi_struct *napi);

/* napi_schedule의 실제 구현 */
static inline void napi_schedule(struct napi_struct *napi)
{
    /* NAPI_STATE_SCHED 비트를 원자적으로 세팅 → 이미 스케줄된 경우 무시 */
    if (napi_schedule_prep(napi))
        __napi_schedule(napi);
}

/* __napi_schedule: 실제로 per-CPU poll_list에 추가 */
void __napi_schedule(struct napi_struct *n)
{
    unsigned long flags;
    struct softnet_data *sd;

    local_irq_save(flags);
    sd = this_cpu_ptr(&softnet_data);

    /* 스레드 NAPI 활성화 시 커널 스레드 깨우기 */
    if (test_bit(NAPI_STATE_THREADED, &n->state)) {
        __napi_schedule_threaded(n);
    } else {
        /* per-CPU poll_list에 추가 (tail) */
        list_add_tail(&n->poll_list, &sd->poll_list);
        /* NET_RX_SOFTIRQ 트리거 */
        __raise_softirq_irqoff(NET_RX_SOFTIRQ);
    }
    local_irq_restore(flags);
}

완료 API와 napi_complete_done() 내부 로직

/* net/core/dev.c — 간략화 */
/* poll() 내에서 패킷 소진 시 호출 */
bool napi_complete_done(struct napi_struct *napi, int work_done)
{
    unsigned long flags, val, new_val;

    /* 1. GRO 버퍼 flush: 남은 병합 패킷을 상위 스택으로 전달 */
    if (napi->gro_bitmask)
        napi_gro_flush(napi, false);

    /* 2. gro_flush_timeout이 설정된 경우 타이머 재설정 */
    if (work_done) {
        if (READ_ONCE(napi->dev->gro_flush_timeout))
            hrtimer_start(&napi->timer,
                           READ_ONCE(napi->dev->gro_flush_timeout),
                           HRTIMER_MODE_REL_PINNED);
    }

    /* 3. NAPI_STATE_MISSED 확인: 폴링 중 새 IRQ가 왔는가? */
    local_irq_save(flags);
    val = READ_ONCE(napi->state);

    if (unlikely(val & NAPIF_STATE_MISSED)) {
        /* MISSED: 즉시 재스케줄 (GRO는 이미 flush됨) */
        __napi_schedule(napi);
        local_irq_restore(flags);
        return false;  /* IRQ 재활성화 금지 */
    }

    /* 4. NAPI_STATE_SCHED 클리어 → NAPI 완전 종료 */
    new_val = val & ~(NAPIF_STATE_MISSED | NAPIF_STATE_SCHED |
                      NAPIF_STATE_SCHED_THREADED | NAPIF_STATE_PREFER_BUSY_POLL);
    WRITE_ONCE(napi->state, new_val);
    local_irq_restore(flags);

    return true;  /* 드라이버는 HW IRQ 재활성화 필요 */
}

/* 단순화 버전 (work_done = 0으로 호출, GRO flush만 수행) */
static inline bool napi_complete(struct napi_struct *napi)
{
    return napi_complete_done(napi, 0);
}

API 호출 순서 다이어그램

전형적인 IRQ 핸들러 패턴

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
static irqreturn_t mynic_irq_handler(int irq, void *data)
{
    struct mynic_rx_ring *ring = data;
    struct mynic_hw      *hw   = ring->hw;
    u32 status;

    status = mynic_read_reg(hw, MYNIC_IRQ_STATUS);
    if (!(status & MYNIC_RX_INT))
        return IRQ_NONE;

    /* 하드웨어 인터럽트 마스크 (이후 NAPI poll에서 처리) */
    mynic_disable_rx_irq(hw, ring->queue_idx);

    /* NAPI 스케줄: NAPI_STATE_SCHED가 이미 세팅된 경우 무시됨
       _irqoff: IRQ 이미 비활성화 상태이므로 local_irq_save 생략 가능 */
    napi_schedule_irqoff(&ring->napi);

    return IRQ_HANDLED;
}

softnet_data 구조체 상세 분석

softnet_data는 per-CPU 네트워크 처리 상태(Per-CPU Network Processing State)를 관리하는 핵심 구조체입니다. 각 CPU마다 하나씩 존재하며, NAPI poll_list, 패킷 입력 큐, 백로그(Backlog) NAPI 인스턴스, 다양한 통계 카운터를 포함합니다. net/core/dev.c에서 DEFINE_PER_CPU_ALIGNED로 정의되어 캐시 라인(Cache Line) 경계에 정렬됩니다.

/* net/core/dev.c (Linux 6.x) — softnet_data 정의 */
DEFINE_PER_CPU_ALIGNED(struct softnet_data, softnet_data);
EXPORT_PER_CPU_SYMBOL(softnet_data);

/* include/linux/netdevice.h — 구조체 전체 필드 (주요 필드 발췌) */
struct softnet_data {
    /* poll_list: 스케줄된 NAPI 인스턴스들의 연결 리스트.
       net_rx_action()이 이 리스트를 순회하며 각 napi_struct->poll()을 호출.
       local_irq_save() 보호 하에 추가/삭제 */
    struct list_head    poll_list;

    /* input_pkt_queue: RPS(Receive Packet Steering) 또는 netif_rx()로
       인입된 패킷을 임시 저장하는 큐. 다른 CPU에서 IPI를 통해 전달됨.
       backlog NAPI가 이 큐에서 패킷을 꺼내 처리 */
    struct sk_buff_head input_pkt_queue;

    /* process_queue: backlog NAPI의 poll 루프에서 처리 중인 패킷 큐.
       input_pkt_queue에서 splice로 옮겨와 lock 경합을 최소화 */
    struct sk_buff_head process_queue;

    /* backlog: per-CPU 디폴트 NAPI 인스턴스.
       poll 콜백은 process_backlog(). 단일 큐 NIC, loopback,
       netif_rx() 경로에서 사용 */
    struct napi_struct  backlog;

    /* output_queue / output_queue_tailp: TX completion 대기 큐.
       qdisc_run()에서 처리하지 못한 TX 패킷들의 Qdisc를 연결 */
    struct Qdisc       *output_queue;
    struct Qdisc      **output_queue_tailp;

    /* completion_queue: 해제 대기 중인 sk_buff 체인.
       IRQ 컨텍스트에서 직접 해제하지 않고 softIRQ에서 일괄 해제 */
    struct sk_buff     *completion_queue;

    /* === 통계 카운터 (per-CPU, 락 불필요) === */

    /* dropped: netif_rx() 경로에서 backlog 큐(input_pkt_queue) 초과 드롭.
       /proc/net/softnet_stat 1번째 열 */
    unsigned int        dropped;

    /* time_squeeze: net_rx_action()에서 버짓 또는 시간 초과로
       poll_list에 아직 작업이 남았는데 조기 종료된 횟수.
       /proc/net/softnet_stat 2번째 열 */
    unsigned int        time_squeeze;

    /* cpu_collision: TX에서 다른 CPU가 이미 Qdisc lock을 잡고 있어
       경합이 발생한 횟수. /proc/net/softnet_stat 3번째 열 */
    unsigned int        cpu_collision;

    /* received_rps: RPS IPI를 통해 이 CPU로 전달된 패킷 수.
       /proc/net/softnet_stat 4번째 열 */
    unsigned int        received_rps;

    /* flow_limit_count: flow limit 히트 횟수.
       특정 플로우가 CPU 시간을 독점하지 못하도록 제한.
       /proc/net/softnet_stat 5번째 열 */
    unsigned int        flow_limit_count;

#ifdef CONFIG_RPS
    /* rps_ipi_list: RPS IPI 전송 대기 중인 remote CPU 리스트.
       NET_RX_SOFTIRQ 처리 시작 시 IPI를 일괄 전송 */
    struct softnet_data *rps_ipi_list;

    /* input_queue_head/tail: flow limit 판단용 큐 위치 추적 */
    unsigned int        input_queue_head;
    unsigned int        input_queue_tail;
#endif

    /* xmit.recursion: TX 재진입 방지 카운터.
       dev_queue_xmit() 내부에서 nesting depth 추적 (최대 XMIT_RECURSION_LIMIT) */
    unsigned int        xmit_recursion;
};

per-CPU softnet_data와 NAPI poll_list 관계

NAPI 핵심 함수 내부 구현

netif_napi_add() 내부 구현

netif_napi_add()는 드라이버가 NAPI 인스턴스를 등록할 때 호출하는 진입점(Entry Point)입니다. 내부에서 napi_struct의 모든 필드를 초기화하고, GRO 해시 버킷 생성, 해시 테이블 등록, 디바이스 연결까지 일괄 수행합니다.

/* net/core/dev.c (Linux 6.x) — netif_napi_add() 내부 구현 */
void netif_napi_add_weight(struct net_device *dev,
                           struct napi_struct *napi,
                           int (*poll)(struct napi_struct *, int),
                           int weight)
{
    /* 커널 소스 분석: poll_list 초기화 — 빈 리스트로 설정 */
    INIT_LIST_HEAD(&napi->poll_list);

    /* 커널 소스 분석: GRO 해시 버킷 초기화 (8개 버킷) */
    for (int i = 0; i < GRO_HASH_BUCKETS; i++) {
        INIT_LIST_HEAD(&napi->gro_hash[i].list);
        napi->gro_hash[i].count = 0;
    }
    napi->gro_bitmask = 0;

    /* 커널 소스 분석: hrtimer 초기화 — GRO flush 타이머 */
    hrtimer_init(&napi->timer, CLOCK_MONOTONIC, HRTIMER_MODE_REL_PINNED);
    napi->timer.function = napi_watchdog;

    /* 커널 소스 분석: 핵심 필드 설정 */
    napi->poll = poll;              /* 드라이버 폴링 콜백 */
    napi->weight = weight;          /* 기본값 NAPI_POLL_WEIGHT = 64 */
    napi->dev = dev;                /* 소속 net_device */
    napi->poll_owner = -1;          /* 현재 poll 실행 중인 CPU (-1 = 없음) */
    napi->list_owner = -1;

    /* 커널 소스 분석: 상태 비트 초기화 — SCHED 비트 설정 (비활성 상태)
       napi_enable()이 호출될 때까지 스케줄 불가 */
    set_bit(NAPI_STATE_SCHED, &napi->state);
    set_bit(NAPI_STATE_NPSVC, &napi->state);

    /* 커널 소스 분석: napi_hash_add() — 전역 해시 테이블에 등록 */
    napi_hash_add(napi);

    /* 커널 소스 분석: dev->napi_list에 RCU 보호 하에 추가 */
    list_add_rcu(&napi->dev_list, &dev->napi_list);

    /* 커널 소스 분석: 디바이스가 threaded NAPI를 사용하면 상속 */
    napi_set_threaded(napi, dev->threaded);
}

napi_hash_add() 내부 동작

napi_hash_add()는 NAPI 인스턴스에 전역 고유 ID를 할당하고, 해시 테이블(Hash Table)에 삽입합니다. 이 ID는 버지 폴링(Busy Polling)에서 napi_by_id()를 통해 소켓과 NAPI 인스턴스를 연결하는 데 사용됩니다.

/* net/core/dev.c — napi_hash_add() 내부 */
static void napi_hash_add(struct napi_struct *napi)
{
    /* 커널 소스 분석: IDA(ID Allocator) 기반 ID 할당
       MIN_NAPI_ID(NR_CPUS+1)부터 시작하여 소켓 ID와 충돌 방지 */
    if (test_bit(NAPI_STATE_NO_BUSY_POLL, &napi->state))
        return;

    spin_lock(&napi_hash_lock);

    /* 전역 카운터에서 원자적 ID 할당 */
    napi->napi_id = napi_gen_id();

    /* 커널 소스 분석: RCU 보호 해시 테이블에 삽입
       napi_hash[]는 NAPI_HASH_SIZE 크기의 hlist 배열 */
    hlist_add_head_rcu(&napi->napi_hash_node,
                       &napi_hash[napi->napi_id % NAPI_HASH_SIZE]);

    spin_unlock(&napi_hash_lock);
}

/* busy polling에서 napi_id로 인스턴스 조회 */
struct napi_struct *napi_by_id(unsigned int napi_id)
{
    struct napi_struct *napi;

    hlist_for_each_entry_rcu(napi,
        &napi_hash[napi_id % NAPI_HASH_SIZE], napi_hash_node)
        if (napi->napi_id == napi_id)
            return napi;
    return NULL;
}

napi_enable() 내부 구현

napi_enable()은 NAPI 인스턴스를 활성 상태로 전환합니다. netif_napi_add()에서 설정한 NAPI_STATE_SCHED 비트를 클리어하여 napi_schedule() 호출이 가능하도록 합니다. 원자적 비교-교환(Compare-and-Exchange, cmpxchg)으로 레이스 컨디션(Race Condition)을 방지합니다.

/* net/core/dev.c (Linux 6.x) — napi_enable() 내부 구현 */
void napi_enable(struct napi_struct *napi)
{
    unsigned long new, val = READ_ONCE(napi->state);

    do {
        /* 커널 소스 분석: SCHED 비트가 반드시 설정되어 있어야 함.
           netif_napi_add()가 SCHED을 설정하므로 정상이면 통과.
           이미 enable된 상태에서 재호출하면 BUG_ON 트리거 */
        BUG_ON(!test_bit(NAPI_STATE_SCHED, &val));

        /* 커널 소스 분석: SCHED와 NPSVC(No Poll Service) 비트를 클리어.
           SCHED 클리어 → napi_schedule_prep()이 성공 가능
           NPSVC 클리어 → poll service 가능 상태 */
        new = val & ~(NAPIF_STATE_SCHED | NAPIF_STATE_NPSVC);

        /* 커널 소스 분석: 스레드 NAPI인 경우 THREADED 비트 설정.
           napi->thread가 존재하면 poll이 커널 스레드에서 실행됨 */
        new |= NAPIF_STATE_THREADED * !!napi->thread;

    } while (!try_cmpxchg(&napi->state, &val, new));

    /* 커널 소스 분석: 스레드 NAPI 활성화 시 커널 스레드를 깨움.
       napi_threaded_poll() 루프가 시작됨 */
    if (napi->thread)
        wake_up_process(napi->thread);
}

napi_schedule_prep() 내부 구현

napi_schedule_prep()은 NAPI 인스턴스가 스케줄 가능한 상태인지 확인하고, 가능하다면 NAPI_STATE_SCHED 비트를 원자적으로 설정합니다. 이 함수가 true를 반환해야만 __napi_schedule()을 호출할 수 있습니다.

/* include/linux/netdevice.h (Linux 6.x) — napi_schedule_prep() */
static inline bool napi_schedule_prep(struct napi_struct *n)
{
    unsigned long new, val = READ_ONCE(n->state);

    do {
        /* 커널 소스 분석: DISABLE 상태면 스케줄 불가.
           napi_disable() 진행 중이면 즉시 실패 반환 */
        if (unlikely(val & NAPIF_STATE_DISABLE))
            return false;

        /* 커널 소스 분석: 이미 SCHED 비트가 설정되어 있으면 중복 스케줄 방지.
           IRQ 핸들러가 동시에 호출되어도 하나만 성공.
           단, MISSED 비트를 설정하여 napi_complete_done()에 알림 */
        if (unlikely(val & NAPIF_STATE_SCHED))
            return false;

        /* 커널 소스 분석: SCHED 비트 원자적 설정 (cmpxchg 루프) */
        new = val | NAPIF_STATE_SCHED;
    } while (!try_cmpxchg(&n->state, &val, new));

    return true;
}

주목할 점은 이미 스케줄된 상태에서 새 패킷이 도착하면 napi_schedule_prep()이 false를 반환하는 것입니다. 이때 최신 커널(6.x)에서는 드라이버가 직접 NAPI_STATE_MISSED 비트를 설정하여, napi_complete_done()이 이를 감지하고 즉시 재스케줄하도록 합니다.

__napi_schedule() 내부 구현

__napi_schedule()은 NAPI 인스턴스를 실제로 per-CPU poll_list에 추가하고 NET_RX_SOFTIRQ를 발생시키는 함수입니다. 스레드 NAPI(Threaded NAPI)인 경우에는 softIRQ 대신 전용 커널 스레드를 깨웁니다.

/* net/core/dev.c (Linux 6.x) — __napi_schedule() */
void __napi_schedule(struct napi_struct *n)
{
    unsigned long flags;

    /* 커널 소스 분석: local_irq_save — 중첩 인터럽트 방지.
       per-CPU poll_list 접근 중 다른 IRQ가 동일 리스트를 수정하는 것을 방지 */
    local_irq_save(flags);
    ____napi_schedule(this_cpu_ptr(&softnet_data), n);
    local_irq_restore(flags);
}

/* 실제 스케줄링 수행 — 이중 언더스코어 버전 */
static inline void ____napi_schedule(
    struct softnet_data *sd,
    struct napi_struct *napi)
{
    struct task_struct *thread;

    /* 커널 소스 분석: 스레드 NAPI 경로 확인.
       NAPI_STATE_THREADED 비트가 설정되어 있으면 커널 스레드에서 poll 실행 */
    if (test_bit(NAPI_STATE_THREADED, &napi->state)) {
        thread = READ_ONCE(napi->thread);
        if (thread) {
            /* 커널 소스 분석: 스레드가 이미 종료되었으면 경고 */
            if (WARN_ON(thread->state == TASK_DEAD))
                return;
            /* SCHED_THREADED 비트 설정: 스레드가 poll을 실행할 것임을 표시 */
            set_bit(NAPI_STATE_SCHED_THREADED, &napi->state);
            /* 커널 스레드 깨우기: napi_threaded_poll() 루프 재개 */
            wake_up_process(thread);
            return;
        }
    }

    /* 커널 소스 분석: 일반 NAPI 경로 — per-CPU poll_list 추가.
       리스트 tail에 추가하여 FIFO 순서 유지 (공정 스케줄링) */
    list_add_tail(&napi->poll_list, &sd->poll_list);

    /* 커널 소스 분석: NET_RX_SOFTIRQ 발생.
       이미 IRQ 비활성화 상태이므로 _irqoff 변형 사용.
       softIRQ는 do_softirq() → net_rx_action()으로 실행됨 */
    __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

napi_complete_done() 내부 구현

napi_complete_done()은 NAPI 폴링 사이클의 완료(Completion)를 처리하는 가장 중요한 함수입니다. GRO 플러시(Flush), IRQ 일시 중단(Suspension), MISSED 비트 감지, 상태 비트 클리어를 모두 하나의 원자적 경로에서 수행합니다.

/* net/core/dev.c (Linux 6.x) — napi_complete_done() 전체 구현 */
bool napi_complete_done(struct napi_struct *n, int work_done)
{
    unsigned long flags, val, new, timeout = 0;
    bool ret = true;

    /* ① GRO flush 타이머 설정 또는 즉시 flush.
       work_done > 0이면 gro_flush_timeout 값을 확인.
       napi->gro_bitmask가 설정되어 있으면 GRO 리스트에 미완성 패킷 존재 */
    if (work_done)
        timeout = READ_ONCE(n->dev->gro_flush_timeout);

    if (n->gro_bitmask) {
        /* 커널 소스 분석: timeout 설정 시 hrtimer로 지연 flush.
           타이머 만료 시 napi_watchdog()이 GRO flush + 재스케줄 */
        if (timeout)
            hrtimer_start(&n->timer, ns_to_ktime(timeout),
                         HRTIMER_MODE_REL_PINNED);
        else
            /* timeout 미설정 시 즉시 GRO flush → 프로토콜 스택 전달 */
            napi_gro_flush(n, false);
    }

    /* ② IRQ suspension (defer_hard_irqs) 처리.
       napi_defer_hard_irqs > 0으로 설정된 경우:
       IRQ unmask를 지연하고 타이머 기반으로 폴링을 반복.
       초저지연(Ultra-Low Latency) 환경에서 IRQ 오버헤드를 더욱 줄임 */
    if (READ_ONCE(n->defer_hard_irqs_count) && work_done) {
        n->defer_hard_irqs_count--;
        timeout = READ_ONCE(n->dev->gro_flush_timeout);
        if (timeout) {
            hrtimer_start(&n->timer, ns_to_ktime(timeout),
                         HRTIMER_MODE_REL_PINNED);
        }
        /* 커널 소스 분석: SCHED 비트를 유지한 채 false 반환.
           → 드라이버가 IRQ를 재활성화하지 않음.
           → 타이머 만료 시 napi_watchdog()이 재스케줄 */
        return false;
    }

    /* ③ SCHED 비트 클리어 (원자적 cmpxchg 루프) */
    val = READ_ONCE(n->state);
    do {
        /* 커널 소스 분석: SCHED와 PREFER_BUSY_POLL 비트 클리어 */
        new = val & ~(NAPIF_STATE_SCHED | NAPIF_STATE_PREFER_BUSY_POLL);

        if (unlikely(val & NAPIF_STATE_MISSED)) {
            /* 커널 소스 분석: MISSED 비트 감지!
               poll() 실행 중에 새 패킷이 도착하여 napi_schedule_prep()이
               실패했지만, 드라이버가 MISSED 비트를 설정한 경우.
               → MISSED 클리어 + SCHED 유지 → 즉시 재스케줄
               → 패킷 손실(Packet Loss) 방지의 핵심 메커니즘 */
            new |= NAPIF_STATE_SCHED;
            new &= ~NAPIF_STATE_MISSED;
            ret = false;  /* 호출자에게 "IRQ 재활성화 불필요" 알림 */
        }
    } while (!try_cmpxchg(&n->state, &val, new));

    /* ④ 반환값 계약(Return Value Contract):
       ret == true  → 정상 완료. 드라이버가 HW IRQ를 재활성화해야 함
       ret == false → MISSED 감지 또는 defer_hard_irqs.
                      IRQ 재활성화 불필요 (자동 재스케줄됨) */

    return ret;
}

MISSED 비트의 핵심 역할

NAPI_STATE_MISSED 비트는 NAPI의 패킷 손실 방지(Packet Loss Prevention) 메커니즘의 핵심입니다. 다음과 같은 레이스 컨디션을 해결합니다:

1. NAPI poll() 실행 중 (SCHED 비트 설정 상태)
2. 새로운 패킷이 NIC에 도착하여 하드웨어 IRQ 발생
3. IRQ 핸들러에서 napi_schedule_prep() 호출 → SCHED 이미 설정되어 false 반환
4. IRQ 핸들러가 NAPI_STATE_MISSED 비트를 설정
5. 현재 poll()이 완료되고 napi_complete_done() 호출
6. napi_complete_done()이 MISSED 비트를 감지하여 SCHED 유지
7. net_rx_action()이 즉시 다음 poll 사이클 실행 → 새 패킷 처리

이 메커니즘이 없으면, poll 중 도착한 패킷의 IRQ가 사라지고 다음 패킷이 올 때까지 해당 큐의 패킷이 영구적으로 처리되지 않는 문제가 발생합니다.

napi_complete_done() 결정 흐름

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* 드라이버 poll() 함수 — 올바른 napi_complete_done() 사용 예 */
static int mynic_poll(struct napi_struct *napi, int budget)
{
    struct mynic_priv *priv = container_of(napi, struct mynic_priv, napi);
    int work_done = 0;

    work_done = mynic_clean_rx(priv, budget);

    if (work_done < budget) {
        /* 커널 소스 분석: budget 미소진 → 모든 패킷 처리 완료 */
        if (napi_complete_done(napi, work_done))
            mynic_enable_irq(priv);  /* true일 때만 IRQ unmask! */
    }

    return work_done;
}

napi_schedule_irqoff() 내부

napi_schedule_irqoff()는 napi_schedule()의 최적화 변형(Optimized Variant)입니다. 이미 인터럽트가 비활성화된 컨텍스트(하드웨어 IRQ 핸들러 내부)에서 호출되며, local_irq_save()/local_irq_restore() 쌍을 생략합니다.

/* include/linux/netdevice.h (Linux 6.x) — napi_schedule_irqoff() */
static inline void napi_schedule_irqoff(struct napi_struct *n)
{
    /* 커널 소스 분석: napi_schedule_prep()로 SCHED 비트 원자적 설정 시도 */
    if (napi_schedule_prep(n))
        /* local_irq_save/restore 없이 직접 ____napi_schedule 호출.
           호출자가 이미 IRQ를 비활성화한 상태를 전제 (하드 IRQ 핸들러 등) */
        ____napi_schedule(this_cpu_ptr(&softnet_data), n);
}

/* 비교: napi_schedule()은 local_irq_save/restore를 포함 */
static inline void napi_schedule(struct napi_struct *n)
{
    if (napi_schedule_prep(n))
        __napi_schedule(n);  /* 내부에서 local_irq_save() 호출 */
}

폴링 메커니즘과 버짓 관리

net_rx_action() 내부 동작

소프트IRQ NET_RX_SOFTIRQ가 트리거되면 net_rx_action()이 실행됩니다. 이 함수가 NAPI 폴링의 핵심 루프를 담당합니다.

/* net/core/dev.c */
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(netdev_budget_usecs);
    int                  budget     = netdev_budget;  /* 기본값 300 */
    struct list_head     list;
    struct list_head     repoll;

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

    INIT_LIST_HEAD(&repoll);

    for (;;) {
        struct napi_struct *n;

        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);
        list_del_init(&n->poll_list);

        /* napi_poll: 실제 드라이버 poll() 호출, work_done 반환 */
        budget -= napi_poll(n, &repoll);

        /* 전체 버짓 소진 또는 시간 초과 시 조기 종료 */
        if (unlikely(budget <= 0 ||
                    time_after_eq(jiffies(), time_limit))) {
            sd->time_squeeze++;  /* /proc/net/softnet_stat 열 3 증가 */
            break;
        }
    }

    /* 처리 못한 NAPI들을 다시 poll_list에 연결 후 softIRQ 재트리거 */
    local_irq_disable();
    list_splice_tail_init(&sd->poll_list, &list);
    list_splice_tail(&repoll, &list);
    list_splice(&list, &sd->poll_list);

    if (!list_empty(&sd->poll_list))
        __raise_softirq_irqoff(NET_RX_SOFTIRQ);  /* ksoftirqd 깨우기 */

    net_rps_action_and_irq_enable(sd);
}

napi_poll() 내부 함수 구현

/* net/core/dev.c: 실제 드라이버 poll()을 호출하는 내부 래퍼 */
static int napi_poll(struct napi_struct *n, struct list_head *repoll)
{
    int  work, weight;

    /* 현재 CPU를 poll_owner로 기록 (동시 실행 방지) */
    WRITE_ONCE(n->poll_owner, smp_processor_id());

    weight = n->weight;

    /* 드라이버 poll() 호출: 처리한 패킷 수 반환 */
    work = n->poll(n, weight);

    if (unlikely(work > weight))
        pr_err_once("NAPI poll function %pS returned %d, exceeding weight %d\n",
                    n->poll, work, weight);

    if (work < weight) {
        /* 버짓 미달: 큐 소진 → poll_list에서 제거됨
           (드라이버가 napi_complete_done() 호출하여 SCHED 비트 클리어) */
        if (unlikely(napi_is_scheduled(n))) {
            /* MISSED 비트 설정됨: 재스케줄이 필요 */
            list_add_tail(&n->poll_list, repoll);
        }
    } else {
        /* 버짓 전부 사용: 더 처리할 패킷 있을 가능성 → repoll 큐에 추가 */
        if (unlikely(test_and_clear_bit(NAPI_STATE_MISSED, &n->state))) {
            /* 폴링 중 IRQ 발생: 즉시 재스케줄 */
            napi_schedule(n);
        } else {
            list_add_tail(&n->poll_list, repoll);
        }
    }

    WRITE_ONCE(n->poll_owner, -1);
    return work;
}

버짓 분배 시나리오

3개의 NIC가 각각 weight=64로 등록된 환경에서 netdev_budget=300일 때의 동작:

time_squeeze 발생 시나리오

time_squeeze는 두 가지 상황에서 증가합니다:

• 버짓 소진: 300 패킷을 처리했지만 poll_list에 더 처리할 NAPI가 남아있을 때
• 시간 초과: netdev_budget_usecs(기본 8000μs) 경과했지만 poll_list가 비어있지 않을 때

time_squeeze가 지속적으로 증가한다면 다음을 검토해야 합니다:

• netdev_budget 증가 (300 → 600 또는 1200)
• netdev_budget_usecs 증가 (8000 → 16000)
• RSS 큐 수 증가로 CPU당 처리량 분산
• NIC 인터럽트 코얼레싱 설정 조정 (rx-usecs 증가)

NAPI 공정성(Fairness) 메커니즘

net_rx_action()은 poll_list를 head에서부터 순차 처리합니다. 버짓을 모두 소진한 NAPI는 repoll 큐의 tail에 추가되고, 다음 라운드에서 다시 head부터 처리됩니다. 이 라운드로빈 방식이 공정성을 보장합니다.

버짓(Budget) 관리 핵심 파라미터

GRO(Generic Receive Offload) 통합

GRO의 역할과 성능 효과

GRO는 NAPI poll 경로에서 동일 TCP/IP 플로우에 속하는 여러 패킷을 하나의 큰 패킷으로 병합하는 최적화입니다. 상위 스택(TCP 등)이 처리해야 하는 패킷 수를 줄여 CPU 사용률과 처리량을 개선합니다.

10 Gbps 링크에서 1500B 패킷의 경우 GRO가 8개를 병합하면 상위 스택이 처리하는 패킷 수가 약 8배 감소합니다. 실제 환경에서 GRO를 활성화했을 때:

• TCP 수신 처리량: 약 20~40% 향상 (플로우 수, 패킷 크기에 따라 다름)
• CPU 사용률: 동일 처리량 대비 약 15~30% 감소
• 캐시 효율: 대형 SKB 단위 처리로 캐시 히트율 향상

GRO 내부 콜체인

/* net/core/gro.c — 간략화 */
/* NAPI poll() → napi_gro_receive() → dev_gro_receive() 경로 */

/* 1단계: napi_gro_receive() - 진입점 */
gro_result_t napi_gro_receive(struct napi_struct *napi,
                               struct sk_buff *skb)
{
    gro_result_t ret;

    skb_mark_napi_id(skb, napi);  /* skb에 napi_id 설정 */
    trace_napi_gro_receive_entry(skb);

    skb_gro_reset_offset(skb, 0);

    ret = dev_gro_receive(napi, skb);
    trace_napi_gro_receive_exit(ret);

    return napi_skb_finish(napi, skb, ret);
}

/* 2단계: dev_gro_receive() - 프로토콜별 GRO 핸들러 호출 */
static gro_result_t dev_gro_receive(struct napi_struct *napi,
                                     struct sk_buff *skb)
{
    u32               hash;
    struct gro_list  *gro_list;
    struct sk_buff   *pp = NULL;
    struct sk_buff   *p;
    const struct packet_offload *ptype;
    gro_result_t       ret = GRO_NORMAL;

    /* VLAN 태그, 터널 헤더 처리 */
    if (skb->protocol == htons(ETH_P_8021Q) || ...)
        goto normal;

    /* 프로토콜별 gro_receive 핸들러 검색
       ETH_P_IP  → inet_gro_receive()
       ETH_P_IPV6 → ipv6_gro_receive() */
    ptype = gro_find_receive_by_type(skb->protocol);
    if (!ptype)
        goto normal;

    /* gro_hash 버킷 선택: 4-tuple 해시 기반 */
    hash = skb_get_hash_raw(skb) & (GRO_HASH_BUCKETS - 1);
    gro_list = &napi->gro_hash[hash];

    /* 동일 플로우 검색 후 병합 시도 */
    pp = ptype->callbacks.gro_receive(&gro_list->list, skb);
    ...
}

/* 3단계: inet_gro_receive() → tcp4_gro_receive() 콜체인 */
struct sk_buff *inet_gro_receive(struct list_head *head,
                                  struct sk_buff *skb)
{
    const struct iphdr *iph = skb_gro_network_header(skb);
    const struct net_offload *ops;

    /* IP 헤더 검증: TTL, ToS, checksum 비교 */
    skb_gro_pull(skb, sizeof(*iph));

    /* 전송 계층 GRO로 위임:
       IPPROTO_TCP → tcp4_gro_receive()
       IPPROTO_UDP → udp4_gro_receive() */
    ops = rcu_dereference(inet_offloads[iph->protocol]);
    if (!ops || !ops->callbacks.gro_receive)
        goto out;
    skb = ops->callbacks.gro_receive(head, skb);
    ...
}

GRO 병합 알고리즘 단계별 설명

tcp4_gro_receive()에서 실제 TCP 세그먼트 병합이 이루어집니다. 아래는 병합 알고리즘의 핵심 단계입니다:

/* net/ipv4/tcp_offload.c — 간략화 */
/* 의사코드: GRO 병합 알고리즘 핵심 로직 */
gro_result_t tcp4_gro_receive(struct list_head *head,
                                struct sk_buff *skb)
{
    struct tcphdr *th = tcp_gro_pull_header(skb);
    struct sk_buff *p;

    list_for_each_entry(p, head, list) {
        struct tcphdr *th2 = tcp_hdr(p);

        /* 단계 1: 동일 플로우 검사 (src/dst port 동일 여부) */
        if (th->source != th2->source || th->dest != th2->dest) {
            NAPI_GRO_CB(p)->same_flow = 0;
            continue;
        }

        /* 단계 2: TCP 플래그 검사 (SYN, FIN, RST, URG 등 거부) */
        if (th->fin || th->syn || th->rst || th->urg) {
            NAPI_GRO_CB(p)->flush = 1;
            continue;
        }

        /* 단계 3: 시퀀스 번호 연속성 검사 */
        if (!tcp_gro_seq_check(p, skb, th)) {
            /* 순서 역전: flush 표시 */
            NAPI_GRO_CB(p)->flush = 1;
            continue;
        }

        /* 단계 4: 크기 검사 (병합 결과가 GRO_MAX_HEAD 이하인지) */
        if (skb_gro_len(skb) > GRO_MAX_HEAD) {
            NAPI_GRO_CB(p)->flush = 1;
            break;
        }

        /* 단계 5: 실제 병합: skb 데이터를 p의 frag_list에 추가 */
        NAPI_GRO_CB(p)->count++;
        NAPI_GRO_CB(p)->tot_len += skb_gro_len(skb);
        tcp_gro_merge(p, skb, th);  /* skb → p의 frag_list 연결 */
        return GRO_MERGED_FREE;
    }

    /* 병합 실패: 새 GRO 헤드로 등록 */
    NAPI_GRO_CB(skb)->count = 1;
    NAPI_GRO_CB(skb)->tot_len = skb_gro_len(skb);
    list_add(&skb->list, head);
    return GRO_HELD;
}

터널(Tunnel) GRO: VXLAN/GRE 중첩 처리

VXLAN이나 GRE 터널 패킷은 중첩 헤더 구조를 가집니다. GRO는 외부 헤더와 내부 헤더를 모두 검사하여 병합 가능 여부를 판단합니다.

/* drivers/net/vxlan/vxlan_core.c — 간략화 */
/* VXLAN GRO: 외부 UDP/IP + 내부 Ethernet/IP/TCP 모두 동일해야 병합 */
/* 내부 패킷의 gro_receive 핸들러도 재귀적으로 호출됨 */
struct sk_buff *vxlan_gro_receive(struct list_head *head,
                                   struct sk_buff *skb,
                                   struct udphdr *uh)
{
    /* 외부 VXLAN VNI 동일 여부 검사 */
    struct vxlanhdr *vh = skb_gro_header_fast(skb, off);
    if (vh->vx_vni != NAPI_GRO_CB(p_skb)->vx_vni)
        continue;

    /* 내부 헤더(Ethernet → IP → TCP)에 대해 재귀적 GRO 처리 */
    skb_gro_pull(skb, sizeof(*vh));
    NAPI_GRO_CB(skb)->encap_mark = 1;
    pp = eth_gro_receive(&gro_head, skb);
    ...
}

GRO 병합 다이어그램

GRO 관련 주요 API

/* net/core/gro.c, include/linux/netdevice.h — 간략화 */
/* GRO 수신 함수: napi poll()에서 직접 호출 */
gro_result_t napi_gro_receive(struct napi_struct *napi,
                               struct sk_buff *skb);

/* Frags(page 기반) GRO 수신: page_pool과 함께 사용 */
gro_result_t napi_gro_frags(struct napi_struct *napi);

/* GRO 결과 코드 */
enum gro_result {
    GRO_MERGED,         /* 기존 GRO 패킷에 병합 완료 */
    GRO_MERGED_FREE,    /* 병합 완료, 원본 skb는 해제 */
    GRO_HELD,           /* GRO 버퍼에 보관 중 (flush 대기) */
    GRO_NORMAL,         /* GRO 미적용, 일반 처리 경로 */
    GRO_CONSUMED,       /* 패킷 소비됨 (드롭 아님) */
};

/* napi poll() 내 GRO 사용 예 */
static int mynic_poll(struct napi_struct *napi, int budget)
{
    struct mynic_rx_ring *ring = container_of(napi, struct mynic_rx_ring, napi);
    int work_done = 0;

    while (work_done < budget) {
        struct sk_buff *skb = mynic_get_next_skb(ring);
        if (!skb)
            break;
        skb->ip_summed = CHECKSUM_UNNECESSARY;
        napi_gro_receive(napi, skb);
        work_done++;
    }

    if (work_done < budget) {
        if (napi_complete_done(napi, work_done))
            mynic_enable_rx_irq(ring);
    }
    return work_done;
}

GRO flush 타이밍과 gro_flush_timeout

GRO 버퍼는 다음 상황에서 즉시 flush됩니다:

• 순서 역전 패킷 도착 시 (동일 플로우)
• 병합 불가 패킷 도착 시 (헤더 불일치, TCP 플래그 차이)
• napi_complete_done() 호출 시 (폴링 종료)
• gro_flush_timeout 타이머 만료 시 (napi_gro_flush())
• 버킷 내 SKB 수가 MAX_GRO_SKBS(8) 초과 시

napi_gro_frags() 상세 — 프래그먼트 기반 GRO 경로

napi_gro_frags()는 드라이버가 헤더와 페이로드(Payload)를 별도 프래그먼트로 분리하여 전달할 때 사용합니다. napi_gro_receive()와 달리 SKB의 linear 영역에는 L2 헤더만 존재하고, 나머지 데이터는 skb_shinfo(skb)->frags[]에 page 프래그먼트로 저장됩니다.

/* net/core/gro.c: napi_gro_frags() 콜체인 */

/* 1단계: napi->skb 캐시에서 SKB 획득 */
struct sk_buff *napi_get_frags(struct napi_struct *napi)
{
    struct sk_buff *skb = napi->skb;
    if (!skb) {
        skb = napi_alloc_skb(napi, GRO_MAX_HEAD);
        napi->skb = skb;
    }
    return skb;
}

/* 2단계: 프래그먼트 GRO 수신 */
gro_result_t napi_gro_frags(struct napi_struct *napi)
{
    struct sk_buff *skb = napi->skb;
    /* skb->data에는 이더넷 헤더만 존재
       skb_shinfo(skb)->frags[]에 IP+TCP+페이로드 page 매핑 */

    gro_normal_one(napi, skb, 1);
    /* → dev_gro_receive() → 프로토콜별 GRO 콜백 체인 */
}

/* napi_gro_receive()와 napi_gro_frags() 선택 기준:
   - napi_gro_receive(): 드라이버가 완전한 skb를 생성한 경우
   - napi_gro_frags():   헤더만 linear, 페이로드는 frag로 전달하는 경우
                         (e1000e, ixgbe, ice 등 대부분의 고성능 드라이버) */

skb_gro_receive() — GRO 패킷 병합 핵심

skb_gro_receive()는 GRO 엔진이 동일 플로우의 패킷을 실제로 하나로 합치는 핵심 함수입니다. 병합 방식은 두 가지가 있으며, 커널이 상황에 따라 자동으로 선택합니다.

/* net/core/skbuff.c: skb_gro_receive() 내부 병합 전략 */
int skb_gro_receive(struct sk_buff *p, struct sk_buff *skb)
{
    unsigned int headlen = skb_headlen(skb);

    /* 전략 1: frag_list 병합
       p->frag_list에 skb를 연결 리스트로 추가.
       단순하지만 이후 TCP coalescing에서 비효율적.
       주로 비표준 프레임이나 frag 공간 부족 시 사용 */
    if (skb_shinfo(p)->frag_list)
        NAPI_GRO_CB(p)->last->next = skb;
    else
        skb_shinfo(p)->frag_list = skb;

    /* 전략 2: frags[] 병합 (선호)
       skb의 데이터를 p->frags[]에 page 프래그먼트로 추가.
       단일 SKB에 모든 데이터가 포함되어 효율적.
       MAX_SKB_FRAGS(17) 제한 내에서만 가능 */
    if (skb_shinfo(p)->nr_frags + delta <= MAX_SKB_FRAGS) {
        skb_frag_list_init(skb);
        /* page 참조를 p->frags[]로 이동 */
    }

    p->len      += skb->len;
    p->data_len += skb->len;
    p->truesize += skb->truesize;
    NAPI_GRO_CB(p)->count++;

    /* MAX_GRO_SKBS(8) 초과 시 flush 트리거 */
    return 0;
}

napi_gro_list_prepare() — 해시 기반 플로우 매칭

GRO 엔진은 수신 패킷을 기존 GRO 리스트와 비교하여 동일 플로우를 찾습니다. napi_gro_list_prepare()는 해시 버킷 내의 모든 대기 중인 SKB를 순회하며 same_flow와 flush 플래그를 설정합니다.

/* net/core/gro.c: GRO 플로우 매칭 */
static void napi_gro_list_prepare(
    const struct napi_struct *napi,
    const struct sk_buff *skb)
{
    struct sk_buff *p;
    unsigned long diffs;

    /* napi->gro_hash[bucket] 리스트를 순회 */
    list_for_each_entry(p, head, list) {
        diffs = (unsigned long)p->dev ^ (unsigned long)skb->dev;
        diffs |= skb_vlan_tag_present(p) ^
                 skb_vlan_tag_present(skb);

        /* MAC 헤더 비교 (EtherType, VLAN 등) */
        diffs |= compare_ether_header(
            skb_mac_header(p), skb_mac_header(skb));

        NAPI_GRO_CB(p)->same_flow = !diffs;
        NAPI_GRO_CB(p)->flush     = 0;
        /* 이후 프로토콜 콜백에서 flush 여부를 정밀 판단 */
    }
}

프로토콜별 GRO 콜백(Callback) 테이블

GRO는 계층별 콜백 함수를 체인으로 호출하여 프로토콜 헤더를 검증하고 병합 가능 여부를 판단합니다. 각 프로토콜은 struct net_offload 또는 struct packet_offload에 gro_receive/gro_complete 콜백을 등록합니다.

NAPI 메모리 및 버퍼 관리

NAPI 메모리 할당 개요

NAPI 폴링 컨텍스트는 softIRQ나 스레드 NAPI 내에서 실행되며, 일반 메모리 할당과 다른 전용 캐시 메커니즘을 사용합니다. 커널은 NAPI 전용 할당 API를 통해 per-CPU 캐시(napi_alloc_cache)를 활용하여 할당 오버헤드를 최소화합니다.

napi_alloc_skb() / __napi_alloc_skb()

NAPI 폴링 컨텍스트 전용 SKB 할당 함수입니다. per-CPU napi_alloc_cache를 통해 slab 할당자의 lock contention을 회피하고, GFP_ATOMIC 없이도 빠르게 할당합니다.

/* include/linux/skbuff.h */
struct sk_buff *napi_alloc_skb(
    struct napi_struct *napi,
    unsigned int length);  /* 헤더 영역 크기 */

struct sk_buff *__napi_alloc_skb(
    struct napi_struct *napi,
    unsigned int length,
    gfp_t gfp_mask);

/* 내부 구현 핵심 (net/core/skbuff.c):
   1. per-CPU napi_alloc_cache에서 skb 구조체 획득 (slab bypass)
   2. 페이지 프래그먼트에서 데이터 영역 할당
   3. skb->head, skb->data, skb->tail 초기화
   4. napi->skb_cache_lock 없이 lockless 동작 */

/* 사용 예 (드라이버 poll 함수 내부) */
struct sk_buff *skb = napi_alloc_skb(napi, 256);
if (!skb)
    return -ENOMEM;
/* skb->data에 256바이트 linear 영역 확보
   나머지 페이로드는 frags[]로 매핑 가능 */

napi_build_skb() / __napi_build_skb()

napi_build_skb()는 이미 할당된 버퍼(페이지)를 기반으로 SKB를 생성합니다. 데이터 복사 없이 SKB 메타데이터만 초기화하므로 제로카피 수신 경로의 핵심입니다. napi_alloc_skb()와 달리 데이터 영역을 별도로 할당하지 않습니다.

/* include/linux/skbuff.h */
struct sk_buff *napi_build_skb(
    void *data,              /* 이미 할당된 버퍼 포인터 */
    unsigned int frag_size); /* 버퍼 전체 크기 */

/* napi_alloc_skb() vs napi_build_skb() 비교:
 *
 * napi_alloc_skb(napi, 256):
 *   - SKB 구조체 할당 + 256바이트 데이터 영역 할당
 *   - DMA 버퍼 → memcpy → SKB 데이터 영역
 *   - 소형 패킷이나 레거시 드라이버에 적합
 *
 * napi_build_skb(page_addr, PAGE_SIZE):
 *   - SKB 구조체만 할당, data는 이미 존재하는 page를 가리킴
 *   - DMA 버퍼 = SKB 데이터 영역 (제로카피)
 *   - page_pool 기반 고성능 드라이버에 적합
 */

/* page_pool + napi_build_skb 패턴 */
struct page *page = page_pool_dev_alloc_pages(ring->page_pool);
void *va = page_address(page) + offset;

/* DMA에서 직접 이 페이지에 수신 데이터를 기록 */
dma_sync_single_for_cpu(dev, dma_addr, len, DMA_FROM_DEVICE);

struct sk_buff *skb = napi_build_skb(va - headroom, frag_size);
skb_reserve(skb, headroom);
skb_put(skb, len);
skb_mark_for_recycle(skb); /* page_pool 재활용 마킹 */

napi_alloc_frag() / napi_alloc_frag_align()

페이지 프래그먼트(page fragment)는 하나의 물리 페이지를 여러 소형 버퍼로 분할 사용하는 기법입니다. napi_alloc_frag()는 per-CPU napi_alloc_cache.page에서 요청 크기만큼의 프래그먼트를 슬라이싱하여 반환합니다.

/* include/linux/skbuff.h */
void *napi_alloc_frag(unsigned int fragsz);

void *napi_alloc_frag_align(
    unsigned int fragsz,
    unsigned int align);  /* 정렬 요구사항 (예: L1_CACHE_BYTES) */

/* 내부 동작:
   1. per-CPU napi_alloc_cache.page에서 남은 공간 확인
   2. fragsz만큼 슬라이싱 (offset 증가)
   3. 페이지 소진 시 새 compound page 할당
   4. refcount로 프래그먼트 수명 관리 */

/* 사용 예: 헤더 영역만 별도 할당 */
void *header = napi_alloc_frag_align(256, SMP_CACHE_BYTES);
if (!header)
    return -ENOMEM;
/* 이 영역에 패킷 헤더를 복사, 페이로드는 page_pool page를 frags[]로 연결 */

napi_get_frags() / napi_reuse_skb()

napi_get_frags()는 GRO 프래그먼트 경로에서 사용하는 per-NAPI SKB 캐시입니다. 각 napi_struct는 napi->skb 필드에 하나의 재사용 가능 SKB를 보관합니다.

/* net/core/gro.c */
struct sk_buff *napi_get_frags(struct napi_struct *napi)
{
    struct sk_buff *skb = napi->skb;
    if (!skb) {
        skb = napi_alloc_skb(napi, GRO_MAX_HEAD);
        if (skb)
            napi->skb = skb;
    }
    return skb;
}

/* GRO 병합 성공 후 SKB 재활용 */
static void napi_reuse_skb(struct napi_struct *napi,
                            struct sk_buff *skb)
{
    if (unlikely(skb->pfmemalloc)) {
        consume_skb(skb);
        return;
    }
    __skb_pull(skb, skb_headlen(skb));
    skb_reserve(skb, NET_IP_ALIGN - skb_headroom(skb));
    __vlan_hwaccel_clear_tag(skb);
    skb->dev = napi->dev;
    napi->skb = skb;  /* 다음 napi_get_frags()에서 재사용 */
}

/* GRO 결과에 따른 경로:
   GRO_MERGED      → napi_reuse_skb(): SKB 재활용
   GRO_MERGED_FREE → napi_skb_free(): SKB 해제 + frag 해제
   GRO_NORMAL      → napi_skb_finish(): netif_receive_skb()로 전달
   GRO_HELD        → GRO 리스트에 보관 (flush 대기) */

napi_consume_skb() — budget 인식 SKB 해제

napi_consume_skb()는 NAPI 컨텍스트에서 SKB를 해제하는 최적화된 함수입니다. TX 완료 경로에서 주로 사용되며, budget 인자를 통해 NAPI poll과 non-NAPI 컨텍스트를 자동으로 구분합니다.

/* net/core/skbuff.c */
void napi_consume_skb(struct sk_buff *skb, int budget)
{
    if (unlikely(!skb))
        return;

    /* budget > 0: NAPI poll 컨텍스트
       → per-CPU napi_alloc_cache로 SKB 반환 (bulk free)
       budget == 0: 비-NAPI 컨텍스트 (예: 타이머, netpoll)
       → 일반 kfree_skb_reason() 경로 */
    if (budget) {
        napi_skb_cache_put(skb); /* lockless 캐시 반환 */
    } else {
        kfree_skb_reason(skb, SKB_DROP_REASON_NOT_SPECIFIED);
    }
}

/* TX 완료 처리에서의 사용 패턴 */
static bool mynic_clean_tx(struct mynic_tx_ring *ring, int budget)
{
    while (tx_cleaned < budget) {
        struct sk_buff *skb = ring->tx_buf[idx].skb;
        dma_unmap_single(dev, dma, len, DMA_TO_DEVICE);
        napi_consume_skb(skb, budget); /* budget 전달! */
        ring->tx_buf[idx].skb = NULL;
        tx_cleaned++;
    }
}

page_pool 통합

page_pool은 NAPI 전용 고성능 페이지 할당/재활용 프레임워크입니다. DMA 매핑(Mapping)을 캐싱하고, 페이지를 재활용하여 메모리 할당 오버헤드와 IOMMU/SWIOTLB 비용을 획기적으로 줄입니다.

/* include/net/page_pool/types.h */
struct page_pool_params {
    int           order;      /* 페이지 order (0=4K, 1=8K) */
    unsigned int  flags;      /* PP_FLAG_DMA_MAP | PP_FLAG_DMA_SYNC_DEV */
    int           pool_size;  /* 초기 풀 크기 (디스크립터 수 권장) */
    int           nid;        /* NUMA 노드 */
    struct device *dev;       /* DMA 매핑용 디바이스 */
    struct napi_struct *napi; /* 연결된 NAPI 인스턴스 */
    enum dma_data_direction dma_dir;
    unsigned int  offset;     /* NET_SKB_PAD + NET_IP_ALIGN */
    unsigned int  max_len;    /* 최대 데이터 길이 */
};

/* page_pool 핵심 API 체인 */
struct page_pool *page_pool_create(
    const struct page_pool_params *params);

/* 할당: 캐시 → 링 → buddy allocator 순서 */
struct page *page_pool_dev_alloc_pages(
    struct page_pool *pool);

/* DMA 주소 획득 (이미 매핑됨, IOMMU 비용 제로) */
dma_addr_t page_pool_get_dma_addr(struct page *page);

/* 직접 재활용: NAPI poll 내에서 즉시 풀로 반환 */
void page_pool_recycle_direct(
    struct page_pool *pool, struct page *page);

/* SKB에 page_pool 재활용 마킹 (네트워크 스택 통과 후 자동 재활용) */
void skb_mark_for_recycle(struct sk_buff *skb);

/* 풀 해제 */
void page_pool_destroy(struct page_pool *pool);

NAPI 메모리 API 종합 비교표

NAPI 메모리 함수 내부 구현 상세

NAPI 컨텍스트에서 사용되는 메모리 할당/해제 함수들은 일반 커널 할당자와 달리 per-CPU 캐시(per-CPU Cache)를 적극 활용하여 락 경합(Lock Contention)을 최소화합니다. 이 절에서는 각 함수의 내부 구현을 커널 소스 수준에서 분석합니다.

napi_alloc_skb() / __napi_alloc_skb() 내부

napi_alloc_skb()는 NAPI poll 컨텍스트 전용 skb 할당 함수로, per-CPU napi_alloc_cache 구조체를 통해 skb 헤더를 사전 할당(Pre-allocation)하여 재사용합니다. 내부적으로 __napi_alloc_skb()를 호출합니다.

/* 커널 소스 분석: net/core/skbuff.c — per-CPU NAPI 할당 캐시 */

#define NAPI_SKB_CACHE_SIZE    64
#define NAPI_SKB_CACHE_BULK    16

struct napi_alloc_cache {
    struct page_frag_cache page;        /* 페이지 프래그먼트 캐시 */
    unsigned int           skb_count;    /* 캐시된 skb 헤더 개수 */
    void                   *skb_cache[NAPI_SKB_CACHE_SIZE]; /* skb 헤더 풀 */
};

static DEFINE_PER_CPU(struct napi_alloc_cache, napi_alloc_cache);

/* 커널 소스 분석: __napi_alloc_skb() 핵심 구현 흐름 */
struct sk_buff *__napi_alloc_skb(struct napi_struct *napi,
                                unsigned int len, gfp_t gfp_mask)
{
    struct napi_alloc_cache *nc;
    struct sk_buff *skb;
    void *data;
    bool pfmemalloc;

    /* 1단계: 데이터 길이 정렬 (SMP_CACHE_BYTES 단위) */
    len += NET_SKB_PAD + NET_IP_ALIGN;
    len = SKB_DATA_ALIGN(len);
    len += SKB_DATA_ALIGN(sizeof(struct skb_shared_info));

    nc = this_cpu_ptr(&napi_alloc_cache);

    /* 2단계: per-CPU 페이지 프래그먼트에서 데이터 영역 할당
     *        slab 할당자를 거치지 않고 페이지 조각을 직접 잘라 사용 */
    data = page_frag_alloc_align(&nc->page, len, gfp_mask,
                                  SMP_CACHE_BYTES);
    if (unlikely(!data))
        return NULL;

    pfmemalloc = page_frag_cache_is_pfmemalloc(&nc->page);

    /* 3단계: napi_alloc_cache에서 skb 헤더 가져오기
     *        캐시가 비었으면 kmem_cache_alloc_bulk()로 일괄 보충 */
    if (unlikely(!nc->skb_count)) {
        nc->skb_count = kmem_cache_alloc_bulk(
            skbuff_cache, gfp_mask,
            NAPI_SKB_CACHE_BULK,
            nc->skb_cache);
        if (unlikely(!nc->skb_count))
            return NULL;
    }
    skb = nc->skb_cache[--nc->skb_count];

    /* 4단계: skb 헤더와 데이터 프래그먼트 연결 */
    __build_skb_around(skb, data, len);
    skb->pfmemalloc = pfmemalloc;

    /* 5단계: 네트워크 정렬을 위한 headroom 예약
     *        NET_SKB_PAD: DMA 정렬, NET_IP_ALIGN: IP 헤더 정렬 */
    skb_reserve(skb, NET_SKB_PAD + NET_IP_ALIGN);
    skb->dev = napi->dev;

    return skb;
}

skb 할당 함수 비교

napi_build_skb() 내부

napi_build_skb()는 드라이버가 이미 할당한 데이터 버퍼(page_pool 또는 드라이버 링 버퍼)에 skb 헤더만 감싸는 함수입니다. napi_alloc_skb()와 달리 데이터 영역을 새로 할당하지 않으므로, DMA로 직접 채워진 페이지를 즉시 skb로 변환할 수 있습니다.

/* 커널 소스 분석: net/core/skbuff.c — napi_build_skb() 구현 */

struct sk_buff *napi_build_skb(void *data, unsigned int frag_size)
{
    struct napi_alloc_cache *nc;
    struct sk_buff *skb;

    nc = this_cpu_ptr(&napi_alloc_cache);

    /* 1단계: napi_alloc_cache에서 skb 헤더만 가져오기
     *        데이터 버퍼는 이미 존재하므로 헤더만 필요 */
    if (unlikely(!nc->skb_count)) {
        nc->skb_count = kmem_cache_alloc_bulk(
            skbuff_cache, GFP_ATOMIC,
            NAPI_SKB_CACHE_BULK,
            nc->skb_cache);
        if (unlikely(!nc->skb_count))
            return NULL;
    }
    skb = nc->skb_cache[--nc->skb_count];

    /* 2단계: skb 초기화 후 기존 데이터 버퍼에 연결
     *        head, data, tail, end 포인터를 data 기반으로 설정 */
    memset(skb, 0, offsetof(struct sk_buff, tail));

    /* __build_skb_around: skb->head = data
     *                      skb->data = data
     *                      skb->tail = 0
     *                      skb->end  = frag_size - skb_shared_info 크기 */
    __build_skb_around(skb, data, frag_size);

    return skb;
}

/* 커널 소스 분석: __build_skb_around() — skb를 기존 데이터에 연결 */
static void __build_skb_around(struct sk_buff *skb,
                               void *data, unsigned int frag_size)
{
    struct skb_shared_info *shinfo;
    unsigned int size = frag_size;

    size -= SKB_DATA_ALIGN(sizeof(struct skb_shared_info));

    /* skb 포인터 설정: head/data → 버퍼 시작, end → 유효 영역 끝 */
    skb->head     = data;
    skb->data     = data;
    skb_reset_tail_pointer(skb);
    skb->end      = size;
    skb->mac_header = (__u16)~0U;
    skb->transport_header = (__u16)~0U;

    /* skb_shared_info 초기화 (데이터 끝 부분에 위치) */
    shinfo = skb_shinfo(skb);
    memset(shinfo, 0, offsetof(struct skb_shared_info, dataref));
    atomic_set(&shinfo->dataref, 1);

    skb->head_frag = 1;
    skb->truesize = SKB_TRUESIZE(size);
    refcount_set(&skb->users, 1);
}

napi_consume_skb() 내부

napi_consume_skb()는 NAPI 컨텍스트에서 skb를 해제하는 최적화된 함수입니다. 즉시 해제하지 않고 skb 헤더를 napi_alloc_cache에 반환하여 재사용합니다.

/* 커널 소스 분석: net/core/skbuff.c — napi_consume_skb() 구현 */

void napi_consume_skb(struct sk_buff *skb, int budget)
{
    /* NULL skb 안전 처리 */
    if (unlikely(!skb))
        return;

    /* 1단계: budget == 0이면 poll 외부 → 즉시 해제
     *        napi_complete_done() 또는 NAPI 외부에서 호출된 경우 */
    if (unlikely(!budget)) {
        dev_kfree_skb_any(skb);
        return;
    }

    /* 2단계: 참조 카운트 감소 — 다른 곳에서 참조 중이면 반환 */
    if (likely(refcount_read(&skb->users) == 1))
        smp_rmb();
    else if (likely(!refcount_dec_and_test(&skb->users)))
        return;

    /* 3단계: skb 데이터 영역 해제 (frag, frag_list 등) */
    skb_release_data(skb, SKB_CONSUMED);

    /* 4단계: skb 헤더를 napi_alloc_cache에 반환하여 재사용
     *        캐시가 가득 차면 kmem_cache_free_bulk()로 일괄 해제 */
    struct napi_alloc_cache *nc = this_cpu_ptr(&napi_alloc_cache);

    if (nc->skb_count == NAPI_SKB_CACHE_SIZE) {
        /* 캐시 만료: 절반을 slab으로 일괄 반환 */
        kmem_cache_free_bulk(skbuff_cache,
                            NAPI_SKB_CACHE_BULK,
                            nc->skb_cache);
        nc->skb_count -= NAPI_SKB_CACHE_BULK;
    }
    nc->skb_cache[nc->skb_count++] = skb;
}

napi_alloc_frag() 내부

napi_alloc_frag()는 per-CPU 페이지 프래그먼트 할당자(Page Fragment Allocator)를 사용하여 하나의 페이지를 여러 개의 작은 조각으로 나누어 할당합니다. 작은 패킷에 특히 효율적입니다.

/* 커널 소스 분석: net/core/skbuff.c — napi_alloc_frag() 구현 */

void *napi_alloc_frag(unsigned int fragsz)
{
    return napi_alloc_frag_align(fragsz, ~0u);
}

void *napi_alloc_frag_align(unsigned int fragsz, unsigned int align)
{
    struct napi_alloc_cache *nc = this_cpu_ptr(&napi_alloc_cache);

    /* page_frag_alloc_align: per-CPU 페이지 프래그먼트 캐시에서 할당
     *
     * 내부 동작:
     *   1. nc->page.va (현재 페이지)에 남은 공간이 있는지 확인
     *   2. 충분하면: offset 계산 후 va + offset 반환, offset 갱신
     *   3. 부족하면: __page_frag_cache_refill()로 새 페이지 할당
     *      - order-3 (32KB) 페이지 시도 → 실패 시 order-0 (4KB) 폴백
     *      - 새 페이지의 refcount를 높게 설정 (조각 수만큼)
     */
    return page_frag_alloc_align(&nc->page, fragsz,
                                  GFP_ATOMIC, align);
}

/* 커널 소스 분석: mm/page_frag_cache.c — 서브페이지 할당 핵심 */
void *page_frag_alloc_align(struct page_frag_cache *nc,
                           unsigned int fragsz,
                           gfp_t gfp_mask,
                           unsigned int align_mask)
{
    unsigned int size, offset;
    struct page *page;

    if (unlikely(!nc->va)) {
refill:
        /* 새 페이지 할당 (order-3 시도 → order-0 폴백) */
        page = __page_frag_cache_refill(nc, gfp_mask);
        if (!page)
            return NULL;
    }

    size = nc->size;
    offset = nc->offset;

    /* fragsz만큼 잘라서 반환 — 페이지 끝에서 시작으로 할당 */
    if (unlikely(offset < fragsz)) {
        /* 남은 공간 부족 → 새 페이지 할당 */
        page_frag_cache_drain(nc);
        goto refill;
    }

    offset = (offset - fragsz) & align_mask;
    nc->offset = offset;

    return nc->va + offset;
}

napi_get_frags() 내부

napi_get_frags()는 GRO 프래그먼트 경로에서 사용하는 캐시된 skb를 반환합니다. napi->skb에 부분적으로 구성된 skb를 유지하면서, 드라이버가 페이지 프래그먼트로 전달하는 RX 데이터를 누적합니다.

/* 커널 소스 분석: net/core/gro.c — napi_get_frags() 구현 */

struct sk_buff *napi_get_frags(struct napi_struct *napi)
{
    struct sk_buff *skb = napi->skb;

    /* 캐시된 skb가 있으면 그대로 반환 */
    if (!skb) {
        /* 없으면 새로 할당 — GRO_MAX_HEAD 크기의 선형 헤더 공간 확보
         * GRO_MAX_HEAD = MAX_HEADER + 128 (L2/L3/L4 헤더용) */
        skb = napi_alloc_skb(napi, GRO_MAX_HEAD);
        if (skb)
            napi->skb = skb;
    }
    return skb;
}

/* 드라이버 사용 패턴 예시 (frag 기반 RX) */
static void driver_rx_frag_path(struct napi_struct *napi,
                                 struct page *page,
                                 unsigned int offset,
                                 unsigned int len)
{
    struct sk_buff *skb = napi_get_frags(napi);

    if (unlikely(!skb))
        return;

    /* 헤더 데이터를 skb 선형 영역에 복사 */
    skb_fill_page_desc(skb, skb_shinfo(skb)->nr_frags,
                       page, offset, len);
    skb->len += len;
    skb->data_len += len;
    skb->truesize += len;

    /* GRO frag 경로로 전달 — napi->skb 캐시 자동 소비 */
    napi_gro_frags(napi);
}

GRO 핵심 함수 내부 구현

GRO(Generic Receive Offload)는 수신된 작은 패킷들을 상위 스택 전달 전에 하나의 큰 패킷으로 병합하여 프로토콜 처리 오버헤드를 줄이는 메커니즘입니다. 이 절에서는 GRO의 핵심 함수들을 커널 소스 수준에서 분석합니다.

napi_gro_receive() 내부

napi_gro_receive()는 수신 패킷을 GRO 엔진에 전달하는 주요 진입점입니다. 내부적으로 dev_gro_receive()를 호출하여 패킷 병합을 시도합니다.

/* 커널 소스 분석: net/core/gro.c — napi_gro_receive() 전체 호출 체인 */

gro_result_t napi_gro_receive(struct napi_struct *napi,
                              struct sk_buff *skb)
{
    /* skb 타임스탬프 기록 (GRO 병합 판정에 사용) */
    skb_gro_reset_offset(skb, 0);

    return napi_skb_finish(napi, skb,
                           dev_gro_receive(napi, skb));
}

/* dev_gro_receive: GRO 엔진 핵심 */
static enum gro_result dev_gro_receive(struct napi_struct *napi,
                                        struct sk_buff *skb)
{
    struct list_head *head = &offload_base;
    struct packet_offload *ptype;
    struct sk_buff *pp = NULL;
    enum gro_result ret;
    __be16 type = skb->protocol;

    /* ── 1단계: gro_list_prepare() ──
     * gro_hash 버킷에서 same_flow 후보 탐색
     * 5-tuple {src_ip, dst_ip, src_port, dst_port, protocol} 비교 */
    gro_list_prepare(&napi->gro_hash[gro_hash_bucket(skb)], skb);

    /* rcu 보호 하에 등록된 프로토콜 오프로드 콜백 탐색 */
    rcu_read_lock();
    list_for_each_entry_rcu(ptype, head, list) {
        if (ptype->type != type || !ptype->callbacks.gro_receive)
            continue;

        /* ── 2단계: 프로토콜 오프로드 콜백 호출 ──
         * L3: inet_gro_receive() → L4: tcp4_gro_receive()
         * 콜백은 병합 가능한 기존 skb 포인터(pp)를 반환하거나
         * NULL을 반환 (병합 불가 또는 새 항목 추가) */
        skb->dev = napi->dev;
        skb_set_network_header(skb, skb_gro_offset(skb));
        pp = call_gro_receive(ptype->callbacks.gro_receive,
                              &napi->gro_hash[gro_hash_bucket(skb)],
                              skb);
        break;
    }
    rcu_read_unlock();

    /* ── 3단계: 결과 처리 ── */
    if (&ptype->list == head)
        goto normal;   /* 해당 프로토콜 오프로드 없음 */

    if (PTR_ERR(pp) == -EINPROGRESS) {
        ret = GRO_CONSUMED;
        goto ok;
    }

    if (same_flow)
        goto ok;       /* GRO_MERGED 또는 GRO_MERGED_FREE */

    if (NAPI_GRO_CB(skb)->flush)
        goto normal;

    /* 새 GRO 항목으로 gro_hash 버킷에 추가 */
    gro_list_add(&napi->gro_hash[gro_hash_bucket(skb)], skb);
    ret = GRO_HELD;
    goto ok;

normal:
    ret = GRO_NORMAL;
ok:
    return ret;
}

/* napi_skb_finish: GRO 결과에 따른 후처리 */
static gro_result_t napi_skb_finish(struct napi_struct *napi,
                                     struct sk_buff *skb,
                                     enum gro_result ret)
{
    switch (ret) {
    case GRO_NORMAL:
        /* GRO 불가 → 즉시 네트워크 스택으로 전달 */
        gro_normal_one(napi, skb, 1);
        break;

    case GRO_MERGED_FREE:
        /* 병합 완료 + 원본 skb 해제 */
        napi_consume_skb(skb, 1);
        break;

    case GRO_HELD:
    case GRO_MERGED:
    case GRO_CONSUMED:
        /* 병합됨/보류/소비됨 → 추가 작업 불필요 */
        break;
    }

    return ret;
}

gro_list_prepare() 내부

gro_list_prepare()는 수신 패킷의 5-tuple(소스 IP, 목적지 IP, 소스 포트, 목적지 포트, 프로토콜)을 해시하여 gro_hash 버킷을 선택하고, 기존 GRO 항목과 same_flow 비교를 수행합니다.

/* 커널 소스 분석: net/core/gro.c — gro_list_prepare() */

static void gro_list_prepare(struct gro_list *gro_list,
                             const struct sk_buff *skb)
{
    struct sk_buff *p;
    unsigned int maclen = skb->dev->hard_header_len;
    u32 hash = skb_get_hash_raw(skb);

    list_for_each_entry(p, &gro_list->list, list) {
        unsigned long diffs;

        NAPI_GRO_CB(p)->flush = 0;
        NAPI_GRO_CB(p)->same_flow = 0;

        if (unlikely(p->dev != skb->dev))
            continue;

        /* 1차 필터: 해시값 비교 (빠른 불일치 검출) */
        if (skb_get_hash_raw(p) != hash)
            continue;

        /* 2차 필터: MAC 헤더 바이트 단위 비교 */
        diffs = (unsigned long)p->dev ^ (unsigned long)skb->dev;
        diffs |= skb_vlan_tag_present(p) ^ skb_vlan_tag_present(skb);
        diffs |= compare_ether_header(
                    skb_mac_header(p),
                    skb_mac_header(skb));

        /* same_flow 판정: 모든 차이가 0이면 동일 플로우 */
        NAPI_GRO_CB(p)->same_flow = !diffs;
    }
}

GRO 병합 조건

napi_gro_flush() 내부

napi_gro_flush()는 gro_hash 버킷에 보류 중인 모든 GRO 패킷을 네트워크 스택으로 전달(flush)합니다. 주로 napi_complete_done()에서 poll 종료 시, 또는 gro_flush_timeout 타이머 만료 시 호출됩니다.

/* 커널 소스 분석: net/core/gro.c — napi_gro_flush() 구현 */

void napi_gro_flush(struct napi_struct *napi, bool flush_old)
{
    struct sk_buff *skb, *p;
    unsigned long bitmask = napi->gro_bitmask;
    unsigned int i, base = ~0U;

    /* 각 gro_hash 버킷을 순회 */
    while ((i = ffs(bitmask)) != 0) {
        bitmask >>= i;
        base += i;

        list_for_each_entry_safe(skb, p,
                &napi->gro_hash[base].list, list) {

            /* flush_old == true: 오래된 패킷만 flush
             * NAPI_GRO_CB(skb)->age가 jiffies보다
             * gro_flush_timeout 이상 오래되었는지 확인 */
            if (flush_old &&
                NAPI_GRO_CB(skb)->age == jiffies)
                continue;

            /* GRO complete 콜백 호출:
             * L4 gro_complete → L3 gro_complete
             * TCP 체크섬 최종화, 헤더 정리 등 */
            skb_list_del_init(skb);
            napi_gro_complete(napi, skb);
            napi->gro_hash[base].count--;
        }

        /* 버킷이 비었으면 bitmask에서 해당 비트 해제 */
        if (list_empty(&napi->gro_hash[base].list))
            napi->gro_bitmask &= ~(1 << base);
    }
}

/* napi_gro_complete: 개별 GRO 패킷 완성 처리 */
static void napi_gro_complete(struct napi_struct *napi,
                              struct sk_buff *skb)
{
    struct packet_offload *ptype;
    __be16 type = skb->protocol;

    rcu_read_lock();
    list_for_each_entry_rcu(ptype, &offload_base, list) {
        if (ptype->type != type || !ptype->callbacks.gro_complete)
            continue;

        /* 프로토콜별 gro_complete 콜백:
         * inet_gro_complete → tcp4_gro_complete
         * TCP pseudo-header 체크섬 갱신, skb 메타데이터 정리 */
        ptype->callbacks.gro_complete(skb, 0);
        break;
    }
    rcu_read_unlock();

    /* 완성된 GRO 패킷을 gro_normal_list에 추가
     * 배치 전달: gro_normal_batch 개수 도달 시
     * netif_receive_skb_list()로 일괄 전달 */
    gro_normal_one(napi, skb, NAPI_GRO_CB(skb)->count);
}

GRO flush 타이밍 전략

napi_gro_frags() 내부

napi_gro_frags()는 napi_get_frags()로 가져온 캐시 skb에 대해 GRO를 수행합니다. napi_gro_receive()와 유사하지만, 입력이 프래그먼트 리스트(Fragment List) 기반 skb라는 점이 다릅니다.

/* 커널 소스 분석: net/core/gro.c — napi_gro_frags() 구현 */

gro_result_t napi_gro_frags(struct napi_struct *napi)
{
    struct sk_buff *skb;
    gro_result_t ret;

    /* 1단계: napi->skb에서 캐시된 frag skb 가져오기 */
    skb = napi_frags_skb(napi);
    if (!skb)
        return GRO_DROP;

    /* 2단계: dev_gro_receive()로 GRO 엔진에 전달
     *        napi_gro_receive()와 동일한 병합 로직 수행 */
    ret = napi_frags_finish(napi, skb,
                            dev_gro_receive(napi, skb));

    return ret;
}

/* napi_frags_skb: frag skb를 GRO 처리용으로 준비 */
static struct sk_buff *napi_frags_skb(struct napi_struct *napi)
{
    struct sk_buff *skb = napi->skb;

    /* napi->skb 캐시 소비 (NULL로 리셋) */
    napi->skb = NULL;

    /* L2 헤더를 frag에서 선형 영역으로 pull
     * 최소 ETH_HLEN (14바이트)만큼 필요 */
    if (unlikely(!pskb_may_pull(skb, ETH_HLEN))) {
        napi_reuse_skb(napi, skb);
        return NULL;
    }

    /* eth_type_trans: MAC 헤더 파싱, skb->protocol 설정
     * skb->data를 L3 헤더 시작으로 이동 */
    skb->protocol = eth_type_trans(skb, skb->dev);

    /* GRO 오프셋 초기화 */
    skb_gro_reset_offset(skb, skb_headlen(skb));

    return skb;
}

GRO 프로토콜별 오프로드 콜백

GRO 엔진은 packet_offload 구조체를 통해 등록된 프로토콜별 콜백을 호출합니다. 각 프로토콜은 gro_receive(병합 시도)와 gro_complete(완성 처리) 두 개의 콜백을 제공합니다.

/* 커널 소스 분석: include/linux/netdevice.h — packet_offload 구조체 */

struct packet_offload {
    __be16            type;   /* ETH_P_IP, ETH_P_IPV6 등 */
    u16               priority;
    struct offload_callbacks callbacks;
    struct list_head  list;
};

struct offload_callbacks {
    /* 병합 시도: 병합 가능한 기존 skb 포인터 반환 또는 NULL */
    struct sk_buff *(*gro_receive)(
        struct list_head *head,
        struct sk_buff *skb);

    /* 완성 처리: flush 시 호출, 체크섬 최종화 등 */
    int (*gro_complete)(
        struct sk_buff *skb,
        int nhoff);
};

/* 커널 소스 분석: net/ipv4/af_inet.c — IPv4 오프로드 등록 */
static struct packet_offload ip_packet_offload __read_mostly = {
    .type = cpu_to_be16(ETH_P_IP),
    .callbacks = {
        .gro_receive = inet_gro_receive,
        .gro_complete = inet_gro_complete,
    },
};

/* 커널 소스 분석: net/ipv4/af_inet.c — inet_gro_receive() 핵심 흐름 */
struct sk_buff *inet_gro_receive(struct list_head *head,
                                 struct sk_buff *skb)
{
    const struct iphdr *iph;
    struct sk_buff *pp = NULL;
    const struct net_offload *ops;
    int proto;

    iph = skb_gro_header(skb, sizeof(*iph), 0);

    /* IP 헤더 검증: 버전, IHL, 프래그먼트 여부 */
    if (unlikely(iph->version != 4))
        goto out;
    if (unlikely(iph->ihl < 5))
        goto out;
    if (unlikely(iph->frag_off & htons(IP_MF | IP_OFFSET)))
        goto out;

    proto = iph->protocol;

    /* L4 오프로드 콜백 조회: TCP, UDP 등 */
    ops = rcu_dereference(inet_offloads[proto]);
    if (!ops || !ops->callbacks.gro_receive)
        goto out;

    /* same_flow 세부 검증: src/dst IP 비교 */
    list_for_each_entry(p, head, list) {
        const struct iphdr *iph2;
        if (!NAPI_GRO_CB(p)->same_flow)
            continue;
        iph2 = (struct iphdr *)(p->data + 0);
        if (iph->saddr != iph2->saddr ||
            iph->daddr != iph2->daddr) {
            NAPI_GRO_CB(p)->same_flow = 0;
            continue;
        }
    }

    /* L4 콜백 호출: 예) tcp4_gro_receive() */
    pp = call_gro_receive(ops->callbacks.gro_receive, head, skb);

out:
    return pp;
}

GRO 오프로드 콜백 등록 테이블

멀티큐 NAPI와 스케일링

멀티큐 아키텍처

현대 NIC는 수십~수백 개의 하드웨어 RX 큐를 갖춥니다. 각 큐는 독립적인 napi_struct와 IRQ를 할당받아 서로 다른 CPU에서 병렬 처리됩니다. 이 구조가 RSS(Receive Side Scaling)의 기반입니다.

RSS 해시 알고리즘: Toeplitz 해시

RSS는 Toeplitz 해시 함수를 사용하여 패킷을 큐에 분산합니다. 해시 입력은 IP/TCP 4-tuple이며, 하드웨어가 직접 계산합니다.

/* net/core/flow_dissector.c, include/net/flow_keys.h — 간략화 */
/* RSS Toeplitz 해시: 4-tuple (src_ip, dst_ip, src_port, dst_port) 기반 */
/* 128비트 무작위 해시 키(ethtool -x 출력)를 사용하여 큐 번호 결정 */

/* RSS 해시 조회 (소프트웨어 계산 시) */
u32 rss_toeplitz_hash(const u8 *key, u32 keylen,
                         const u8 *data, u32 datalen)
{
    u32 result = 0;
    u32 i, b;
    u32 key_data = 0;

    for (i = 0; i < keylen; i++)
        key_data = (key_data << 8) | key[i];

    for (b = 0; b < datalen * 8; b++) {
        if (data[b / 8] & (0x80 >> (b % 8)))
            result ^= key_data;
        key_data = (key_data << 1) |
                   ((key[(keylen - 1 - b / 8)] >> (b % 8)) & 1);
    }
    return result;
}

/* 큐 번호 결정: 해시값 → indirection table(RETA) 조회 */
/* RETA(Redirection Table): 128~512 엔트리, 각 엔트리가 큐 번호 */
u16 queue = reta[hash & (reta_size - 1)];

XPS: eXpress Path Send (송신 큐 CPU 어피니티)

RSS가 수신 큐를 CPU에 매핑하는 것처럼, XPS는 송신 큐도 CPU에 매핑합니다. 동일 CPU에서 RX/TX를 처리하여 캐시 지역성을 극대화합니다.

# XPS 설정: TX 큐 0을 CPU 0에 할당
echo 1 > /sys/class/net/eth0/queues/tx-0/xps_cpus

# XPS RXQS 모드: RX 큐와 동일한 CPU로 TX 큐 매핑 (RSS/XPS 통합)
echo 1 > /sys/class/net/eth0/queues/tx-0/xps_rxqs

# 4큐 NIC에서 CPU-큐 1:1 대응 설정 스크립트
for i in 0 1 2 3; do
    echo $((1 << i)) > /sys/class/net/eth0/queues/tx-$i/xps_cpus
done

NUMA 토폴로지(Topology)와 NIC 큐 배치

PCIe 슬롯의 NUMA 노드와 NIC 큐를 처리하는 CPU의 NUMA 노드가 다르면 메모리 접근 레이턴시가 증가합니다. NUMA 노드를 확인하고 큐-CPU를 같은 노드로 배치해야 합니다.

# NIC의 NUMA 노드 확인
cat /sys/class/net/eth0/device/numa_node

# PCIe 슬롯의 NUMA 노드 확인 (PCI 주소 먼저 확인)
ethtool -i eth0 | grep bus-info
cat /sys/bus/pci/devices/0000:81:00.0/numa_node

# NUMA 노드 0의 CPU 목록 확인
numactl --hardware | grep "node 0 cpus"

# NUMA 노드 0에 속한 CPU에만 IRQ 어피니티 설정 (예: CPU 0-7이 NUMA 0)
for irq in $(cat /proc/interrupts | grep eth0 | awk '{print $1}' | tr -d ':'); do
    echo 00ff > /proc/irq/$irq/smp_affinity  # CPU 0-7 = 0x00ff
done

# 프로세스도 동일 NUMA 노드에 바인딩
numactl --cpunodebind=0 --membind=0 ./myapp

aRFS: accelerated Receive Flow Steering

aRFS는 HW flow director(Intel Ethernet 등)를 활용하여 특정 플로우를 특정 큐로 자동 라우팅(Routing)합니다. RFS가 소프트웨어로 CPU를 선택한다면, aRFS는 하드웨어가 직접 큐를 선택합니다.

/* aRFS: 커널이 ndo_rx_flow_steer()로 드라이버에 플로우→큐 매핑 설정 */
struct net_device_ops {
    /* ... */
    int  (*ndo_rx_flow_steer)(struct net_device *dev,
                              const struct sk_buff *skb,
                              u16 rxq_index,
                              u32 flow_id);
};

# aRFS 활성화 (ntuple 필터 지원 NIC 필요)
ethtool -K eth0 ntuple on

# RFS 전역 플로우 테이블 크기 설정 (aRFS도 이 테이블 활용)
echo 32768 > /proc/sys/net/core/rps_sock_flow_entries

# 큐별 플로우 수 설정
echo 2048 > /sys/class/net/eth0/queues/rx-0/rps_flow_cnt

멀티큐 NAPI 드라이버 초기화 패턴

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
struct mynic_adapter {
    struct net_device   *netdev;
    int                  num_queues;
    struct mynic_rx_ring rx_rings[MYNIC_MAX_QUEUES];
};

static int mynic_open(struct net_device *netdev)
{
    struct mynic_adapter *adapter = netdev_priv(netdev);
    int i, err;

    for (i = 0; i < adapter->num_queues; i++) {
        struct mynic_rx_ring *ring = &adapter->rx_rings[i];

        netif_napi_add(netdev, &ring->napi,
                       mynic_poll, NAPI_POLL_WEIGHT);
        napi_enable(&ring->napi);

        err = request_irq(adapter->msix_entries[i].vector,
                           mynic_irq_handler, 0,
                           adapter->irq_names[i], ring);
        if (err)
            goto err_irq;

        irq_set_affinity_hint(adapter->msix_entries[i].vector,
                              cpumask_of(i % num_online_cpus()));
    }
    return 0;

err_irq:
    while (--i >= 0) {
        free_irq(adapter->msix_entries[i].vector, &adapter->rx_rings[i]);
        napi_disable(&adapter->rx_rings[i].napi);
        netif_napi_del(&adapter->rx_rings[i].napi);
    }
    return err;
}

실전 스크립트: 큐 수, IRQ 어피니티, XPS 일괄 설정

#!/bin/bash
# multiqueue_setup.sh: 멀티큐 NAPI 최적화 일괄 설정

NIC=eth0
NUM_QUEUES=8

# 1. 큐 수 설정 (NIC 지원 최대값 확인 후)
ethtool -L $NIC combined $NUM_QUEUES

# 2. irqbalance 중지 (수동 어피니티 설정 시 필수)
systemctl stop irqbalance

# 3. IRQ 어피니티: 각 큐 IRQ를 해당 CPU에 고정
i=0
for irq in $(grep "${NIC}-rx" /proc/interrupts | awk -F: '{print $1}'); do
    echo $((1 << i)) > /proc/irq/$irq/smp_affinity
    echo "IRQ $irq → CPU $i"
    i=$((i + 1))
    [ $i -ge $NUM_QUEUES ] && break
done

# 4. XPS 설정: TX 큐도 동일 CPU에 바인딩
for i in $(seq 0 $((NUM_QUEUES - 1))); do
    echo $((1 << i)) > /sys/class/net/$NIC/queues/tx-$i/xps_cpus
done

# 5. RPS/RFS 설정 (단일 큐 NIC 폴백 또는 추가 분산)
echo 32768 > /proc/sys/net/core/rps_sock_flow_entries
for i in $(seq 0 $((NUM_QUEUES - 1))); do
    echo $((0xFF << (i * 0))) > /sys/class/net/$NIC/queues/rx-$i/rps_cpus
    echo 2048 > /sys/class/net/$NIC/queues/rx-$i/rps_flow_cnt
done

# 6. 링 버퍼 크기 최대화
MAX_RX=$(ethtool -g $NIC | grep "RX:" | head -1 | awk '{print $2}')
ethtool -G $NIC rx $MAX_RX

echo "멀티큐 NAPI 설정 완료: $NIC, $NUM_QUEUES 큐"

RSS와 인터럽트 어피니티

# RX 큐별 IRQ 확인
cat /proc/interrupts | grep eth0

# IRQ 126을 CPU 3에 고정
echo 8 > /proc/irq/126/smp_affinity  # CPU 3 = 비트 3 = 0x8

# ethtool로 RSS 큐 수 확인/변경
ethtool -l eth0
ethtool -L eth0 combined 8

# RSS 해시 키 및 필드 설정
ethtool -x eth0
ethtool -X eth0 hkey <key>

# irqbalance 중지 후 수동 어피니티 설정 권장
systemctl stop irqbalance

스레드 NAPI (Threaded NAPI)

스레드 NAPI의 배경

기존 NAPI는 소프트IRQ 컨텍스트에서 실행되므로 실시간(RT) 커널과 충돌이 발생합니다. 소프트IRQ는 실시간 태스크(Task)보다 낮은 우선순위(Priority)를 가지지만, 선점(Preemption) 불가 구간에서 실행되므로 레이턴시 스파이크를 유발합니다. 스레드 NAPI(Threaded NAPI)는 poll()을 커널 스레드(Kernel Thread)로 옮겨 이 문제를 해결합니다.

스레드 NAPI 활성화 API

/* 스레드 NAPI 활성화 (드라이버 probe에서 또는 런타임에) */
int napi_set_threaded(struct napi_struct *napi, bool threaded);

/* 활성화 시 커널이 자동으로 스레드 생성:
   스레드 이름: "napi/<netdev_name>-<queue_idx>"
   예: "napi/eth0-0", "napi/eth0-1" */

/* 전체 디바이스에 대해 스레드 NAPI 활성화 */
void dev_set_threaded(struct net_device *dev, bool threaded);

napi_threaded_poll() 커널 스레드 함수 구현

/* net/core/dev.c: 스레드 NAPI의 커널 스레드 메인 함수 */
static int napi_threaded_poll(void *data)
{
    struct napi_struct *napi = data;
    struct net_device  *dev  = napi->dev;
    void               *have;

    while (!kthread_should_stop()) {
        /* 1. 처리할 패킷이 있을 때까지 대기 */
        do {
            set_current_state(TASK_INTERRUPTIBLE);
            if (kthread_should_stop())
                break;
            if (napi_schedule_prep(napi)) {
                __set_current_state(TASK_RUNNING);
                break;
            }
            schedule();  /* CPU 반납, wake_up_process()로 깨어남 */
        } while (1);

        if (kthread_should_stop())
            break;

        /* 2. local_bh_disable: softIRQ와의 동시 실행 방지 */
        local_bh_disable();
        have = netpoll_poll_lock(napi);

        /* 3. NAPI poll 실행: 드라이버 poll() 직접 호출 */
        if (test_bit(NAPI_STATE_SCHED_THREADED, &napi->state)) {
            napi_poll(napi, NULL);
        }

        netpoll_poll_unlock(have);
        local_bh_enable();
    }

    __set_current_state(TASK_RUNNING);
    return 0;
}

런타임 sysfs 제어

# 특정 NIC의 스레드 NAPI 활성화
echo 1 > /sys/class/net/eth0/threaded

# 스레드 NAPI 스레드 확인
ps aux | grep napi/eth0

# 스레드 우선순위 조정 (SCHED_FIFO RT 스케줄러 사용)
chrt -f -p 50 $(pgrep "napi/eth0-0")

# 스레드를 특정 CPU에 고정 (CPU 격리과 함께 사용)
taskset -p 0x10 $(pgrep "napi/eth0-0")  # CPU 4에 고정

# cgroup cpuset으로 스레드 격리
echo $(pgrep "napi/eth0-0") > /sys/fs/cgroup/cpuset/realtime/tasks

PREEMPT_RT와 소프트IRQ 스레드화

CONFIG_PREEMPT_RT가 활성화된 실시간 커널에서는 소프트IRQ가 자동으로 스레드화됩니다. 이 경우 ksoftirqd가 각 CPU에서 실시간 스케줄러(Scheduler)로 동작합니다.

스레드 NAPI 우선순위 정책 권장 표

스레드 NAPI + CPU 격리(Isolation) 조합

# GRUB 설정: CPU 8-15를 일반 스케줄러에서 격리
# /etc/default/grub: GRUB_CMDLINE_LINUX="isolcpus=8-15 nohz_full=8-15 rcu_nocbs=8-15"
update-grub && reboot

# 격리 후 스레드 NAPI를 격리 CPU에 배치
echo 1 > /sys/class/net/eth0/threaded
for i in $(seq 0 7); do
    pid=$(pgrep "napi/eth0-$i")
    taskset -p $((1 << (i + 8))) $pid  # CPU 8+i에 배치
    chrt -f -p 60 $pid                   # RT 우선순위 60
done

# cgroup cpuset으로 네트워크 전용 CPU 격리
mkdir -p /sys/fs/cgroup/cpuset/netpoll
echo 8-15 > /sys/fs/cgroup/cpuset/netpoll/cpuset.cpus
echo 0 > /sys/fs/cgroup/cpuset/netpoll/cpuset.mems
for pid in $(pgrep "napi/eth0"); do
    echo $pid > /sys/fs/cgroup/cpuset/netpoll/tasks
done

결정론적 지연(Deterministic Latency) 측정 방법

# cyclictest로 인터럽트 레이턴시 측정 (스레드 NAPI 효과 확인)
cyclictest -m -sp99 -d0 -i200 -l10000 --cpu=8

# hping3으로 왕복 레이턴시 측정 (마이크로초 단위)
hping3 -S --fast -p 80 --icmp target_ip 2>&1 | awk '/rtt/{print $NF}'

# perf latency 추적: NAPI poll 시작부터 소켓 수신까지
perf trace -e 'napi:napi_poll,sock:inet_sock_set_state' -a sleep 5

# bpftrace로 IRQ → NAPI poll 레이턴시 측정
bpftrace -e '
kprobe:__napi_schedule { @t[arg0] = nsecs; }
kprobe:napi_poll / @t[arg0] / {
    $lat = (nsecs - @t[arg0]) / 1000;
    @sched_to_poll_us = hist($lat);
    delete(@t[arg0]);
}
interval:s:5 { print(@sched_to_poll_us); }'

Per-NAPI 버지 폴링 via Netlink (Linux 6.6+)

Linux 6.6부터 ethtool Netlink 인터페이스를 통해 개별 NAPI 인스턴스에 대한 세밀한 제어가 가능해졌습니다. ETHTOOL_MSG_NAPI_SET 명령으로 per-NAPI IRQ suspend timeout과 버지 폴링 파라미터를 설정할 수 있습니다.

/* net/ethtool/napi.c — 간략화 */
/* ethtool Netlink: per-NAPI 설정 (Linux 6.6+) */

/* NAPI ID 조회 */
/* ethtool --json -S eth0 로 napi_id 확인 가능 */

/* Netlink 명령 구조:
   ETHTOOL_MSG_NAPI_GET    → NAPI 인스턴스 목록/상태 조회
   ETHTOOL_MSG_NAPI_SET    → per-NAPI 파라미터 설정

   설정 가능 속성:
   ETHTOOL_A_NAPI_IRQ_SUSPEND_TIMEOUT  → IRQ 유예 타임아웃 (ns)
   ETHTOOL_A_NAPI_DEFER_HARD_IRQS      → 하드 IRQ 연기 횟수 */

/* 사용자 공간에서 per-NAPI 설정 예 (libnl 기반) */
struct nlattr *nla;
nla_put_u32(msg, ETHTOOL_A_NAPI_ID, napi_id);
nla_put_u64_64bit(msg, ETHTOOL_A_NAPI_IRQ_SUSPEND_TIMEOUT,
                  100000, /* 100μs */
                  ETHTOOL_A_NAPI_PAD);

IRQ/스레드 마이그레이션 전략

스레드 NAPI에서는 IRQ 어피니티와 NAPI 스레드 어피니티를 동기화하는 것이 중요합니다. 불일치 시 IRQ가 CPU A에서 발생하지만 poll()은 CPU B에서 실행되어 캐시 바운싱과 불필요한 IPI(Inter-Processor Interrupt)가 발생합니다.

# IRQ와 NAPI 스레드를 동일 CPU에 고정하는 스크립트
# 1. IRQ 번호와 NAPI 스레드 PID 매핑
for q in $(seq 0 7); do
    irq=$(grep "eth0-TxRx-$q" /proc/interrupts | awk '{print $1}' | tr -d ':')
    pid=$(pgrep -f "napi/eth0-$q")
    cpu=$q

    # IRQ 어피니티 설정
    echo $((1 << cpu)) > /proc/irq/$irq/smp_affinity

    # NAPI 스레드도 동일 CPU에 고정
    taskset -p $((1 << cpu)) $pid
done

napi_thread_fn() 상태 전이 상세

스레드 NAPI의 커널 스레드는 NAPI_STATE_SCHED_THREADED 비트를 통해 softIRQ 경로와 구분됩니다. IRQ 핸들러에서 napi_schedule() 호출 시 이 비트의 존재 여부에 따라 softIRQ 또는 스레드 wake-up 경로가 선택됩니다.

/* net/core/dev.c — 간략화 */
/* napi_schedule() → 스레드 NAPI 경로 분기 */
void ____napi_schedule(struct softnet_data *sd,
                        struct napi_struct *napi)
{
    if (test_bit(NAPI_STATE_THREADED, &napi->state)) {
        /* 스레드 NAPI: kthread를 wake_up */
        if (!__napi_schedule_irqoff(napi))
            wake_up_process(napi->thread);
        return;
    }
    /* 일반 NAPI: softIRQ poll_list에 추가 */
    list_add_tail(&napi->poll_list, &sd->poll_list);
    __raise_softirq_irqoff(NET_RX_SOFTIRQ);
}

/* 상태 전이:
   [IRQ 발생]
     → napi_schedule_prep(): NAPI_STATE_SCHED 비트 설정
     → NAPI_STATE_THREADED 확인
       ├─ YES → wake_up_process(napi->thread)
       │        → napi_threaded_poll() 실행
       │        → napi_complete_done() → NAPI_STATE_SCHED 해제
       │        → schedule() (다음 IRQ 대기)
       └─ NO  → poll_list에 추가
                → NET_RX_SOFTIRQ 발생
                → net_rx_action() → poll()
                → napi_complete_done() → NAPI_STATE_SCHED 해제 */

NAPI 일시 중단 (IRQ Suspension)

배경과 필요성

Linux 6.3에서 도입된 NAPI IRQ Suspension은 유휴 상태의 NAPI 인스턴스에서 불필요한 인터럽트를 억제하여 전력 소비와 CPU 오버헤드를 줄이는 기능입니다. 멀티큐 NIC에서 일부 큐만 활성화되고 나머지는 유휴 상태인 경우가 흔한데, 기존에는 유휴 큐도 인터럽트를 주기적으로 받아 CPU를 깨웠습니다.

IRQ Suspension은 일정 기간 패킷이 도착하지 않은 NAPI 인스턴스의 인터럽트를 일시 중단하고, 패킷이 다시 도착하면 자동으로 재개합니다. 이는 특히 다음 환경에서 효과적입니다:

• 다수의 RX 큐를 가진 고속 NIC (25G/100G) — 유휴 큐 비율이 높음
• 서버 통합 환경 — 여러 VM/컨테이너(Container)가 NIC를 공유
• 전력 효율이 중요한 데이터센터 — C-state 진입 빈도 증가

napi_suspend_irqs() / napi_resume_irqs() API

/* include/linux/netdevice.h */

/* IRQ 일시 중단: poll() 완료 시 유휴 판단 후 호출 */
bool napi_suspend_irqs(struct napi_struct *napi);

/* IRQ 재개: 패킷 도착 또는 타임아웃 시 호출 */
void napi_resume_irqs(struct napi_struct *napi);

/* napi_suspend_irqs() 내부 구현:
   1. NAPI_STATE_SCHED 비트 유지 (다른 스케줄링 차단)
   2. NIC의 해당 큐 인터럽트 마스킹
   3. gro_flush_timeout을 suspend timeout으로 활용
   4. 타이머 등록: timeout 만료 시 napi_resume_irqs() 호출

   반환값:
   true  → 성공적으로 중단됨
   false → 이미 스케줄됨 또는 중단 불가 */

/* napi_resume_irqs() 내부 구현:
   1. NAPI_STATE_SCHED 비트 해제
   2. NIC의 해당 큐 인터럽트 언마스킹
   3. 대기 중인 패킷이 있으면 즉시 napi_schedule() */

유휴 기간 최적화 전략

IRQ Suspension의 효과를 극대화하려면 Adaptive ITR과 결합하여 트래픽 패턴에 따라 suspension timeout을 동적으로 조절해야 합니다.

드라이버 구현 예제

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* poll 함수에서 IRQ Suspension 통합 패턴 */
static int mynic_poll(struct napi_struct *napi, int budget)
{
    struct mynic_ring *ring = container_of(napi, struct mynic_ring, napi);
    int work_done = mynic_clean_rx(ring, budget);

    if (work_done < budget) {
        if (napi_complete_done(napi, work_done)) {
            /* 유휴 판단: 연속 N회 빈 poll이면 suspend */
            if (work_done == 0 && ++ring->idle_count > 3) {
                if (napi_suspend_irqs(napi)) {
                    ring->suspended = true;
                    return work_done;
                }
            }
            mynic_enable_rx_irq(ring);
        }
    }

    if (work_done > 0)
        ring->idle_count = 0;

    return work_done;
}

/* IRQ 핸들러에서 resume */
static irqreturn_t mynic_irq_handler(int irq, void *data)
{
    struct mynic_ring *ring = data;

    if (ring->suspended) {
        napi_resume_irqs(&ring->napi);
        ring->suspended = false;
        ring->idle_count = 0;
    }

    napi_schedule_irqoff(&ring->napi);
    return IRQ_HANDLED;
}

전력 최적화: C-state와 NAPI Suspension

NAPI Suspension은 CPU C-state 진입과 밀접하게 관련됩니다. IRQ가 완전히 중단된 상태에서 CPU는 더 깊은 C-state로 진입할 수 있어 전력 소비를 크게 줄일 수 있습니다.

# 레이턴시 민감 환경: 깊은 C-state 비활성화
# (busy polling 또는 낮은 gro_flush_timeout 사용 시)
cpupower idle-set -D 1  # C1까지만 허용

# 전력 효율 환경: IRQ suspension + 깊은 C-state
echo 10 > /sys/class/net/eth0/napi_defer_hard_irqs
echo 200000 > /sys/class/net/eth0/gro_flush_timeout  # 200μs

# 전력 소비 모니터링
turbostat --interval 5 --show PkgWatt,CorWatt,IRQ

버지 폴링 (Busy Polling)

버지 폴링의 개념과 원리

버지 폴링(Busy Polling)은 소켓 수신 대기 중에 커널이 NAPI poll()을 반복 호출하여 패킷이 도착하면 인터럽트나 소프트IRQ를 거치지 않고 즉시 처리하는 기법입니다. 레이턴시를 수십 마이크로초에서 수 마이크로초로 줄일 수 있지만, CPU를 100% 점유하는 트레이드오프가 있습니다.

일반 수신 경로는 패킷 → NIC DMA → HW IRQ → softIRQ → 소켓 버퍼 → epoll/recv 순서이지만, 버지 폴링은 recv/recvmsg() 호출 시 소켓이 속한 NAPI를 직접 폴링하여 HW IRQ/softIRQ 경로 자체를 우회합니다.

sk_napi_id 할당 경로

/* include/linux/netdevice.h, net/core/sock.c — 간략화 */
/* 패킷 수신 시 skb → sock → sk_napi_id 설정 경로 */

/* 1단계: napi_gro_receive()에서 skb에 napi_id 기록 */
static inline void skb_mark_napi_id(struct sk_buff *skb,
                                    struct napi_struct *napi)
{
    skb->napi_id = napi->napi_id;
}

/* 2단계: tcp_v4_rcv() → sk_mark_napi_id() → 소켓에 napi_id 전파 */
static inline void sk_mark_napi_id(struct sock *sk,
                                    const struct sk_buff *skb)
{
    if (READ_ONCE(sk->sk_napi_id) != skb->napi_id)
        WRITE_ONCE(sk->sk_napi_id, skb->napi_id);
}

/* 3단계: recvmsg() 진입 시 sk_napi_id로 버지 폴링 NAPI 결정 */
static inline int sock_recvmsg(struct socket *sock, struct msghdr *msg,
                                int flags)
{
    /* SO_BUSY_POLL 또는 SO_PREFER_BUSY_POLL 설정 시 버지 폴링 먼저 시도 */
    if (sk_can_busy_loop(sock->sk) &&
        skb_queue_empty_lockless(&sock->sk->sk_receive_queue))
        sk_busy_loop(sock->sk, flags & MSG_DONTWAIT);
    ...
}

소켓 수준 버지 폴링 설정

/* SO_BUSY_POLL: 폴링 대기 시간 설정 (마이크로초) */
int busy_poll_us = 50;  /* 50μs 동안 버지 폴링 */
setsockopt(fd, SOL_SOCKET, SO_BUSY_POLL,
           &busy_poll_us, sizeof(busy_poll_us));

/* SO_PREFER_BUSY_POLL: 항상 버지 폴링 선호 (Linux 5.11+) */
int val = 1;
setsockopt(fd, SOL_SOCKET, SO_PREFER_BUSY_POLL,
           &val, sizeof(val));

/* SO_BUSY_POLL_BUDGET: NAPI poll()당 처리할 최대 패킷 수 (Linux 5.11+) */
int budget = 8;
setsockopt(fd, SOL_SOCKET, SO_BUSY_POLL_BUDGET,
           &budget, sizeof(budget));

epoll + 버지 폴링 연동

/* fs/eventpoll.c — 간략화 */
/* epoll_wait() 내부에서 BUSY_POLL 처리 흐름 */
/* ep_poll() → ep_busy_loop() → sk_busy_loop() 경로 */

static int ep_busy_loop(struct eventpoll *ep, int nonblock)
{
    unsigned int napi_id = ep_get_busy_poll_napi_id(ep);

    if (!napi_id)
        return false;

    return napi_busy_loop(napi_id,
                          nonblock ? NULL : ep_busy_loop_end,
                          ep,
                          prefer_busy_poll(ep),
                          ep->busy_poll_budget);
}

/* 실제 사용: epoll + SO_BUSY_POLL 조합 */
int setup_epoll_busy_poll(int epfd, int sockfd)
{
    struct epoll_event ev = { .events = EPOLLIN, .data.fd = sockfd };

    /* 소켓에 버지 폴링 활성화 */
    int bp = 50;
    setsockopt(sockfd, SOL_SOCKET, SO_BUSY_POLL, &bp, sizeof(bp));
    int prefer = 1;
    setsockopt(sockfd, SOL_SOCKET, SO_PREFER_BUSY_POLL, &prefer, sizeof(prefer));

    return epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);
}

/* epoll_wait 호출 시 내부적으로 버지 폴링 먼저 시도 후 블록 */
int ready = epoll_wait(epfd, events, MAX_EVENTS, timeout_ms);

io_uring + 버지 폴링

/* io_uring: IORING_FEAT_FAST_POLL을 통한 버지 폴링 통합 */
struct io_uring_params params = {};
int ring_fd = io_uring_setup(256, &params);

/* IORING_FEAT_FAST_POLL 지원 여부 확인 */
if (params.features & IORING_FEAT_FAST_POLL) {
    /* io_uring이 소켓의 버지 폴링을 자동으로 활용
       IORING_OP_RECV, IORING_OP_RECVMSG 등에서 적용됨 */
}

/* io_uring SQE 제출: 버지 폴링 활성화 플래그 */
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_recv(sqe, sockfd, buf, sizeof(buf), 0);
/* io_uring은 SO_BUSY_POLL 설정된 소켓에 대해 자동으로 fast_poll 경로 사용 */
io_uring_submit(&ring);

버지 폴링 수신 경로 비교

내부 구현: napi_busy_loop()

/* net/socket.c의 recvmsg() → sock_recvmsg() 경로에서 호출 */
bool sk_busy_loop(struct sock *sk, int nonblock)
{
    unsigned int napi_id = READ_ONCE(sk->sk_napi_id);
    struct napi_struct *napi;

    if (napi_id < MIN_NAPI_ID)
        return false;

    napi = napi_by_id(napi_id);
    if (!napi)
        return false;

    return napi_busy_loop(napi_id, nonblock ? NULL : sk_busy_loop_end, sk,
                          prefer_busy_poll(sk), READ_ONCE(sk->sk_ll_usec));
}

/* napi_busy_loop: 지정된 NAPI를 직접 반복 폴링 */
bool napi_busy_loop(unsigned int napi_id,
                    bool (*loop_end)(void *, unsigned long),
                    void *loop_end_arg,
                    bool prefer_busy_poll,
                    u16 budget)
{
    struct napi_struct *napi;
    unsigned long       start_time = local_clock();

    do {
        rcu_read_lock();
        napi = napi_by_id(napi_id);
        if (napi) {
            /* NAPI_STATE_IN_BUSY_POLL 세팅으로 softIRQ와 동시 실행 방지 */
            if (!napi_try_get(napi))
                goto busy_loop_end;
            napi_poll(napi, NULL);  /* 직접 폴링 */
            napi_put(napi);
        }
        rcu_read_unlock();

        if (loop_end && loop_end(loop_end_arg, start_time))
            return true;
        cpu_relax();  /* PAUSE 명령으로 CPU 전력 절감 + 하이퍼스레딩 힌트 */
    } while (!need_resched());

busy_loop_end:
    rcu_read_unlock();
    return false;
}

레이턴시 비교 표

시스템 전역 설정

# 전역 기본 버지 폴링 시간 (μs, 0이면 비활성)
echo 50 > /proc/sys/net/core/busy_poll

# 전역 기본 버지 읽기 시간
echo 50 > /proc/sys/net/core/busy_read

sk_busy_loop() / sk_can_busy_loop() 내부

sk_busy_loop()는 소켓의 recvmsg()/epoll_wait() 경로에서 호출되어 지정된 시간 동안 NAPI를 직접 폴링합니다. 이 함수가 버지 폴링의 실체입니다.

/* net/core/dev.c: sk_busy_loop() 핵심 로직 */
void sk_busy_loop(struct sock *sk, int nonblock)
{
    unsigned int napi_id = READ_ONCE(sk->sk_napi_id);
    unsigned long end_time = busy_loop_end_time(sk, nonblock);
    int (*busy_poll)(
        struct napi_struct *napi, int budget);

    /* NAPI ID → napi_struct 조회 (해시 테이블) */
    struct napi_struct *napi = napi_by_id(napi_id);
    if (!napi)
        return;

    /* 반복 폴링 루프 */
    do {
        /* NAPI poll 직접 호출 (budget = SO_BUSY_POLL_BUDGET) */
        napi_busy_loop(napi, busy_poll,
                       sk_busy_loop_end, sk);
    } while (!sk_busy_loop_end(sk, end_time));
}

/* sk_can_busy_loop(): 버지 폴링 가능 여부 판단 */
static inline bool sk_can_busy_loop(struct sock *sk)
{
    return READ_ONCE(sk->sk_napi_id) &&
           !signal_pending(current) &&
           !need_resched();
    /* 조건: NAPI ID 바인딩 + 시그널 없음 + 재스케줄 불필요 */
}

/* 종료 조건 (sk_busy_loop_end):
   1. 소켓에 데이터 도착 (sk_rcvlowat 충족)
   2. timeout 만료 (busy_poll/busy_read 설정값)
   3. need_resched() — 다른 태스크가 CPU 요청
   4. signal_pending() — 시그널 수신 */

SO_INCOMING_NAPI_ID 소켓 옵션 활용

SO_INCOMING_NAPI_ID는 소켓에 마지막으로 패킷을 전달한 NAPI 인스턴스의 ID를 사용자 공간에서 조회할 수 있게 합니다. 이 정보를 활용하면 특정 NAPI(=특정 CPU)에 소켓을 어피니티 바인딩하여 캐시 효율을 극대화할 수 있습니다.

/* 사용자 공간: SO_INCOMING_NAPI_ID 조회 */
unsigned int napi_id;
socklen_t len = sizeof(napi_id);

getsockopt(fd, SOL_SOCKET, SO_INCOMING_NAPI_ID, &napi_id, &len);

/* napi_id를 활용한 CPU 어피니티 최적화:
   1. napi_id → IRQ 번호 → CPU 매핑 조회
   2. 워커 스레드를 해당 CPU에 고정
   3. epoll 그룹별 NAPI 어피니티 분리 */

/* ethtool Netlink로 NAPI 정보 조회 (Linux 6.6+) */
/* ETHTOOL_MSG_NAPI_GET:
   응답에 NAPI ID, IRQ 번호, 큐 인덱스,
   per-NAPI 통계 포함 */

/* 커널 내부: 소켓에 NAPI ID가 기록되는 시점 */
/* TCP RX: tcp_v4_rcv() → sk_mark_napi_id()
   UDP RX: udp_queue_rcv_skb() → sk_mark_napi_id()
   → sk->sk_napi_id = skb->napi_id; */

SO_PREFER_BUSY_POLL / SO_BUSY_POLL_BUDGET

Linux 5.11에서 추가된 이 소켓 옵션들은 per-소켓 버지 폴링을 더 세밀하게 제어합니다.

/* per-소켓 버지 폴링 설정 */
int prefer = 1;
setsockopt(fd, SOL_SOCKET, SO_PREFER_BUSY_POLL,
           &prefer, sizeof(prefer));
/* 효과: 이 소켓의 NAPI에 NAPI_STATE_PREFER_BUSY_POLL 설정
   → napi_complete_done()에서 IRQ 재활성화를 지연
   → 버지 폴링 소켓이 독점적으로 NAPI 사용 가능 */

int budget = 32;
setsockopt(fd, SOL_SOCKET, SO_BUSY_POLL_BUDGET,
           &budget, sizeof(budget));
/* 효과: 버지 폴링 시 per-call budget 조절
   기본값 8 → 32로 증가 시 한 번의 폴링에서 더 많은 패킷 처리
   트레이드오프: 높은 budget = 높은 처리량, 긴 폴링 시간 */

/* NAPI_STATE_PREFER_BUSY_POLL 상호작용:
   napi_complete_done() 내부:
   if (test_bit(NAPI_STATE_PREFER_BUSY_POLL, &napi->state)) {
       // IRQ 재활성화를 gro_flush_timeout 후로 연기
       // → 버지 폴링 소켓이 다시 폴링할 기회 제공
       napi_schedule_irqoff(napi);  // 바로 재스케줄
       return false;
   } */

NAPI 해시 테이블과 소켓 바인딩

NAPI 해시 테이블 개요

커널은 모든 활성 NAPI 인스턴스를 전역 해시 테이블 napi_hash[]에 등록합니다. 이 테이블의 주요 목적은 NAPI ID로 napi_struct를 빠르게 조회하는 것이며, 버지 폴링과 Netlink 인터페이스에서 핵심적으로 사용됩니다.

/* net/core/dev.c */
#define NAPI_HASH_SIZE  256  /* 해시 버킷 수 (2^8) */

/* 전역 해시 테이블: hlist_head 배열 */
static struct hlist_head napi_hash[NAPI_HASH_SIZE];

/* 해시 함수: NAPI ID → 버킷 인덱스 */
static inline struct hlist_head *
napi_hash_bucket(unsigned int napi_id)
{
    return &napi_hash[napi_id % NAPI_HASH_SIZE];
}

/* NAPI ID는 per-net_device 순차 할당:
   netif_napi_add() → napi->napi_id = ++napi_gen_id;
   (전역 atomic counter) */

napi_hash_add() / napi_hash_del()

NAPI 인스턴스가 생성/삭제될 때 해시 테이블에 자동으로 추가/제거됩니다. RCU(Read-Copy-Update)로 보호되어 조회 측은 lock 없이 안전하게 접근할 수 있습니다.

/* net/core/dev.c */
static void napi_hash_add(struct napi_struct *napi)
{
    /* netif_napi_add()에서 자동 호출 */
    if (test_bit(NAPI_STATE_NO_BUSY_POLL, &napi->state))
        return;  /* 버지 폴링 비활성 NAPI는 해시 등록 생략 */

    spin_lock(&napi_hash_lock);
    hlist_add_head_rcu(&napi->napi_hash_node,
                       napi_hash_bucket(napi->napi_id));
    spin_unlock(&napi_hash_lock);
}

static void napi_hash_del(struct napi_struct *napi)
{
    spin_lock(&napi_hash_lock);
    hlist_del_init_rcu(&napi->napi_hash_node);
    spin_unlock(&napi_hash_lock);

    /* RCU grace period 대기:
       이미 napi_by_id()로 조회 중인 reader가
       안전하게 완료할 때까지 실제 해제 지연 */
    synchronize_rcu();
}

napi_by_id() — NAPI 조회

napi_by_id()는 NAPI ID를 키로 해시 테이블에서 napi_struct를 조회합니다. RCU read-side critical section 내에서 호출되며, 버지 폴링과 Netlink 인터페이스의 핵심입니다.

/* net/core/dev.c */
struct napi_struct *napi_by_id(unsigned int napi_id)
{
    struct napi_struct *napi;
    struct hlist_head *head =
        napi_hash_bucket(napi_id);

    /* RCU 보호 하에 해시 체인 순회 */
    hlist_for_each_entry_rcu(napi, head, napi_hash_node) {
        if (napi->napi_id == napi_id)
            return napi;
    }
    return NULL;
}

/* 사용처:
   1. sk_busy_loop() → napi_by_id(sk->sk_napi_id)
      → 소켓의 NAPI 인스턴스 직접 폴링
   2. ethtool Netlink ETHTOOL_MSG_NAPI_GET
      → NAPI 인스턴스 정보 조회
   3. SO_INCOMING_NAPI_ID getsockopt
      → 소켓에 바인딩된 NAPI ID 반환 */

sk_mark_napi_id() / sk_mark_napi_id_once()

소켓-NAPI 바인딩은 패킷 수신 경로에서 자동으로 이루어집니다. sk_mark_napi_id()는 수신된 SKB의 napi_id를 소켓에 기록하여 이후 버지 폴링 시 올바른 NAPI 인스턴스를 찾을 수 있게 합니다.

/* include/net/busy_poll.h */
static inline void sk_mark_napi_id(
    struct sock *sk,
    const struct sk_buff *skb)
{
    /* 매 패킷 수신 시 갱신 (마이그레이션 추적) */
    WRITE_ONCE(sk->sk_napi_id, skb->napi_id);
}

static inline void sk_mark_napi_id_once(
    struct sock *sk,
    const struct sk_buff *skb)
{
    /* 최초 한 번만 기록 (TCP 연결 수립 시) */
    if (!READ_ONCE(sk->sk_napi_id))
        WRITE_ONCE(sk->sk_napi_id, skb->napi_id);
}

/* 호출 경로:
   TCP: tcp_v4_rcv() → tcp_v4_do_rcv()
        → sk_mark_napi_id(sk, skb)
   UDP: udp_queue_rcv_skb()
        → sk_mark_napi_id(sk, skb)
   TCP Listener: tcp_v4_conn_request()
        → sk_mark_napi_id_once(sk, skb)

   skb->napi_id는 NAPI poll() 시작 시 설정:
   napi_gro_receive() → skb->napi_id = napi->napi_id; */

SO_INCOMING_NAPI_ID 상세

사용자 공간 애플리케이션은 SO_INCOMING_NAPI_ID로 소켓에 바인딩된 NAPI ID를 조회하고, 이를 기반으로 CPU 어피니티를 최적화할 수 있습니다.

/* 실전 패턴: NAPI 어피니티 기반 워커 배치 */
int optimize_worker_affinity(int sockfd)
{
    unsigned int napi_id;
    socklen_t len = sizeof(napi_id);

    /* 1. 소켓의 NAPI ID 조회 */
    getsockopt(sockfd, SOL_SOCKET,
               SO_INCOMING_NAPI_ID, &napi_id, &len);

    /* 2. /sys/class/net/eth0/napi_defer_hard_irqs 등으로
          NAPI ID → CPU 매핑 확인 */

    /* 3. 워커 스레드를 해당 CPU에 고정 */
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(target_cpu, &cpuset);
    sched_setaffinity(0, sizeof(cpuset), &cpuset);

    return 0;
}

/* Netlink 기반 NAPI 정보 조회 (Linux 6.6+):
   ETHTOOL_MSG_NAPI_GET → 응답:
   ETHTOOL_A_NAPI_ID           → NAPI 인스턴스 ID
   ETHTOOL_A_NAPI_IFINDEX      → 네트워크 인터페이스 인덱스
   ETHTOOL_A_NAPI_IRQ          → 연결된 IRQ 번호
   ETHTOOL_A_NAPI_PID          → 스레드 NAPI PID (있는 경우) */

XDP와 NAPI 연동

XDP의 실행 위치

XDP(eXpress Data Path)는 NAPI poll() 내부에서 패킷이 sk_buff로 변환되기 전에 실행됩니다. NIC 드라이버가 DMA에서 직접 페이지를 받아 XDP 프로그램에 전달하므로 커널 네트워크 스택(Network Stack) 오버헤드를 완전히 우회할 수 있습니다.

XDP 모드 비교

xdp_rxq_info와 napi_struct 연결

/* include/net/xdp.h — 간략화 */
/* XDP RX 큐 정보 구조체: NAPI와 XDP 프로그램을 연결 */
struct xdp_rxq_info {
    struct net_device  *dev;
    u32                 queue_index;
    u32                 reg_state;
    struct xdp_mem_info mem;
    unsigned int        napi_id;   /* NAPI 인스턴스 ID */
    u32                 frag_size;
} __rcu;

/* 드라이버 probe/open에서 xdp_rxq_info 등록 */
static int mynic_setup_xdp_rxq(struct mynic_rx_ring *ring)
{
    int err;

    /* XDP RX 큐 정보 등록 */
    err = xdp_rxq_info_reg(&ring->xdp_rxq,
                           ring->hw->netdev,
                           ring->queue_idx,
                           ring->napi.napi_id);  /* NAPI ID 연결 */
    if (err)
        return err;

    /* 메모리 모델 등록: page_pool 사용 시 */
    err = xdp_rxq_info_reg_mem_model(&ring->xdp_rxq,
                                       MEM_TYPE_PAGE_POOL,
                                       ring->page_pool);
    return err;
}

XDP metadata: xdp_buff.data_meta 활용

/* include/net/xdp.h — 간략화 */
/* XDP metadata 영역: data_meta ~ data 사이에 드라이버/BPF 메타데이터 저장 */
struct xdp_buff {
    void               *data;        /* 패킷 데이터 시작 */
    void               *data_end;    /* 패킷 데이터 끝 */
    void               *data_meta;   /* 메타데이터 시작 (data 이전) */
    void               *data_hard_start; /* 페이지 헤드룸 시작 */
    struct xdp_rxq_info *rxq;
    struct xdp_txq_info *txq;
    u32                  frame_sz;
    u32                  flags;
};

/* BPF 프로그램에서 메타데이터 조작 */
/* bpf_xdp_adjust_meta(ctx, delta): data_meta 포인터를 delta만큼 이동
   양수 delta: 메타데이터 영역 축소, 음수: 확장 */

/* 예: 드라이버가 타임스탬프를 메타데이터에 기록 */
struct meta {
    u64 rx_timestamp;
} *meta;

/* BPF 코드: */
int bpf_prog(struct xdp_md *ctx) {
    if (bpf_xdp_adjust_meta(ctx, -(int)sizeof(*meta)))
        return XDP_ABORTED;
    meta = (void *)(long)ctx->data_meta;
    meta->rx_timestamp = bpf_ktime_get_ns();
    return XDP_PASS;
}

AF_XDP zero-copy 경로 상세

/* include/net/xdp_sock_drv.h, net/xdp/xsk.c — 간략화 */
/* AF_XDP 소켓 구조: 사용자 공간 ↔ 커널 간 zero-copy 패킷 교환 */

/* 4개의 링 구조:
   1. UMEM fill ring: 사용자 → 커널 (빈 버퍼 공급)
   2. UMEM completion ring: 커널 → 사용자 (TX 완료 버퍼 반환)
   3. RX ring: 커널 → 사용자 (수신 패킷 전달)
   4. TX ring: 사용자 → 커널 (송신 패킷 전달) */

/* 드라이버 측 XDP_REDIRECT → xsk_map 경로 */
int xsk_rcv(struct xdp_sock *xs, struct xdp_buff *xdp)
{
    u64 addr;
    int err;

    /* 패킷 데이터를 UMEM RX 버퍼에 직접 기록 (zero-copy) */
    addr = xp_get_handle(xs->pool, xdp->data);
    err  = xskq_prod_reserve_desc(xs->rx, addr, xdp->data_end - xdp->data);
    if (err)
        return err;

    xsk_set_rx_need_wakeup(xs->pool);
    return 0;
}

# XDP 프로그램 로드 (native mode)
ip link set eth0 xdp obj myxdp.o

# offloaded mode: NIC 하드웨어에서 실행
ip link set eth0 xdpoffload obj myxdp.o

# generic mode: 스택 최상단 (드라이버 지원 불필요)
ip link set eth0 xdpgeneric obj myxdp.o

# AF_XDP 버지 폴링과 SO_PREFER_BUSY_POLL 결합
# → NAPI가 XDP verdict 처리 후 사용자 공간까지 레이턴시 최소화

XDP 처리 경로 다이어그램

XDP_REDIRECT와 xdp_do_flush()

/* poll() 끝에서 XDP 리다이렉트 플러시 */
if (xdp_redirect_used) {
    xdp_do_flush();  /* 리다이렉트 큐 → 목적지로 일괄 전송 */
}
if (work_done < budget && napi_complete_done(napi, work_done))
    mynic_enable_rx_irq(ring);

NAPI poll() 내 XDP 처리 전체 흐름

드라이버 poll() 함수 내에서 XDP 프로그램이 실행되는 정확한 위치와 각 verdict별 처리 로직입니다. 이 코드는 실제 드라이버(ixgbe, ice, mlx5 등)의 공통 패턴을 정리한 것입니다.

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* 드라이버 poll() 내부의 XDP 처리 전체 흐름 (공통 패턴) */
static int mynic_poll_xdp(struct napi_struct *napi, int budget)
{
    struct mynic_rx_ring *ring = container_of(napi, ...);
    struct bpf_prog *xdp_prog;
    int work_done = 0;
    bool xdp_xmit = false;
    u32 act;

    /* RCU 보호 하에 XDP 프로그램 참조 (hot path 최적화) */
    xdp_prog = READ_ONCE(ring->xdp_prog);

    while (likely(work_done < budget)) {
        struct mynic_rx_desc *desc = mynic_get_rx_desc(ring);
        struct xdp_buff xdp;
        struct sk_buff *skb;

        if (!desc)
            break;

        /* DMA 동기화: 디바이스 → CPU */
        dma_sync_single_range_for_cpu(ring->dev,
            page_pool_get_dma_addr(desc->page),
            desc->offset, desc->len, DMA_FROM_DEVICE);

        if (xdp_prog) {
            /* ① xdp_buff 초기화: 페이지 → xdp_buff 변환 */
            xdp_init_buff(&xdp, PAGE_SIZE, &ring->xdp_rxq);
            xdp_prepare_buff(&xdp,
                page_address(desc->page) + desc->offset,
                ring->rx_headroom,  /* XDP headroom: 보통 256B */
                desc->len, true);

            /* ② XDP 프로그램 실행 (BPF JIT 코드 호출) */
            act = bpf_prog_run_xdp(xdp_prog, &xdp);

            switch (act) {
            case XDP_PASS:
                /* ③ 정상 경로: sk_buff 생성 → GRO/스택 전달 */
                break;  /* 아래 skb 생성으로 진행 */

            case XDP_TX:
                /* ④ 동일 NIC으로 재전송 */
                if (mynic_xmit_xdp_ring(ring, &xdp))
                    goto consumed;
                trace_xdp_exception(ring->netdev, xdp_prog, act);
                goto xdp_drop;

            case XDP_REDIRECT:
                /* ⑤ 다른 NIC/AF_XDP/cpumap으로 리다이렉트 */
                if (!xdp_do_redirect(ring->netdev, &xdp, xdp_prog)) {
                    xdp_xmit = true;
                    goto consumed;
                }
                goto xdp_drop;

            case XDP_ABORTED:
                /* ⑥ BPF 오류: 드롭 + xdp:xdp_exception 트레이스 */
                trace_xdp_exception(ring->netdev, xdp_prog, act);
                /* fallthrough */
            default: /* XDP_DROP */
xdp_drop:
                /* ⑦ 페이지를 page_pool로 즉시 반환 (sk_buff 미생성) */
                page_pool_recycle_direct(ring->page_pool, desc->page);
                goto consumed;
            }
        }

        /* XDP_PASS 또는 XDP 미설정: sk_buff 생성 */
        skb = napi_build_skb(page_address(desc->page) + desc->offset,
                              ring->rx_buf_len);
        if (unlikely(!skb))
            goto xdp_drop;

        /* XDP metadata → skb metadata 전파 */
        if (xdp_prog && xdp.data_meta < xdp.data)
            skb_metadata_set(skb, xdp.data - xdp.data_meta);

        skb_mark_for_recycle(skb);  /* page_pool 반환 표시 */
        napi_gro_receive(napi, skb);
consumed:
        work_done++;
    }

    /* poll() 종료 전: XDP_REDIRECT 버퍼 일괄 플러시 */
    if (xdp_xmit)
        xdp_do_flush();

    if (work_done < budget && napi_complete_done(napi, work_done))
        mynic_enable_rx_irq(ring);

    return work_done;
}

XDP 멀티버퍼 (Multi-buffer XDP)

Linux 6.0부터 XDP는 단일 페이지를 초과하는 프레임(점보 프레임, TSO 세그먼트 등)을 xdp_buff의 skb_shared_info frags 영역으로 처리할 수 있습니다. 이전에는 MTU > PAGE_SIZE인 패킷에 XDP를 적용할 수 없었습니다.

/* include/net/xdp.h — 간략화 */
/* XDP 멀티버퍼 구조: xdp_buff의 끝에 skb_shared_info가 위치 */
struct xdp_buff {
    void *data;           /* 선형 데이터 시작 */
    void *data_end;       /* 선형 데이터 끝 */
    void *data_meta;
    void *data_hard_start;
    struct xdp_rxq_info *rxq;
    struct xdp_txq_info *txq;
    u32 frame_sz;
    u32 flags;            /* XDP_FLAGS_HAS_FRAGS: 멀티버퍼 표시 */
};

/* 멀티버퍼 확인 */
static inline bool xdp_buff_has_frags(struct xdp_buff *xdp)
{
    return !!(xdp->flags & XDP_FLAGS_HAS_FRAGS);
}

/* 드라이버: 멀티버퍼 xdp_buff 구성 */
if (desc->len > ring->rx_buf_len) {
    /* 첫 페이지: 선형 데이터 */
    xdp_prepare_buff(&xdp, data, headroom, ring->rx_buf_len, true);

    /* 나머지 페이지들: frags로 추가 */
    struct skb_shared_info *sinfo = xdp_get_shared_info_from_buff(&xdp);
    sinfo->nr_frags = 0;

    for (frag_idx = 1; remaining > 0; frag_idx++) {
        skb_frag_fill_page_desc(&sinfo->frags[sinfo->nr_frags++],
                                next_page, 0, min(remaining, PAGE_SIZE));
        remaining -= PAGE_SIZE;
    }
    xdp.flags |= XDP_FLAGS_HAS_FRAGS;
}

/* BPF 헬퍼: 멀티버퍼 데이터 접근 */
/* bpf_xdp_load_bytes(ctx, offset, buf, len) - 프래그먼트 경계 투명 읽기
   bpf_xdp_store_bytes(ctx, offset, buf, len) - 프래그먼트 경계 투명 쓰기
   bpf_xdp_adjust_tail(ctx, delta)            - 프레임 크기 조정 */

XDP Hints / kfuncs (Linux 6.3+)

XDP hints는 NIC 하드웨어가 제공하는 메타데이터(RX 타임스탬프, 체크섬(Checksum), 해시 등)를 XDP 프로그램에서 직접 읽을 수 있게 하는 kfunc 기반 인터페이스입니다. 기존 data_meta 방식보다 타입 안전하고 표준화되어 있습니다.

/* include/net/xdp.h — 간략화 */
/* XDP hints kfunc 정의 (드라이버 측) */
struct xdp_metadata_ops {
    /* HW RX 타임스탬프 조회 */
    int (*xmo_rx_timestamp)(const struct xdp_md *ctx, u64 *timestamp);
    /* HW 해시값 조회 */
    int (*xmo_rx_hash)(const struct xdp_md *ctx, u32 *hash,
                        enum xdp_rss_hash_type *rss_type);
    /* VLAN 태그 조회 */
    int (*xmo_rx_vlan_tag)(const struct xdp_md *ctx,
                           __be16 *vlan_proto, u16 *vlan_tci);
};

/* BPF 프로그램에서 사용 (kfunc 호출) */
SEC("xdp")
int xdp_hints_prog(struct xdp_md *ctx)
{
    u64 timestamp;
    u32 hash;
    enum xdp_rss_hash_type hash_type;

    /* kfunc: 하드웨어 타임스탬프 읽기 (NIC가 지원하는 경우) */
    if (!bpf_xdp_metadata_rx_timestamp(ctx, &timestamp))
        bpf_printk("HW timestamp: %llu", timestamp);

    /* kfunc: RSS 해시 읽기 */
    if (!bpf_xdp_metadata_rx_hash(ctx, &hash, &hash_type))
        bpf_printk("RSS hash: 0x%x type: %d", hash, hash_type);

    return XDP_PASS;
}

cpumap: XDP를 통한 CPU 간 패킷 분산

BPF_MAP_TYPE_CPUMAP은 XDP에서 패킷을 다른 CPU로 리다이렉트하는 메커니즘으로, 소프트웨어 RPS의 XDP 버전입니다. NAPI poll()에서 XDP_REDIRECT를 통해 특정 CPU의 전용 큐로 패킷을 전송하며, 수신 CPU에서 sk_buff 생성과 스택 처리가 이루어집니다.

/* cpumap BPF 프로그램 */
struct {
    __uint(type, BPF_MAP_TYPE_CPUMAP);
    __uint(key_size, sizeof(u32));    /* CPU ID */
    __uint(value_size, sizeof(u32)); /* 큐 크기 */
    __uint(max_entries, 256);        /* 최대 CPU 수 */
} cpu_map SEC(".maps");

/* XDP에서 플로우 해시 기반 CPU 분산 */
SEC("xdp")
int xdp_cpumap_redirect(struct xdp_md *ctx)
{
    u32 cpu = (get_flow_hash(ctx)) % num_cpus;
    return bpf_redirect_map(&cpu_map, cpu, 0);
}

/* cpumap 내부 처리:
   1. 원래 CPU (NAPI poll 실행 CPU):
      - XDP_REDIRECT → cpumap 큐에 xdp_frame 인큐
      - ptr_ring 기반 lockless SPSC 큐
   2. 대상 CPU:
      - kthread가 ptr_ring에서 디큐
      - xdp_frame → sk_buff 변환
      - netif_receive_skb() 호출 → 정상 스택 경로
      - cpumap에서 2차 XDP 프로그램 실행 가능 (6.0+) */

devmap: XDP를 통한 NIC 간 포워딩

BPF_MAP_TYPE_DEVMAP/DEVMAP_HASH는 XDP에서 패킷을 다른 네트워크 인터페이스로 리다이렉트합니다. L2 스위칭, 로드 밸런싱, NAT 등에 사용됩니다.

/* devmap: 인터페이스 인덱스 → ifindex 매핑 */
struct {
    __uint(type, BPF_MAP_TYPE_DEVMAP_HASH);
    __uint(key_size, sizeof(u32));
    __uint(value_size, sizeof(struct bpf_devmap_val));
    __uint(max_entries, 64);
} tx_port SEC(".maps");

/* bpf_devmap_val: devmap 엔트리에 2차 XDP 프로그램 부착 가능 */
struct bpf_devmap_val {
    __u32 ifindex;    /* 대상 인터페이스 */
    __u32 bpf_prog_fd; /* 선택적: TX 시 실행할 XDP 프로그램 */
};

/* L2 포워딩 예: MAC 주소 기반 출력 포트 결정 */
SEC("xdp")
int xdp_l2_forward(struct xdp_md *ctx)
{
    void *data_end = (void *)(long)ctx->data_end;
    void *data     = (void *)(long)ctx->data;
    struct ethhdr *eth = data;

    if (data + sizeof(*eth) > data_end)
        return XDP_DROP;

    /* FDB(MAC 테이블) 조회 → 출력 포트 결정 */
    u32 *port = bpf_map_lookup_elem(&fdb_map, eth->h_dest);
    if (!port)
        return XDP_PASS;  /* 알 수 없는 MAC → 커널 스택으로 */

    return bpf_redirect_map(&tx_port, *port, 0);
}

XDP 성능 비교

NAPI poll() 내 XDP 실행 타이밍

XDP와 page_pool 연동

XDP는 page_pool과 밀접하게 연동됩니다. XDP_DROP 시 페이지를 page_pool로 즉시 재활용하고, XDP_TX/XDP_REDIRECT에서는 페이지 소유권이 전달 대상으로 이동한 뒤 TX 완료 시점에 page_pool로 반환됩니다.

드라이버 구현 패턴

완전한 NAPI 드라이버 예제

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
#include <linux/netdevice.h>
#include <linux/etherdevice.h>
#include <linux/pci.h>
#include <net/page_pool/api.h>

#define MYNIC_RX_DESC_NUM   256
#define MYNIC_TX_DESC_NUM   256
#define MYNIC_NAPI_WEIGHT  64

struct mynic_rx_ring {
    struct napi_struct   napi;
    struct mynic_hw     *hw;
    struct pci_dev      *pdev;
    struct page_pool    *page_pool;  /* page_pool 통합 */
    struct xdp_rxq_info  xdp_rxq;    /* XDP 큐 정보 */
    void                *desc_base;
    u16                  next_to_clean;
    u16                  next_to_alloc;
    u32                  queue_idx;
};

page_pool 통합

page_pool은 NAPI 전용 고성능 페이지 할당자(Page Allocator)입니다. DMA 재매핑 없이 페이지를 재활용하여 수신 경로의 메모리 할당 오버헤드를 대폭 줄입니다.

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* page_pool 생성 및 NAPI 연결 */
static int mynic_setup_page_pool(struct mynic_rx_ring *ring)
{
    struct page_pool_params pp_params = {
        .order     = 0,         /* 4K 페이지 */
        .flags     = PP_FLAG_DMA_MAP | PP_FLAG_DMA_SYNC_DEV,
        .pool_size = MYNIC_RX_DESC_NUM,
        .nid       = dev_to_node(&ring->pdev->dev),
        .dev       = &ring->pdev->dev,
        .napi      = &ring->napi,    /* NAPI와 page_pool 연결 */
        .dma_dir   = DMA_FROM_DEVICE,
        .offset    = NET_SKB_PAD,
        .max_len   = PAGE_SIZE - NET_SKB_PAD,
    };

    ring->page_pool = page_pool_create(&pp_params);
    return PTR_ERR_OR_ZERO(ring->page_pool);
}

/* RX 링 버퍼 할당: page_pool에서 페이지 가져오기 */
static int mynic_alloc_rx_buf(struct mynic_rx_ring *ring)
{
    struct page *page;
    dma_addr_t   dma;

    /* page_pool에서 DMA 매핑된 페이지 할당 (캐시에서 재활용) */
    page = page_pool_alloc_pages(ring->page_pool, GFP_ATOMIC | __GFP_NOWARN);
    if (unlikely(!page))
        return -ENOMEM;

    dma = page_pool_get_dma_addr(page);

    /* 디스크립터에 DMA 주소 등록 */
    mynic_set_rx_dma(ring, ring->next_to_alloc, dma);
    ring->pages[ring->next_to_alloc] = page;
    ring->next_to_alloc = (ring->next_to_alloc + 1) % MYNIC_RX_DESC_NUM;
    return 0;
}

/* SKB 생성 후 page_pool 재활용 마킹 */
static inline void mynic_build_skb(struct mynic_rx_ring *ring,
                                    struct page *page, u16 len)
{
    struct sk_buff *skb;

    skb = build_skb(page_address(page) + NET_SKB_PAD, PAGE_SIZE);
    if (unlikely(!skb)) {
        page_pool_recycle_direct(ring->page_pool, page);
        return;
    }

    skb_put(skb, len);
    /* skb_mark_for_recycle: sk_buff 해제 시 자동으로 page_pool에 반환 */
    skb_mark_for_recycle(skb);

    napi_gro_receive(&ring->napi, skb);
}

TX 완료 처리를 같은 poll()에서 처리하는 패턴

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* TX 완료와 RX 수신을 동일 poll()에서 처리 (인터럽트 절약) */
static int mynic_poll(struct napi_struct *napi, int budget)
{
    struct mynic_rx_ring *rx_ring =
        container_of(napi, struct mynic_rx_ring, napi);
    struct mynic_tx_ring *tx_ring = rx_ring->hw->tx_rings[rx_ring->queue_idx];
    int work_done = 0;
    bool tx_cleaned;

    /* 1. TX 완료 처리 먼저 (TX ring 공간 확보) */
    tx_cleaned = mynic_clean_tx_ring(tx_ring);

    /* 2. RX 패킷 처리 */
    while (work_done < budget) {
        struct page *page;
        u16 len;

        if (!mynic_get_rx_page(rx_ring, &page, &len))
            break;

        mynic_build_skb(rx_ring, page, len);
        work_done++;
    }

    /* 3. TX 완료 후 netdev_tx_completed_queue() 호출 */
    if (tx_cleaned)
        netif_tx_wake_all_queues(rx_ring->hw->netdev);

    if (work_done < budget && napi_complete_done(napi, work_done))
        mynic_enable_irq(rx_ring);

    return work_done;
}

에러 처리와 카운터 관리

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* 드라이버 통계 구조체 */
struct mynic_stats {
    u64 rx_packets;
    u64 rx_bytes;
    u64 rx_dropped;        /* 소프트웨어 드롭 (예: skb 할당 실패) */
    u64 rx_csum_errors;    /* 체크섬 오류 패킷 수 */
    u64 rx_missed;         /* HW 링 버퍼 오버플로우 (NIC 통계) */
    u64 rx_gro_packets;    /* GRO로 병합된 패킷 수 */
};

/* poll() 내 에러 처리 패턴 */
static void mynic_process_rx_desc(struct mynic_rx_ring *ring,
                                   struct mynic_rx_desc *desc)
{
    struct mynic_stats *stats = ring->stats;

    /* 체크섬 에러 감지 */
    if (unlikely(mynic_has_csum_error(desc))) {
        stats->rx_csum_errors++;
        /* CHECKSUM_NONE: 스택이 직접 체크섬 검증 수행 */
        ring->current_skb->ip_summed = CHECKSUM_NONE;
    }

    /* 드롭 처리: skb 할당 실패 */
    if (unlikely(!ring->current_skb)) {
        stats->rx_dropped++;
        return;
    }

    /* HW missed 카운터 주기적 폴링 (ethtool -S 출력용) */
    stats->rx_missed += mynic_read_rx_missed(ring->hw);
}

netdev_alloc_skb_ip_align() vs build_skb() vs napi_alloc_skb() 비교

net_device_ops에서 napi_enable/disable

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
static int mynic_open(struct net_device *netdev)
{
    struct mynic_hw *hw = netdev_priv(netdev);
    int i;

    for (i = 0; i < hw->num_queues; i++) {
        napi_enable(&hw->rx_rings[i].napi);
        request_irq(..., mynic_msix_rx, 0, ..., &hw->rx_rings[i]);
    }
    netif_carrier_on(netdev);
    return 0;
}

static int mynic_stop(struct net_device *netdev)
{
    struct mynic_hw *hw = netdev_priv(netdev);
    int i;

    netif_carrier_off(netdev);
    for (i = 0; i < hw->num_queues; i++) {
        free_irq(hw->msix_entries[i].vector, &hw->rx_rings[i]);
        napi_disable(&hw->rx_rings[i].napi);
    }
    return 0;
}

실전 드라이버 참조 — ICE NAPI 패턴

Intel E810(ice) 드라이버는 큐 벡터-NAPI 1:1 매핑 구조를 사용하며, 각 큐 벡터(ice_q_vector)가 하나의 NAPI 인스턴스를 소유합니다. 이 패턴은 앞서 설명한 mynic 예제의 실전 적용 사례입니다.

/* drivers/net/ethernet/intel/ice/ice_txrx.c — 간략화 */
/* ICE NAPI poll 구조 (단순화) */
int ice_napi_poll(struct napi_struct *napi, int budget)
{
    struct ice_q_vector *q_vector =
        container_of(napi, struct ice_q_vector, napi);
    bool clean_complete = true;
    int budget_per_ring;

    /* TX 완료 처리 (budget 무관) */
    ice_for_each_tx_ring(tx_ring, q_vector->tx) {
        if (!ice_clean_tx_irq(tx_ring, budget))
            clean_complete = false;
    }

    /* RX 처리 (budget 분배) */
    budget_per_ring = max(budget / q_vector->num_ring_rx, 1);
    ice_for_each_rx_ring(rx_ring, q_vector->rx) {
        int cleaned = ice_clean_rx_irq(rx_ring, budget_per_ring);
        if (cleaned >= budget_per_ring)
            clean_complete = false;
    }

    /* 완료 시: NAPI complete + Adaptive ITR 갱신 + IRQ 재활성화 */
    if (clean_complete && napi_complete_done(napi, budget)) {
        ice_update_itr(q_vector);        /* ← Adaptive ITR 핵심 */
        ice_enable_interrupt(q_vector);
    }
    return min(budget, work_done);
}

TX 전용 NAPI 패턴 (netif_napi_add_tx())

netif_napi_add_tx()는 TX 완료 처리 전용 NAPI를 등록합니다. RX+TX 결합 NAPI에서 TX 완료가 RX budget을 소비하지 않도록 분리할 때 사용합니다. weight가 고정(NAPI_POLL_WEIGHT)이며, TX 전용이므로 NAPI_STATE_NO_BUSY_POLL이 자동 설정됩니다.

/* include/linux/netdevice.h */
static inline void
netif_napi_add_tx(struct net_device *dev,
                   struct napi_struct *napi,
                   int (*poll)(struct napi_struct *, int))
{
    netif_napi_add_weight(dev, napi, poll, NAPI_POLL_WEIGHT);
    set_bit(NAPI_STATE_NO_BUSY_POLL, &napi->state);
}

/* RX+TX 결합 vs RX/TX 분리 NAPI 비교:
 *
 * 결합 패턴 (ice, ixgbe 등):
 *   poll() {
 *     ice_clean_tx_irq();  // TX 완료 (budget 무관)
 *     ice_clean_rx_irq();  // RX 처리 (budget 사용)
 *   }
 *   장점: IRQ 1개, 컨텍스트 전환 최소
 *   단점: TX 지연이 RX에 영향
 *
 * 분리 패턴 (mlx5 등):
 *   rx_poll() { mlx5_rx_clean(); }
 *   tx_poll() { mlx5_tx_clean(); }  // netif_napi_add_tx()
 *   장점: RX/TX 독립 budget, 세밀한 제어
 *   단점: IRQ 2개 필요, 약간의 오버헤드 */

/* TX 전용 poll 구현 예 */
static int mynic_tx_poll(struct napi_struct *napi, int budget)
{
    struct mynic_tx_ring *ring =
        container_of(napi, struct mynic_tx_ring, napi);
    int cleaned = mynic_clean_tx(ring, budget);

    if (cleaned < budget && napi_complete_done(napi, cleaned))
        mynic_enable_tx_irq(ring);

    /* 정지된 TX 큐 재개 */
    if (netif_tx_queue_stopped(ring->txq) &&
        mynic_tx_avail(ring) > MYNIC_TX_WAKE_THRESH)
        netif_tx_wake_queue(ring->txq);

    return cleaned;
}

napi_consume_skb() TX 정리 패턴

TX 완료 경로에서 전송 완료된 SKB를 해제할 때는 반드시 napi_consume_skb()를 사용해야 합니다. budget 인자를 전달하면 per-CPU 캐시를 활용한 bulk free가 가능해져 dev_kfree_skb_any() 대비 최대 30% 성능 향상을 얻을 수 있습니다.

/* drivers/net/ethernet/ 기반 예제 — 간략화 */
/* TX 완료 정리 — 올바른 패턴 */
static bool mynic_clean_tx(struct mynic_tx_ring *ring, int budget)
{
    unsigned int total_bytes = 0, total_pkts = 0;
    u16 ntc = ring->next_to_clean;

    while (total_pkts < (unsigned int)budget) {
        struct mynic_tx_buf *buf = &ring->tx_buf[ntc];

        /* HW 소유 디스크립터는 건너뜀 */
        if (!mynic_tx_desc_done(ring, ntc))
            break;

        /* DMA 언매핑 */
        dma_unmap_single(&ring->pdev->dev,
                        buf->dma, buf->len, DMA_TO_DEVICE);

        total_bytes += buf->skb->len;
        total_pkts++;

        /* ★ napi_consume_skb: budget 전달 필수! */
        napi_consume_skb(buf->skb, budget);
        buf->skb = NULL;

        ntc = (ntc + 1) % ring->count;
    }

    ring->next_to_clean = ntc;

    /* 통계 갱신 (struct u64_stats_sync 보호) */
    u64_stats_update_begin(&ring->syncp);
    ring->stats.bytes += total_bytes;
    ring->stats.packets += total_pkts;
    u64_stats_update_end(&ring->syncp);

    return total_pkts < budget;
}

드라이버 메모리 전략 선택 가이드

napi_consume_skb() vs dev_kfree_skb_*() 비교표

성능 튜닝과 파라미터

핵심 커널 파라미터

인터럽트 코얼레싱(Interrupt Coalescing) 개요

인터럽트 코얼레싱은 NIC가 패킷 하나마다 즉시 IRQ를 발생시키지 않고, 일정 패킷 수(rx-frames) 또는 일정 시간(rx-usecs)이 경과한 후에 IRQ를 발생시키는 기법입니다. 하드웨어 타이머 동작 원리, Adaptive ITR 알고리즘, 드라이버별 구현 비교 등 내용은 아래 전용 섹션에서 다룹니다.

→ ITR(Interrupt Throttle Rate) 하드웨어 섹션 참조

ethtool 설정

# 인터럽트 코얼레싱 설정
ethtool -C eth0 rx-usecs 50 tx-usecs 50 rx-frames 16

# adaptive 코얼레싱 활성화
ethtool -C eth0 adaptive-rx on adaptive-tx on

# 현재 코얼레싱 설정 확인
ethtool -c eth0

# 링 버퍼 크기 조정
ethtool -G eth0 rx 4096 tx 4096

# RSS 큐 수 설정
ethtool -L eth0 combined 16

# GRO/LRO 활성화 확인
ethtool -k eth0 | grep -E "generic-receive-offload|large-receive-offload"

# GRO 비활성화 (디버깅 목적)
ethtool -K eth0 gro off

C-state 영향과 설정

CPU C-state가 깊을수록 절전 효과는 크지만, C-state 탈출(wakeup) 레이턴시가 증가합니다. 이는 첫 번째 인터럽트 처리 레이턴시에 직접 영향을 줍니다.

# C-state 제한: C1 이하만 허용 (저지연 필요 시)
echo 1 > /sys/devices/system/cpu/cpu0/cpuidle/state2/disable  # C3 비활성화
echo 1 > /sys/devices/system/cpu/cpu0/cpuidle/state3/disable  # C6 비활성화

# 전체 CPU에 적용 (bash 루프)
for cpu in /sys/devices/system/cpu/cpu*/cpuidle/state[2-9]; do
    echo 1 > $cpu/disable
done

# GRUB 설정으로 영속화: intel_idle.max_cstate=1 또는 processor.max_cstate=1

# tuned-adm으로 레이턴시 프로파일 적용
tuned-adm profile latency-performance

NUMA 최적화

# NIC의 NUMA 노드 확인
cat /sys/class/net/eth0/device/numa_node
# 결과 예: 1  → NUMA 노드 1에 연결된 NIC

# NUMA 노드 1의 CPU 목록 확인
numactl --hardware | grep "node 1 cpus"
# 결과 예: node 1 cpus: 8 9 10 11 12 13 14 15

# NUMA 노드 1의 CPU에만 IRQ 어피니티 설정
for irq in $(grep eth0 /proc/interrupts | cut -d: -f1); do
    echo ff00 > /proc/irq/$irq/smp_affinity  # CPU 8-15 = 0xff00
done

# 애플리케이션을 동일 NUMA 노드에서 실행
numactl --cpunodebind=1 --membind=1 ./server_app

시나리오별 튜닝 가이드

sysctl -w net.core.netdev_budget=1200
sysctl -w net.core.netdev_budget_usecs=16000
sysctl -w net.core.gro_flush_timeout=200000
sysctl -w net.core.napi_defer_hard_irqs=128
ethtool -C eth0 rx-usecs 100 adaptive-rx on
ethtool -G eth0 rx 8192

sysctl -w net.core.netdev_budget=64
sysctl -w net.core.netdev_budget_usecs=1000
sysctl -w net.core.gro_flush_timeout=0
sysctl -w net.core.busy_poll=50
sysctl -w net.core.busy_read=50
ethtool -C eth0 rx-usecs 0 rx-frames 1
ethtool -K eth0 gro off

# GRUB: isolcpus=8-15 nohz_full=8-15 rcu_nocbs=8-15
for i in $(seq 0 7); do
  irq=$(grep eth0-rx-$i /proc/interrupts | awk '{print $1}' | tr -d ':')
  echo $((1 << (i+8))) > /proc/irq/$irq/smp_affinity
done
echo 1 > /sys/class/net/eth0/threaded

sysctl 영속화 설정 파일

# /etc/sysctl.d/99-napi-tuning.conf (고처리량 서버용)
cat > /etc/sysctl.d/99-napi-tuning.conf <<'EOF'
# NAPI 버짓 및 처리량 튜닝
net.core.netdev_budget = 1200
net.core.netdev_budget_usecs = 16000
net.core.netdev_max_backlog = 10000

# GRO 타임아웃 (100μs)
net.core.gro_flush_timeout = 100000

# IRQ 지연 (폴링 모드 유지)
net.core.napi_defer_hard_irqs = 64

# 소켓 버퍼 크기 확장
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
net.ipv4.tcp_rmem = 4096 87380 134217728
net.ipv4.tcp_wmem = 4096 65536 134217728

# TCP 오프로드 최적화
net.ipv4.tcp_timestamps = 1
net.ipv4.tcp_sack = 1
EOF

# 즉시 적용
sysctl -p /etc/sysctl.d/99-napi-tuning.conf

napi_defer_hard_irqs와 gro_flush_timeout 최적화

이 두 파라미터를 함께 사용하면 IRQ를 더 오래 비활성화된 상태로 유지하여 패킷 배치 크기를 키울 수 있습니다.

• napi_defer_hard_irqs=64: NAPI poll이 64번 완료될 때까지 IRQ 재활성화 지연
• gro_flush_timeout=100000: 100μs 주기로 GRO 버퍼 강제 flush
• 이 조합은 실질적으로 NAPI를 항상 폴링 모드로 유지하는 효과 (유사 DPDK)

NAPI 튜닝 파라미터 상호작용 매트릭스

NAPI 관련 파라미터들은 서로 밀접하게 연관되어 있습니다. 하나를 변경하면 다른 파라미터의 효과가 달라지므로 전체적으로 이해해야 합니다.

워크로드별 추천 설정

ITR(Interrupt Throttle Rate) 하드웨어

ITR 하드웨어 타이머 동작 원리

ITR(Interrupt Throttle Rate)은 NIC 인터럽트 컨트롤러(Interrupt Controller) 내부의 하드웨어 타이머로, 이벤트(패킷 수신/송신 완료) 발생 후 인터럽트 어서션(assertion)을 일정 시간 지연시키는 메커니즘입니다. ethtool의 rx-usecs / tx-usecs 파라미터가 이 HW 레지스터(Register)에 직접 매핑됩니다.

타이머 생명주기:

1. 패킷 도착 → NIC가 DMA로 링 버퍼(Ring Buffer)에 디스크립터 기록
2. DMA 완료 → ITR 타이머 카운트다운 시작
3. 타이머 만료 → MSI-X 인터럽트 어서션 → CPU에 IRQ 전달

타이머 재시작(Reboot) 정책:

프레임 카운트 임계값: rx-frames 값이 설정된 경우, 누적 패킷 수가 임계값에 도달하면 타이머 만료 이전이라도 즉시 인터럽트를 발생시킵니다. 이는 버스트 트래픽에서 지연 시간의 상한을 보장하는 안전장치 역할을 합니다.

ITR과 NAPI 상태 전이 연동

ITR은 하드웨어 계층에서 인터럽트 빈도를 제한하고, NAPI는 소프트웨어 계층에서 인터럽트를 폴링으로 전환합니다. 두 메커니즘은 상호 보완적으로 동작합니다.

연동 흐름:

1. ITR 타이머 만료 → MSI-X 인터럽트 어서션
2. IRQ 핸들러 → napi_schedule() 호출, IRQ 비활성화
3. NAPI poll 루프 → 링 버퍼에서 패킷 배치 처리
4. napi_complete_done() → poll 완료, ice_update_itr()로 다음 ITR 값 계산
5. IRQ 재활성화 → 다음 ITR 타이머 사이클 시작

Adaptive ITR 알고리즘 상세

Adaptive ITR은 드라이버가 NAPI poll 완료 시점에서 수집한 통계(처리된 바이트 수, 패킷 수)를 기반으로 다음 인터럽트 간격을 동적으로 결정하는 알고리즘입니다. ICE 드라이버의 ice_update_itr() 함수를 예로 살펴봅니다.

3가지 동작 모드:

rx-usecs-high 파라미터가 설정된 경우, Adaptive 알고리즘이 산출한 값이 이 상한을 초과하지 않도록 바운딩됩니다. HW 타이머 해상도가 4μs이므로 실제 적용값은 항상 4의 배수로 반올림됩니다.

드라이버별 ITR 구현 비교

각 NIC 드라이버는 서로 다른 Adaptive ITR 알고리즘을 구현합니다. 아래 표는 주요 드라이버의 구현 특성을 비교합니다.

# ICE per-queue Adaptive ITR 설정 예제
# 큐 0~3에 Adaptive 활성화 + 상한 100μs
ethtool --per-queue eth0 queue_mask 0xf --coalesce adaptive-rx on rx-usecs-high 100

# 특정 큐만 고정 코얼레싱 (저지연 전용 큐)
ethtool --per-queue eth0 queue_mask 0x10 --coalesce adaptive-rx off rx-usecs 8

디버깅과 모니터링

/proc/net/softnet_stat 해석

# CPU별 소프트넷 통계 확인
cat /proc/net/softnet_stat

출력 예시 (각 행이 하나의 CPU):

00094e79 00000000 00000004 00000000 00000000 00000000 00000000 00000000 00000000 00000000 00000000 00000000 00000000
00036f22 00000000 00000001 ...

/proc/net/softnet_stat 파싱 스크립트

#!/bin/bash
# softnet_stat_parse.sh: /proc/net/softnet_stat 열 이름 매핑하여 출력

COLS=(total dropped time_squeeze throttled
      irq_poll cpu_collision received_rps
      flow_limit_count backlog_drops filter_drops
      unknown1 unknown2 unknown3)

echo "=== /proc/net/softnet_stat 분석 ==="
cpu=0
while IFS= read -r line; do
    echo -n "CPU $cpu: "
    i=0
    for val in $line; do
        dec=$((16#$val))
        if [ $dec -gt 0 ] && [ $i -lt ${#COLS[@]} ]; then
            echo -n "${COLS[$i]}=$dec "
        fi
        i=$((i+1))
    done
    echo
    cpu=$((cpu+1))
done < /proc/net/softnet_stat

# 경고 감지
echo
echo "=== 경고 감지 ==="
cpu=0
while IFS= read -r line; do
    vals=($line)
    dropped=$((16#${vals[1]}))
    squeeze=$((16#${vals[2]}))
    [ $dropped -gt 0 ] && echo "  CPU $cpu: 드롭 $dropped개 (netdev_max_backlog 증가 권장)"
    [ $squeeze -gt 1000 ] && echo "  CPU $cpu: time_squeeze $squeeze회 (netdev_budget 증가 권장)"
    cpu=$((cpu+1))
done < /proc/net/softnet_stat

커널 트레이스포인트 목록

perf stat으로 소프트IRQ 관련 PMU 이벤트 수집

# softIRQ 전용 CPU 사이클 측정
perf stat -e cycles:k,instructions:k,cache-misses \
          -a --per-cpu sleep 5 2>&1 | grep CPU

# net_rx_action() 함수 프로파일링
perf record -g -F 999 -e cycles:k -a -- sleep 10
perf report --stdio --dsos vmlinux | grep -A 20 net_rx_action

# softIRQ 처리 시간 측정 (irq 이벤트 활용)
perf stat -e softirqs/NET_RX/ -a sleep 5

# NAPI poll CPU 점유율 확인 (함수별 분류)
perf top -e cycles:k --stdio -d 5 | grep -E "napi|gro|net_rx"

bpftrace 레시피

# 레시피 1: NAPI poll 레이턴시 히스토그램 (스케줄 → poll 시작까지)
bpftrace -e '
kprobe:__napi_schedule {
    @start[arg0] = nsecs;
}
kprobe:napi_poll / @start[arg0] / {
    $lat_us = (nsecs - @start[arg0]) / 1000;
    @sched_to_poll_us = hist($lat_us);
    delete(@start[arg0]);
}
interval:s:10 {
    printf("=== NAPI 스케줄 → poll 레이턴시 (μs) ===\n");
    print(@sched_to_poll_us);
    clear(@sched_to_poll_us);
}'

# 레시피 2: GRO 병합률 측정 (초당 GRO_MERGED vs GRO_NORMAL)
bpftrace -e '
tracepoint:net:napi_gro_receive_exit {
    if (args->ret == 0) @gro_merged = count();       // GRO_MERGED
    else if (args->ret == 3) @gro_normal = count();  // GRO_NORMAL
}
interval:s:1 {
    $total = @gro_merged + @gro_normal;
    if ($total > 0) {
        printf("GRO 병합률: %d/%d (%d%%)\n",
               @gro_merged, $total, @gro_merged * 100 / $total);
    }
    clear(@gro_merged); clear(@gro_normal);
}'

# 레시피 3: 패킷 드롭 위치 추적 (이유별 분류)
bpftrace -e '
tracepoint:skb:kfree_skb {
    @drop_reason[args->reason] = count();
}
interval:s:5 {
    printf("=== 패킷 드롭 이유별 통계 ===\n");
    print(@drop_reason);
    clear(@drop_reason);
}'

ethtool 통계

# NIC별 상세 통계 (NAPI 관련 포함)
ethtool -S eth0 | grep -E "rx_missed|rx_dropped|rx_csum|gro"

# 주요 카운터:
#   rx_missed_errors: NIC 버퍼 오버플로우 (ring 크기 늘리기)
#   rx_dropped:       소프트웨어 드롭
#   rx_gro_packets:   GRO로 병합된 패킷 수
#   rx_gro_chunks:    GRO 병합 결과 청크 수

perf와 ftrace 활용

# NAPI poll CPU 사용 추적
perf record -g -e cycles:k -- sleep 5
perf report --stdio | grep -A5 net_rx_action

# ftrace로 NAPI 이벤트 추적
cd /sys/kernel/debug/tracing
echo napi:napi_poll > set_event
cat trace

# 함수 그래프 추적 (poll() 내부 시간 측정)
echo function_graph > current_tracer
echo mynic_poll > set_graph_function
cat trace_pipe

NAPI 상태 진단 플로우

일반적인 NAPI 버그 패턴

lockdep으로 NAPI 관련 락 순서 검증

# lockdep 활성화 커널 빌드 옵션
# CONFIG_LOCK_STAT=y
# CONFIG_DEBUG_LOCK_ALLOC=y
# CONFIG_PROVE_LOCKING=y

# NAPI 락 순서 위반 감지 로그 확인
dmesg | grep -E "lockdep|WARNING.*napi|possible circular"

# NAPI 관련 락 통계 확인
cat /proc/lock_stat | grep -E "napi|softirq|bh"

/* drivers/net/ethernet/ 일반 예제 — 간략화 */
/* NAPI 락 사용 패턴: bh 컨텍스트에서만 접근해야 하는 자료구조 */

/* 올바른 패턴: softIRQ(BH) 컨텍스트에서 spin_lock_bh() 불필요 */
static int mynic_poll(struct napi_struct *napi, int budget)
{
    /* softIRQ 내부: local_bh_disable 상태이므로 spin_lock 충분 */
    spin_lock(&ring->lock);
    /* ... */
    spin_unlock(&ring->lock);
}

/* 잘못된 패턴: 프로세스 컨텍스트에서 BH를 비활성화하지 않고 접근 */
static void mynic_bad_access(void)
{
    /* 위험: softIRQ(NAPI poll)와 경합 가능 */
    spin_lock(&ring->lock);    /* spin_lock_bh() 사용해야 함 */
    /* ... */
    spin_unlock(&ring->lock);
}

일반적인 문제 진단

RPS/RFS와 NAPI 연동

단일 큐 NIC에서도 소프트웨어 멀티큐를 구현하는 RPS(Receive Packet Steering)와 RFS(Receive Flow Steering)는 NAPI poll 후 패킷을 다른 CPU로 재분산합니다.

# RPS 활성화: eth0의 첫 번째 큐에서 모든 CPU로 분산
echo ff > /sys/class/net/eth0/queues/rx-0/rps_cpus

# RFS 활성화: 소켓이 실행 중인 CPU로 패킷 유도
echo 32768 > /proc/sys/net/core/rps_sock_flow_entries
echo 2048 > /sys/class/net/eth0/queues/rx-0/rps_flow_cnt

네트워크 성능 모니터링 스크립트

#!/bin/bash
# napi_monitor.sh: NAPI 관련 핵심 지표 실시간 모니터링

NIC=${1:-eth0}
INTERVAL=${2:-1}

prev_total=0; prev_dropped=0; prev_squeeze=0

while true; do
    clear
    echo "=== NAPI 성능 모니터: $NIC ($(date)) ==="
    echo

    # softnet_stat 합산
    total=0; dropped=0; squeeze=0
    while IFS= read -r line; do
        vals=($line)
        total=$((total + 16#${vals[0]}))
        dropped=$((dropped + 16#${vals[1]}))
        squeeze=$((squeeze + 16#${vals[2]}))
    done < /proc/net/softnet_stat

    echo "패킷 처리량:  $((total - prev_total)) pkt/s"
    echo "드롭:         $((dropped - prev_dropped)) pkt/s"
    echo "time_squeeze: $((squeeze - prev_squeeze)) /s"
    prev_total=$total; prev_dropped=$dropped; prev_squeeze=$squeeze

    echo
    echo "=== NIC 통계 ($NIC) ==="
    ethtool -S $NIC 2>/dev/null | \
        grep -E "rx_packets|rx_bytes|rx_missed|rx_dropped|rx_csum|rx_gro" | head -10

    echo
    echo "=== IRQ 분포 ==="
    grep $NIC /proc/interrupts | \
        awk '{printf "IRQ %s: total=%s\n", $1, $NF}' | head -8

    echo
    echo "=== CPU별 NET_RX softIRQ ==="
    cat /proc/softirqs | grep NET_RX

    sleep $INTERVAL
done

NAPI와 SR-IOV/컨테이너 환경

SR-IOV(Single Root I/O Virtualization) VF 드라이버도 PF와 동일한 NAPI 패턴을 따릅니다. 컨테이너 환경에서 NAPI 관련 설정은 호스트 커널이 관리합니다.

NAPI 관련 커널 설정 옵션

NAPI 성능 벤치마크

# pktgen: 커널 내장 패킷 생성기로 NIC 처리량 측정
modprobe pktgen

echo "add_device eth0" > /proc/net/pktgen/kpktgend_0
echo "clone_skb 1000" > /proc/net/pktgen/eth0
echo "pkt_size 64" > /proc/net/pktgen/eth0
echo "count 10000000" > /proc/net/pktgen/eth0
echo "start" > /proc/net/pktgen/pgctrl

# iperf3: TCP 처리량 측정 (GRO 효과 측정)
iperf3 -s -p 5201  # 서버
iperf3 -c server_ip -p 5201 -P 8 -t 10  # 클라이언트 (8스트림)

# sockperf: 초저지연 왕복 측정 (버지 폴링 효과)
sockperf server -i server_ip -p 11111  # 서버
sockperf ping-pong -i server_ip -p 11111 --time 10  # 클라이언트

# netperf: TCP RR 레이턴시 측정
netperf -H server_ip -t TCP_RR -l 30 -- -r 1,1

NAPI 트러블슈팅 플로우차트

NAPI 관련 문제가 발생했을 때 순차적으로 확인해야 할 진단 절차입니다.

자주 묻는 질문 (FAQ)

NAPI Backlog과 process_backlog()

softnet_data.backlog의 역할

모든 CPU는 softnet_data 내에 기본(default) NAPI 인스턴스인 backlog을 갖고 있습니다. 이 backlog NAPI는 다음 상황에서 사용됩니다:

• RPS(Receive Packet Steering)가 다른 CPU로 패킷을 전달할 때
• 드라이버가 NAPI를 사용하지 않고 netif_rx()를 직접 호출할 때
• loopback 인터페이스의 패킷 수신
• netif_receive_skb()가 RPS 활성 상태에서 호출될 때

/* net/core/dev.c - per-CPU backlog NAPI 초기화 */
static int __init net_dev_init(void)
{
    int i;
    for_each_possible_cpu(i) {
        struct softnet_data *sd = &per_cpu(softnet_data, i);

        INIT_LIST_HEAD(&sd->poll_list);
        skb_queue_head_init(&sd->input_pkt_queue);

        /* backlog NAPI 등록:
           poll 함수 = process_backlog
           weight = NAPI_POLL_WEIGHT * 2 = 128
           (일반 드라이버 NAPI보다 2배 높은 버짓) */
        init_gro_hash(&sd->backlog);
        sd->backlog.poll = process_backlog;
        sd->backlog.weight = weight_p;  /* 기본 64 */

        /* backlog은 napi_hash에 등록되지 않음 (busy poll 불가) */
    }
    /* NET_RX_SOFTIRQ 핸들러 등록 */
    open_softirq(NET_RX_SOFTIRQ, net_rx_action);
    return 0;
}

process_backlog() 내부 구현

process_backlog()은 backlog NAPI의 poll 함수로, input_pkt_queue에서 sk_buff를 꺼내 __netif_receive_skb()로 전달합니다.

/* net/core/dev.c */
static int process_backlog(struct napi_struct *napi, int quota)
{
    struct softnet_data *sd = container_of(napi, struct softnet_data, backlog);
    int work = 0;

    /* ① input_pkt_queue → process_queue로 일괄 이동
       (입력 큐는 다른 CPU에서도 접근하므로 락 보호 필요) */
    while (work < quota) {
        struct sk_buff *skb;
        unsigned int qlen;

        local_irq_disable();
        rps_lock(sd);
        /* input_pkt_queue의 내용을 process_queue로 스플라이스 */
        qlen = skb_queue_len(&sd->input_pkt_queue);
        if (qlen) {
            skb_queue_splice_tail_init(&sd->input_pkt_queue,
                                      &sd->process_queue);
        }
        rps_unlock(sd);
        local_irq_enable();

        /* ② process_queue에서 하나씩 꺼내 처리 */
        while ((skb = __skb_dequeue(&sd->process_queue))) {
            __netif_receive_skb(skb);
            work++;
            if (work >= quota)
                return work;
        }

        /* ③ 큐가 비었으면 NAPI 완료 */
        if (!qlen) {
            napi_complete_done(napi, work);
            return work;
        }
    }
    return work;
}

netif_rx()의 내부 흐름

NAPI를 사용하지 않는 레거시 드라이버나 특수한 경로(loopback, tun/tap)에서 netif_rx()가 호출됩니다. 이 함수는 backlog 큐에 skb를 인큐(Enqueue)하고 backlog NAPI를 스케줄합니다.

/* net/core/dev.c */
int netif_rx(struct sk_buff *skb)
{
    struct softnet_data *sd;
    int ret;

    sd = this_cpu_ptr(&softnet_data);

    rps_lock(sd);
    /* input_pkt_queue 길이 확인 (netdev_max_backlog 초과 시 드롭) */
    if (skb_queue_len(&sd->input_pkt_queue) >= netdev_max_backlog) {
        sd->dropped++;  /* /proc/net/softnet_stat 열 2 */
        rps_unlock(sd);
        kfree_skb(skb);
        return NET_RX_DROP;
    }

    /* 큐에 인큐 */
    __skb_queue_tail(&sd->input_pkt_queue, skb);

    /* backlog NAPI 스케줄 */
    if (napi_schedule_prep(&sd->backlog))
        ____napi_schedule(sd, &sd->backlog);
    rps_unlock(sd);

    return NET_RX_SUCCESS;
}

backlog 관련 sysctl 파라미터

RPS/RFS와 NAPI 통합

RPS(Receive Packet Steering) 개요

RPS는 소프트웨어 기반 RSS(Receive Side Scaling)입니다. 하드웨어 RSS를 지원하지 않는 NIC이나 단일 큐 NIC에서 패킷 처리를 여러 CPU로 분산합니다. NAPI poll()에서 수신한 패킷의 플로우 해시를 계산하고 rps_cpu_mask에 따라 대상 CPU를 결정합니다.

/* net/core/dev.c - RPS CPU 선택 */
static int get_rps_cpu(struct net_device *dev,
                      struct sk_buff *skb,
                      struct rps_dev_flow **rflowp)
{
    struct netdev_rx_queue *rxqueue;
    struct rps_map *map;
    struct rps_sock_flow_table *sock_flow_table;
    u32 hash, next_cpu, ident;
    int cpu = -1;

    /* 1. 플로우 해시 계산 (skb->hash 또는 Toeplitz 해시) */
    hash = skb_get_hash(skb);
    if (!hash)
        return -1;

    /* 2. RPS 맵에서 후보 CPU 선택 */
    rxqueue = &dev->_rx[skb_get_rx_queue(skb)];
    map = rcu_dereference(rxqueue->rps_map);
    if (map) {
        next_cpu = map->cpus[reciprocal_scale(hash, map->len)];
    }

    /* 3. RFS 활성 시: 소켓이 마지막으로 실행된 CPU 우선 */
    sock_flow_table = rcu_dereference(rps_sock_flow_table);
    if (sock_flow_table) {
        ident = sock_flow_table->ents[hash & sock_flow_table->mask];
        /* 소켓의 CPU가 RPS 맵에 포함되면 그 CPU 사용 */
        if (cpu_online(ident & rps_cpu_mask))
            cpu = ident & rps_cpu_mask;
    }

    return cpu;
}

/* RPS가 선택한 CPU의 backlog에 패킷 전달 */
static int enqueue_to_backlog(struct sk_buff *skb, int cpu,
                              unsigned int *qtail)
{
    struct softnet_data *sd = &per_cpu(softnet_data, cpu);

    rps_lock(sd);
    if (skb_queue_len(&sd->input_pkt_queue) <= netdev_max_backlog) {
        __skb_queue_tail(&sd->input_pkt_queue, skb);
        /* 대상 CPU에 IPI(Inter-Processor Interrupt) 전송 */
        napi_schedule_rps(sd);  /* → __napi_schedule(&sd->backlog) */
        rps_unlock(sd);
        return NET_RX_SUCCESS;
    }
    sd->dropped++;
    rps_unlock(sd);
    kfree_skb(skb);
    return NET_RX_DROP;
}

RFS(Receive Flow Steering)

RFS는 RPS를 확장하여 패킷을 해당 소켓을 처리하는 CPU로 전달합니다. 이를 통해 L1/L2 캐시 히트율을 극대화합니다.

/* RFS 소켓 플로우 테이블 */
struct rps_sock_flow_table {
    u32 mask;    /* entries - 1 (2의 거듭제곱 - 1) */
    u32 ents[];  /* 해시 → CPU 매핑 (동적 갱신) */
};

/* 소켓이 recvmsg() 호출 시 자동 갱신:
   inet_recvmsg() → sock_rps_record_flow()
   → rps_sock_flow_table[hash] = 현재 CPU */

/* aRFS(Accelerated RFS): 하드웨어 지원 RFS
   NIC가 ntuple 필터로 플로우를 특정 큐로 직접 스티어링
   드라이버 콜백: ndo_rx_flow_steer() */

RPS/RFS 설정

# RPS: 패킷 처리 CPU 지정 (비트마스크)
# eth0의 큐 0에서 CPU 0-7로 분산
echo ff > /sys/class/net/eth0/queues/rx-0/rps_cpus

# RPS 플로우 해시 테이블 크기 (전역)
echo 32768 > /proc/sys/net/core/rps_sock_flow_entries

# per-큐 RFS 플로우 테이블 크기
echo 4096 > /sys/class/net/eth0/queues/rx-0/rps_flow_cnt

# 확인: RPS가 활성화되면 softnet_stat의 received_rps 열 증가
watch -n1 cat /proc/net/softnet_stat

# aRFS 활성화 (드라이버 지원 필요)
ethtool -K eth0 ntuple on
echo 1 > /proc/sys/net/core/rps_sock_flow_entries

RPS vs RSS vs XDP cpumap 비교

네트워크 스택 전달 경로 (NAPI → 소켓)

NAPI에서 소켓까지의 전체 패킷 경로

NAPI poll()에서 수신된 패킷이 최종적으로 애플리케이션의 소켓 버퍼에 도달하기까지의 전체 경로입니다. 각 단계에서 어떤 처리가 이루어지는지 이해하면 성능 병목(Bottleneck)을 정확히 진단할 수 있습니다.

단계별 상세 설명

GRO flush에서 프로토콜 핸들러까지

/* net/core/gro.c, net/core/dev.c — 간략화 */
/* GRO flush 후 패킷 전달 경로 */

/* 1. napi_gro_flush() → GRO 병합된 skb를 상위로 전달 */
void napi_gro_flush(struct napi_struct *napi, bool flush_old)
{
    /* gro_hash 버킷 순회: 병합된 skb들을 napi->rx_list에 추가 */
    for (i = 0; i < GRO_HASH_BUCKETS; i++) {
        list_for_each_entry_safe(skb, ...) {
            __napi_gro_flush_chain(napi, i, flush_old);
        }
    }
}

/* 2. gro_normal_list() → 일괄 netif_receive_skb() */
static void gro_normal_list(struct napi_struct *napi)
{
    /* rx_list의 skb들을 배치로 전달 (GRO_NORMAL_BATCH=8) */
    netif_receive_skb_list_internal(&napi->rx_list);
    napi->rx_count = 0;
}

/* 3. __netif_receive_skb() 내부 */
static int __netif_receive_skb_core(struct sk_buff **pskb, ...)
{
    /* a) Generic XDP (드라이버 미지원 NIC용) */
    if (static_branch_unlikely(&generic_xdp_needed_key))
        do_xdp_generic(skb);

    /* b) TC ingress (tc filter, cls_bpf) */
    skb = sch_handle_ingress(skb, ...);

    /* c) ptype_all 핸들러 (tcpdump, AF_PACKET 등) */
    list_for_each_entry_rcu(ptype, &ptype_all, list)
        deliver_skb(skb, ptype, orig_dev);

    /* d) ptype 프로토콜 핸들러 디스패치 (L3) */
    /*    ETH_P_IP → ip_rcv()
         ETH_P_IPV6 → ipv6_rcv()
         ETH_P_ARP → arp_rcv() */
    deliver_ptype_list_skb(skb, ...);
}

TCP Fast Path와 NAPI 최적화

TCP의 fast path는 GRO로 병합된 대용량 세그먼트를 효율적으로 처리합니다. NAPI/GRO와 TCP fast path의 시너지가 현대 리눅스의 높은 TCP 처리량의 핵심입니다.

실제 드라이버 사례 연구

ixgbe (Intel 10GbE) — 멀티큐 NAPI의 교과서

ixgbe는 Intel 82599/X540 기반 10G NIC 드라이버로, 리눅스 NAPI 멀티큐 구현의 참조 모델입니다. 최대 64개의 RX/TX 큐를 지원하며, RSS + Adaptive ITR + Flow Director를 결합합니다.

/* drivers/net/ethernet/intel/ixgbe/ixgbe_main.c */

/* ixgbe의 NAPI 구조: q_vector가 NAPI + IRQ를 묶는 단위 */
struct ixgbe_q_vector {
    struct ixgbe_adapter *adapter;
    int                  v_idx;        /* 벡터 인덱스 */
    int                  cpu;          /* 어피니티 CPU */
    struct napi_struct   napi;         /* NAPI 인스턴스 */
    struct ixgbe_ring   *rx_ring;     /* 연결된 RX 링 */
    struct ixgbe_ring   *tx_ring;     /* 연결된 TX 링 */
    struct ixgbe_ring_container rx, tx;
    /* Adaptive ITR 상태 */
    u16                  itr;
    char                 name[IFNAMSIZ + 9]; /* "eth0-TxRx-0" */
};

/* ixgbe poll: RX + TX 동시 처리 */
static int ixgbe_poll(struct napi_struct *napi, int budget)
{
    struct ixgbe_q_vector *q_vector =
        container_of(napi, struct ixgbe_q_vector, napi);
    int per_ring_budget, work_done = 0;
    bool clean_complete = true;

    /* ① TX 링 정리 (TX 완료 인터럽트 처리) */
    ixgbe_for_each_ring(ring, q_vector->tx)
        if (!ixgbe_clean_tx_irq(q_vector, ring, budget))
            clean_complete = false;

    /* ② RX 버짓 분배: 여러 RX 링에 균등 분배 */
    if (q_vector->rx.count > 1)
        per_ring_budget = max(budget / q_vector->rx.count, 1);
    else
        per_ring_budget = budget;

    /* ③ 각 RX 링 폴링 */
    ixgbe_for_each_ring(ring, q_vector->rx) {
        int cleaned = ixgbe_clean_rx_irq(q_vector, ring, per_ring_budget);
        work_done += cleaned;
        if (cleaned >= per_ring_budget)
            clean_complete = false;
    }

    /* ④ NAPI 완료 + Adaptive ITR 갱신 */
    if (!clean_complete)
        return budget;  /* 버짓 소진: 재스케줄 */

    if (likely(napi_complete_done(napi, work_done)))
        ixgbe_irq_enable_queues(adapter, BIT(q_vector->v_idx));

    return min(work_done, budget - 1);
}

mlx5 (Mellanox/NVIDIA ConnectX) — 고성능 NAPI

mlx5는 ConnectX-4/5/6/7 시리즈 NIC 드라이버로, 100G+ 환경에서 NAPI + page_pool + XDP를 가장 적극적으로 활용합니다.

/* drivers/net/ethernet/mellanox/mlx5/core/en_rx.c */

/* mlx5의 핵심 특징:
   1. Completion Queue (CQ)를 NAPI와 1:1 매핑
   2. Striding RQ: 큰 페이지를 여러 패킷이 공유 (page_pool 효율 극대화)
   3. SHAMPO (Split Header And Mark Payload Offload): 헤더/페이로드 분리
   4. XDP_REDIRECT가 zero-copy AF_XDP과 통합 */

/* mlx5 poll: CQ 기반 배치 처리 */
static int mlx5e_napi_poll(struct napi_struct *napi, int budget)
{
    struct mlx5e_channel *c = container_of(napi, ...);
    int work_done = 0;
    bool busy = false;

    /* TX CQ 처리 (TX 완료) */
    for (i = 0; i < c->num_tc; i++)
        busy |= mlx5e_poll_tx_cq(&c->sq[i].cq, budget);

    /* XDP TX CQ 처리 */
    busy |= mlx5e_poll_xdpsq_cq(&c->xdpsq.cq);

    /* RX CQ 처리: 패킷 수신 + XDP 실행 */
    work_done = mlx5e_poll_rx_cq(&c->rq.cq, budget);
    busy |= (work_done == budget);

    /* AF_XDP: 별도 RX CQ 처리 */
    if (c->xskrq_active)
        busy |= mlx5e_poll_xsk_rx_cq(&c->xskrq.cq, budget);

    if (busy) {
        mlx5e_cq_arm(&c->rq.cq);  /* CQ doorbell */
        return budget;
    }

    if (napi_complete_done(napi, work_done))
        mlx5e_cq_arm(&c->rq.cq);

    return work_done;
}

virtio-net — 가상화 환경 NAPI

virtio-net은 KVM/QEMU 가상머신의 네트워크 드라이버로, 하이퍼바이저(Hypervisor)와 게스트 간 virtqueue를 통해 패킷을 교환합니다. NAPI + 멀티큐 + XDP를 지원합니다.

/* drivers/net/virtio_net.c */

/* virtio-net의 NAPI 특수성:
   1. virtqueue는 vring(공유 메모리 링)으로 구현
   2. 인터럽트 = virtqueue callback (호스트 → 게스트 통지)
   3. 배치 버퍼 할당으로 vmexit 횟수 최소화
   4. mergeable rx buffers: 가변 크기 패킷 효율적 처리 */

static int virtnet_poll(struct napi_struct *napi, int budget)
{
    struct receive_queue *rq =
        container_of(napi, struct receive_queue, napi);
    unsigned int received;

    /* XDP 활성 시: mergeable 대신 1:1 페이지 모드 */
    received = virtnet_receive(rq, budget, ...);

    /* 빈 버퍼 보충 (vring에 새 버퍼 추가) */
    if (rq->vq->num_free >= virtqueue_get_vring_size(rq->vq) / 2)
        try_fill_recv(rq->vi, rq, GFP_ATOMIC);

    if (received < budget) {
        napi_complete_done(napi, received);
        /* virtqueue 인터럽트 재활성화 */
        if (unlikely(!virtqueue_enable_cb_delayed(rq->vq)))
            virtqueue_napi_schedule(napi, rq->vq);
    }

    return received;
}

/* virtio-net 특유의 인터럽트 재활성화 패턴:
   virtqueue_enable_cb_delayed()는 "좀 더 기다렸다가 인터럽트"
   → 배치 처리 효율 향상 (HW ITR과 유사한 효과) */

드라이버별 NAPI 구현 비교

드라이버 NAPI 초기화 패턴 비교

/* drivers/net/ethernet/intel/ixgbe/ixgbe_main.c — 간략화 */
/* 패턴 1: ixgbe — q_vector 기반 (RX+TX 결합 인터럽트) */
for (v_idx = 0; v_idx < adapter->num_q_vectors; v_idx++) {
    q_vector = adapter->q_vector[v_idx];
    netif_napi_add(adapter->netdev, &q_vector->napi,
                   ixgbe_poll, 64);
    /* MSI-X 벡터: 하나의 IRQ가 RX+TX 모두 처리 */
    request_irq(entry->vector, ixgbe_msix_clean_rings,
                0, q_vector->name, q_vector);
}

/* 패턴 2: mlx5 — 채널 기반 (분리된 CQ) */
for (i = 0; i < priv->channels.num; i++) {
    struct mlx5e_channel *c = priv->channels.c[i];
    netif_napi_add(priv->netdev, &c->napi,
                   mlx5e_napi_poll, 64);
    /* 각 채널에 독립 EQ(Event Queue) → CQ 연결 */
}

/* 패턴 3: virtio-net — virtqueue 콜백 */
for (i = 0; i < vi->curr_queue_pairs; i++) {
    netif_napi_add(vi->dev, &vi->rq[i].napi,
                   virtnet_poll, napi_weight);
    /* virtqueue callback → NAPI schedule */
    virtio_device_ready(vi->vdev);
}

컨테이너/네임스페이스(Namespace) 환경의 NAPI

veth 쌍과 NAPI

컨테이너 네트워킹에서 가장 흔한 가상 인터페이스인 veth(virtual Ethernet pair)는 NAPI를 사용하여 패킷을 수신합니다. veth의 한쪽에서 전송된 패킷은 상대편의 NAPI를 통해 수신됩니다.

/* drivers/net/veth.c */

/* veth의 NAPI 흐름:
   1. 컨테이너 A에서 패킷 전송 (vethA의 xmit)
   2. 상대편 vethB의 수신 큐에 인큐
   3. vethB의 NAPI 스케줄
   4. vethB의 네임스페이스에서 네트워크 스택 처리 */

/* veth xmit → 상대편 NAPI */
static netdev_tx_t veth_xmit(struct sk_buff *skb,
                              struct net_device *dev)
{
    struct veth_priv *rcv_priv, *priv = netdev_priv(dev);
    struct net_device *rcv;

    /* 피어 디바이스 참조 */
    rcv = rcu_dereference(priv->peer);
    rcv_priv = netdev_priv(rcv);

    /* XDP 활성 시: NAPI 기반 처리 */
    if (rcv_priv->_xdp_prog) {
        /* ptr_ring에 인큐 → NAPI 스케줄 */
        veth_xdp_rcv(rq, budget, ...);
    } else {
        /* XDP 없으면 직접 netif_rx() */
        netif_rx(skb);
    }
    return NETDEV_TX_OK;
}

/* veth NAPI poll */
static int veth_poll(struct napi_struct *napi, int budget)
{
    struct veth_rq *rq =
        container_of(napi, struct veth_rq, xdp_napi);
    int done;

    /* XDP 프로그램 실행 + 패킷 수신 */
    done = veth_xdp_rcv(rq, budget, ...);

    if (done < budget && napi_complete_done(napi, done)) {
        /* veth는 IRQ가 없으므로 재활성화 불필요 */
        if (unlikely(!__ptr_ring_empty(&rq->xdp_ring)))
            napi_schedule(napi);
    }
    return done;
}

네임스페이스 격리와 NAPI 동작

컨테이너 네트워크 성능 최적화

# 1. veth에 XDP 부착 (컨테이너 → 호스트 방향 가속)
ip link set veth_host xdp obj veth_xdp.o

# 2. TC BPF로 veth 패킷 처리 가속
tc qdisc add dev veth_host clsact
tc filter add dev veth_host ingress bpf obj tc_fwd.o

# 3. macvlan/ipvlan: veth 대신 직접 연결 (NAPI 경유 감소)
ip link add mvlan0 link eth0 type macvlan mode bridge
ip link set mvlan0 netns container_ns

# 4. SR-IOV VF passthrough: 물리 NIC NAPI 직접 사용
#    → 컨테이너에서 물리 NIC 성능에 가장 가까움
echo 4 > /sys/class/net/eth0/device/sriov_numvfs
ip link set eth0 vf 0 mac 00:11:22:33:44:55
ip link set enp1s0f0v0 netns container_ns

# 5. 컨테이너 내 busy polling 활성화
# (veth NAPI에 대한 busy poll — 효과는 물리 NIC보다 작음)
sysctl -w net.core.busy_poll=50
sysctl -w net.core.busy_read=50

NAPI 성능 벤치마킹 방법론

벤치마크 도구 비교

pktgen으로 NAPI 처리량 측정

# 커널 pktgen 모듈 로드
modprobe pktgen

# CPU 0에서 eth0으로 64B 패킷 전송
# /proc/net/pktgen/ 인터페이스 사용
cat > /proc/net/pktgen/kpktgend_0 <<'EOF'
rem_device_all
add_device eth0@0
EOF

cat > /proc/net/pktgen/eth0@0 <<'EOF'
count 10000000
min_pkt_size 64
max_pkt_size 64
dst 192.168.1.2
dst_mac aa:bb:cc:dd:ee:ff
delay 0
clone_skb 100000
EOF

# 테스트 시작
echo start > /proc/net/pktgen/pgctrl

# 결과 확인
cat /proc/net/pktgen/eth0@0
# Result: OK: ... (pps)
# 수신 측에서 /proc/net/softnet_stat 모니터링

NAPI 성능 기준선 (참고 수치)