如果覺得文章對你有所啟發，可以考慮用 🌟 支持 Gthulhu 專案，短期目標是集齊 300 個 🌟 藉此被 CNCF Landscape 採納 [ref]。

前言

在上一篇文章中，我們成功編寫了第一個 eBPF 程式，並學會了基本的程式結構和載入流程。今天我們將深入探討 eBPF Maps——這個連接內核空間和使用者空間的重要橋樑。

eBPF Maps 不僅僅是簡單的資料結構，它們是 eBPF 程式持久化狀態、程式間通訊、以及與使用者空間互動的核心機制。

認識 eBPF Maps

eBPF Maps 的設計理念

eBPF Maps 解決了幾個核心問題：

1. 狀態持久化：eBPF 程式重新載入後保持狀態 ref
2. 資料共享：多個 eBPF 程式間共享資料 ref
3. 用戶介面：使用者空間程式訪問內核資料
4. 高效通訊：避免系統呼叫的開銷

Maps 的分類

eBPF Maps 可以按照不同維度分類：

按資料結構分類

• Array Maps：基於索引的陣列
• Hash Maps：基於鍵值的雜湊表
• Trie Maps：最長前綴匹配
• Stack/Queue Maps：堆疊和佇列結構

按使用場景分類

• 統計 Maps：Per-CPU Maps、Percpu Array/Hash
• 通訊 Maps：(User-)Ring Buffer、Perf Event Array
• 配置 Maps：Program Array、Map in Map
• 網路 Maps：Socket Maps、Device Maps

筆者補充：詳細的 Map 種類請參考：https://docs.ebpf.io/linux/map-type/

實作演練：深入探索各種 Maps

第一步：專案結構準備

mkdir -p ~/ebpf-maps-demo/{src/kernel,src/userspace,headers,bin}
cd ~/ebpf-maps-demo

# 建立 Makefile 和 Go module
cp ~/ebpf-hello-world/Makefile .
cp ~/ebpf-hello-world/go.mod .

第二步：Array Maps 實作

Array Maps 是最簡單的 Map 型別，適合存儲固定大小的陣列資料。

// src/kernel/array_maps_demo.bpf.c
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

// 定義一個簡單的陣列 Map
struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 256);  // 最大 256 個元素
    __type(key, __u32);        // 鍵類型：32位整數（索引）
    __type(value, __u64);      // 值類型：64位整數（計數）
} syscall_count_array SEC(".maps");

// 定義一個 Per-CPU 陣列 Map
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
    __uint(max_entries, 10);
    __type(key, __u32);
    __type(value, __u64);
} percpu_stats SEC(".maps");

// 定義統計結構
struct syscall_stats {
    __u64 count;
    __u64 total_time;
    __u64 min_time;
    __u64 max_time;
};

// 複雜的陣列 Map
struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 512);
    __type(key, __u32);
    __type(value, struct syscall_stats);
} detailed_stats SEC(".maps");

char LICENSE[] SEC("license") = "GPL";

SEC("tp/raw_syscalls/sys_enter")
int trace_syscall_enter(struct trace_event_raw_sys_enter *ctx)
{
    __u32 syscall_nr = ctx->id;
    __u64 *count;
    __u64 timestamp = bpf_ktime_get_ns();
    
    // 限制系統呼叫號範圍，避免越界
    if (syscall_nr >= 256)
        return 0;
    
    // 更新簡單計數陣列
    count = bpf_map_lookup_elem(&syscall_count_array, &syscall_nr);
    if (count) {
        __sync_fetch_and_add(count, 1);
    } else {
        __u64 initial = 1;
        bpf_map_update_elem(&syscall_count_array, &syscall_nr, &initial, BPF_ANY);
    }
    
    // 更新 Per-CPU 統計（假設只統計前10個系統呼叫）
    if (syscall_nr < 10) {
        __u64 *percpu_count = bpf_map_lookup_elem(&percpu_stats, &syscall_nr);
        if (percpu_count) {
            *percpu_count += 1;
        } else {
            __u64 initial = 1;
            bpf_map_update_elem(&percpu_stats, &syscall_nr, &initial, BPF_ANY);
        }
    }
    
    // 更新詳細統計（限制範圍）
    if (syscall_nr < 512) {
        struct syscall_stats *stats = bpf_map_lookup_elem(&detailed_stats, &syscall_nr);
        if (stats) {
            __sync_fetch_and_add(&stats->count, 1);
            if (stats->min_time == 0 || timestamp < stats->min_time) {
                stats->min_time = timestamp;
            }
            if (timestamp > stats->max_time) {
                stats->max_time = timestamp;
            }
        } else {
            struct syscall_stats new_stats = {
                .count = 1,
                .total_time = 0,
                .min_time = timestamp,
                .max_time = timestamp
            };
            bpf_map_update_elem(&detailed_stats, &syscall_nr, &new_stats, BPF_ANY);
        }
    }
    
    return 0;
}

第三步：Hash Maps 實作

Hash Maps 提供靈活的鍵值對存儲，適合動態資料。

// src/kernel/hash_maps_demo.bpf.c
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

// 定義進程統計結構
struct process_stats {
    __u64 file_opens;
    __u64 memory_allocs;
    __u64 network_ops;
    __u64 first_seen;
    __u64 last_seen;
    char comm[16];
};

// PID -> 進程統計的 Hash Map
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 10000);
    __type(key, __u32);  // PID
    __type(value, struct process_stats);
} process_map SEC(".maps");

// 檔案名 Hash -> 存取次數
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 1000);
    __type(key, __u64);  // 檔名雜湊
    __type(value, __u64); // 存取次數
} file_access_map SEC(".maps");

// Per-CPU Hash Map 示例
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_HASH);
    __uint(max_entries, 1000);
    __type(key, __u32);
    __type(value, __u64);
} percpu_process_stats SEC(".maps");

char LICENSE[] SEC("license") = "GPL";

// 簡單的字串雜湊函數
static __always_inline __u64 hash_string(const char *str, int len)
{
    __u64 hash = 5381;
    for (int i = 0; i < len && i < 16; i++) {
        if (str[i] == 0) break;
        hash = ((hash << 5) + hash) + str[i];
    }
    return hash;
}

SEC("tp/syscalls/sys_enter_openat")
int trace_file_opens(struct trace_event_raw_sys_enter *ctx)
{
    __u32 pid = bpf_get_current_pid_tgid() >> 32;
    __u64 now = bpf_ktime_get_ns();
    
    // 更新進程統計
    struct process_stats *stats = bpf_map_lookup_elem(&process_map, &pid);
    if (stats) {
        __sync_fetch_and_add(&stats->file_opens, 1);
        stats->last_seen = now;
    } else {
        struct process_stats new_stats = {
            .file_opens = 1,
            .memory_allocs = 0,
            .network_ops = 0,
            .first_seen = now,
            .last_seen = now
        };
        bpf_get_current_comm(&new_stats.comm, sizeof(new_stats.comm));
        bpf_map_update_elem(&process_map, &pid, &new_stats, BPF_ANY);
    }
    
    // 嘗試獲取檔案名並雜湊
    char filename[64];
    long ret = bpf_probe_read_user_str(filename, sizeof(filename), 
                                      (void *)ctx->args[1]);
    if (ret > 0) {
        __u64 file_hash = hash_string(filename, ret);
        __u64 *count = bpf_map_lookup_elem(&file_access_map, &file_hash);
        if (count) {
            __sync_fetch_and_add(count, 1);
        } else {
            __u64 initial = 1;
            bpf_map_update_elem(&file_access_map, &file_hash, &initial, BPF_ANY);
        }
    }
    
    // 更新 Per-CPU 統計
    __u64 *percpu_count = bpf_map_lookup_elem(&percpu_process_stats, &pid);
    if (percpu_count) {
        *percpu_count += 1;
    } else {
        __u64 initial = 1;
        bpf_map_update_elem(&percpu_process_stats, &pid, &initial, BPF_ANY);
    }
    
    return 0;
}

第四步：Ring Buffer 實作

Ring Buffer 是現代 eBPF 推薦的事件通信機制。

// src/kernel/ringbuf_demo.bpf.c
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

// 定義事件結構
struct file_event {
    __u32 pid;
    __u32 tgid;
    __u32 uid;
    __u64 timestamp;
    __u32 filename_len;
    char comm[16];
    char filename[256];
};

// Ring Buffer Map
struct {
    __uint(type, BPF_MAP_TYPE_RINGBUF);
    __uint(max_entries, 256 * 1024); // 256KB buffer
} events SEC(".maps");

char LICENSE[] SEC("license") = "GPL";

SEC("tp/syscalls/sys_enter_openat")
int trace_file_events(struct trace_event_raw_sys_enter *ctx)
{
    struct file_event *event;
    
    // 從 ring buffer 分配空間
    event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
    if (!event) {
        return 0; // 分配失敗，buffer 可能已滿
    }
    
    // 填充事件資料
    __u64 pid_tgid = bpf_get_current_pid_tgid();
    event->pid = pid_tgid >> 32;
    event->tgid = pid_tgid & 0xffffffff;
    event->uid = bpf_get_current_uid_gid() & 0xffffffff;
    event->timestamp = bpf_ktime_get_ns();
    
    // 獲取進程名稱
    bpf_get_current_comm(&event->comm, sizeof(event->comm));
    
    // 讀取檔案名
    long ret = bpf_probe_read_user_str(event->filename, sizeof(event->filename),
                                      (void *)ctx->args[1]);
    if (ret > 0) {
        event->filename_len = ret;
    } else {
        event->filename_len = 0;
        event->filename[0] = '\0';
    }
    
    // 提交事件到 ring buffer
    bpf_ringbuf_submit(event, 0);
    
    return 0;
}

第五步：Go 使用者空間程式

現在編寫一個完整的 Go 程式來操作所有這些 Maps：

// src/userspace/maps_demo.go
package main

import (
    "bytes"
    "context"
    "encoding/binary"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sort"
    "syscall"
    "time"
    "unsafe"

    bpf "github.com/aquasecurity/libbpfgo"
)

// Go 結構對應 C 結構
type ProcessStats struct {
    FileOpens    uint64
    MemoryAllocs uint64
    NetworkOps   uint64
    FirstSeen    uint64
    LastSeen     uint64
    Comm         [16]byte
}

type SyscallStats struct {
    Count     uint64
    TotalTime uint64
    MinTime   uint64
    MaxTime   uint64
}

type FileEvent struct {
    PID         uint32
    TGID        uint32
    UID         uint32
    Timestamp   uint64
    FilenameLen uint32
    Comm        [16]byte
    Filename    [256]byte
}

func main() {
    if os.Geteuid() != 0 {
        log.Fatal("需要 root 權限執行")
    }

    // 選擇要載入的程式
    var objFile string
    var progName string

    if len(os.Args) > 1 {
        switch os.Args[1] {
        case "array":
            objFile = "array_maps_demo.bpf.o"
            progName = "trace_syscall_enter"
        case "hash":
            objFile = "hash_maps_demo.bpf.o"
            progName = "trace_file_opens"
        case "ringbuf":
            objFile = "ringbuf_demo.bpf.o"
            progName = "trace_file_events"
        default:
            fmt.Println("使用方法: ./maps_demo [array|hash|ringbuf]")
            os.Exit(1)
        }
    } else {
        objFile = "hash_maps_demo.bpf.o"
        progName = "trace_file_opens"
    }

    fmt.Printf("載入 eBPF 程式: %s\n", objFile)

    // 載入 eBPF 程式
    bpfModule, err := bpf.NewModuleFromFile(objFile)
    if err != nil {
        log.Fatalf("載入 eBPF 模組失敗: %v", err)
    }
    defer bpfModule.Close()

    if err := bpfModule.BPFLoadObject(); err != nil {
        log.Fatalf("載入 eBPF 物件失敗: %v", err)
    }

    // 獲取並附加程式
    prog, err := bpfModule.GetProgram(progName)
    if err != nil {
        log.Fatalf("獲取程式失敗: %v", err)
    }

    var link *bpf.BPFLink
    if progName == "trace_syscall_enter" {
        link, err = prog.AttachTracepoint(&bpf.TracepointOpts{
            Group: "raw_syscalls",
            Name:  "sys_enter",
        })
    } else {
        link, err = prog.AttachTracepoint(&bpf.TracepointOpts{
            Group: "syscalls",
            Name:  "sys_enter_openat",
        })
    }

    if err != nil {
        log.Fatalf("附加程式失敗: %v", err)
    }
    defer link.Close()

    fmt.Printf("✓ eBPF 程式 %s 附加成功\n", progName)

    // 設定信號處理
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    // 根據程式類型執行不同的處理邏輯
    switch os.Args[1] {
    case "array":
        runArrayDemo(ctx, bpfModule, sigChan)
    case "hash":
        runHashDemo(ctx, bpfModule, sigChan)
    case "ringbuf":
        runRingBufDemo(ctx, bpfModule, sigChan)
    default:
        runHashDemo(ctx, bpfModule, sigChan)
    }
}

func runArrayDemo(ctx context.Context, module *bpf.Module, sigChan chan os.Signal) {
    syscallCountMap, err := module.GetMap("syscall_count_array")
    if err != nil {
        log.Fatalf("獲取 syscall_count_array 失敗: %v", err)
    }

    detailedStatsMap, err := module.GetMap("detailed_stats")
    if err != nil {
        log.Fatalf("獲取 detailed_stats 失敗: %v", err)
    }

    ticker := time.NewTicker(3 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case sig := <-sigChan:
            fmt.Printf("\n收到信號 %v，正在退出...\n", sig)
            cancel()
        case <-ticker.C:
            fmt.Println("\n=== Array Maps 統計 ===")
            
            // 顯示系統呼叫計數
            fmt.Println("系統呼叫計數 (前10個):")
            for i := uint32(0); i < 10; i++ {
                valueBytes, err := syscallCountMap.GetValue(unsafe.Pointer(&i))
                if err == nil && len(valueBytes) == 8 {
                    count := binary.LittleEndian.Uint64(valueBytes)
                    if count > 0 {
                        fmt.Printf("  syscall_%d: %d 次\n", i, count)
                    }
                }
            }

            // 顯示詳細統計
            fmt.Println("\n詳細統計 (有活動的系統呼叫):")
            for i := uint32(0); i < 20; i++ {
                valueBytes, err := detailedStatsMap.GetValue(unsafe.Pointer(&i))
                if err == nil && len(valueBytes) == int(unsafe.Sizeof(SyscallStats{})) {
                    stats := (*SyscallStats)(unsafe.Pointer(&valueBytes[0]))
                    if stats.Count > 0 {
                        fmt.Printf("  syscall_%d: %d次, 最小時間: %d, 最大時間: %d\n",
                            i, stats.Count, stats.MinTime, stats.MaxTime)
                    }
                }
            }
        }
    }
}

func runHashDemo(ctx context.Context, module *bpf.Module, sigChan chan os.Signal) {
    processMap, err := module.GetMap("process_map")
    if err != nil {
        log.Fatalf("獲取 process_map 失敗: %v", err)
    }

    fileAccessMap, err := module.GetMap("file_access_map")
    if err != nil {
        log.Fatalf("獲取 file_access_map 失敗: %v", err)
    }

    ticker := time.NewTicker(3 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case sig := <-sigChan:
            fmt.Printf("\n收到信號 %v，正在退出...\n", sig)
            cancel()
        case <-ticker.C:
            fmt.Println("\n=== Hash Maps 統計 ===")
            
            // 顯示進程統計
            fmt.Println("進程檔案操作統計:")
            
            type ProcessInfo struct {
                PID   uint32
                Stats ProcessStats
            }
            
            var processes []ProcessInfo
            
            iterator := processMap.Iterator()
            for iterator.Next() {
                keyBytes := iterator.Key()
                valueBytes, _ := iterator.Value()
                
                if len(keyBytes) == 4 && len(valueBytes) == int(unsafe.Sizeof(ProcessStats{})) {
                    pid := binary.LittleEndian.Uint32(keyBytes)
                    stats := (*ProcessStats)(unsafe.Pointer(&valueBytes[0]))
                    processes = append(processes, ProcessInfo{PID: pid, Stats: *stats})
                }
            }
            
            // 按檔案操作次數排序
            sort.Slice(processes, func(i, j int) bool {
                return processes[i].Stats.FileOpens > processes[j].Stats.FileOpens
            })
            
            // 顯示前10個進程
            count := 10
            if len(processes) < count {
                count = len(processes)
            }
            
            for i := 0; i < count; i++ {
                p := processes[i]
                comm := string(bytes.TrimRight(p.Stats.Comm[:], "\x00"))
                fmt.Printf("  PID %d (%s): %d 次檔案操作\n", 
                    p.PID, comm, p.Stats.FileOpens)
            }
            
            // 顯示檔案存取統計
            fmt.Println("\n檔案存取統計 (雜湊):")
            fileIterator := fileAccessMap.Iterator()
            fileCount := 0
            for fileIterator.Next() && fileCount < 5 {
                keyBytes := fileIterator.Key()
                valueBytes, _ := fileIterator.Value()
                
                if len(keyBytes) == 8 && len(valueBytes) == 8 {
                    hash := binary.LittleEndian.Uint64(keyBytes)
                    count := binary.LittleEndian.Uint64(valueBytes)
                    fmt.Printf("  Hash 0x%x: %d 次存取\n", hash, count)
                    fileCount++
                }
            }
        }
    }
}

func runRingBufDemo(ctx context.Context, module *bpf.Module, sigChan chan os.Signal) {
    eventsMap, err := module.GetMap("events")
    if err != nil {
        log.Fatalf("獲取 events ringbuf 失敗: %v", err)
    }

    // 建立 ring buffer
    rb, err := module.InitRingBuf("events", make(chan []byte, 100))
    if err != nil {
        log.Fatalf("初始化 ring buffer 失敗: %v", err)
    }
    defer rb.Close()

    // 開始 polling
    rb.Poll(300) // 300ms timeout

    fmt.Println("開始監聽檔案事件 (Ring Buffer)...")
    fmt.Println("執行一些檔案操作來觸發事件...")

    eventCount := 0

    for {
        select {
        case <-ctx.Done():
            return
        case sig := <-sigChan:
            fmt.Printf("\n收到信號 %v，正在退出...\n", sig)
            cancel()
        case data := <-rb.EventsChannel:
            if len(data) == int(unsafe.Sizeof(FileEvent{})) {
                event := (*FileEvent)(unsafe.Pointer(&data[0]))
                
                comm := string(bytes.TrimRight(event.Comm[:], "\x00"))
                filename := string(bytes.TrimRight(event.Filename[:event.FilenameLen], "\x00"))
                
                eventCount++
                fmt.Printf("[%d] PID %d (%s) 開啟檔案: %s\n", 
                    eventCount, event.PID, comm, filename)
            }
        }
    }
}

第六步：編譯和測試

更新 Makefile 以支援多個程式：

# 更新 Makefile
BPF_PROGS := array_maps_demo.bpf.o hash_maps_demo.bpf.o ringbuf_demo.bpf.o

all: vmlinux.h $(BPF_PROGS) userspace

$(BPF_PROGS): %.bpf.o: $(KERNEL_SRC)/%.bpf.c | vmlinux.h
    @echo "編譯 eBPF 程式: $<"
    $(CLANG) $(BPF_CFLAGS) -I$(HEADERS) -c $< -o $@
    $(LLVM_STRIP) -g $@

userspace: bin/maps_demo

bin/maps_demo: $(USERSPACE_SRC)/maps_demo.go
    @echo "編譯使用者空間程式..."
    mkdir -p bin
    $(GO) build -o $@ $<

test-array: all
    sudo ./bin/maps_demo array

test-hash: all  
    sudo ./bin/maps_demo hash

test-ringbuf: all
    sudo ./bin/maps_demo ringbuf

編譯和測試：

# 編譯所有程式
make all

# 測試 Array Maps
make test-array

# 測試 Hash Maps  
make test-hash

# 測試 Ring Buffer
make test-ringbuf

深度分析：Maps 的效能考量

1. Map 類型選擇

不同 Map 類型的效能特性：

2. Per-CPU Maps 的優勢

// Per-CPU Maps 避免鎖爭用
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_HASH);
    __uint(max_entries, 1000);
    __type(key, __u32);
    __type(value, __u64);
} percpu_map SEC(".maps");

// 在多核系統上，每個 CPU 有獨立的副本
SEC("tp/syscalls/sys_enter_write")
int count_writes(void *ctx)
{
    __u32 pid = bpf_get_current_pid_tgid() >> 32;
    __u64 *count = bpf_map_lookup_elem(&percpu_map, &pid);
    if (count) {
        *count += 1; // 無需原子操作，只在當前 CPU 上修改
    }
    return 0;
}

使用者空間需要聚合所有 CPU 的值：

// 讀取 Per-CPU Map 的值
func readPerCPUValue(m *bpf.BPFMap, key unsafe.Pointer) (uint64, error) {
    values, err := m.GetValuePerCPU(key)
    if err != nil {
        return 0, err
    }
    
    var total uint64
    for _, valueBytes := range values {
        if len(valueBytes) == 8 {
            val := binary.LittleEndian.Uint64(valueBytes)
            total += val
        }
    }
    return total, nil
}

3. 記憶體管理最佳化

// Map 大小設計原則
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 10000);  // 根據實際需求設定
    __uint(map_flags, BPF_F_NO_PREALLOC); // 延遲分配記憶體
    __type(key, __u32);
    __type(value, struct large_value);
} optimized_map SEC(".maps");

疑難排解與最佳化

常見問題

1. Map 滿了怎麼辦？

// 使用 LRU Map 自動淘汰舊項目
struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(max_entries, 1000);
    __type(key, __u32);
    __type(value, __u64);
} lru_map SEC(".maps");

2. 如何處理 Map 更新衝突？

// 使用原子操作
__sync_fetch_and_add(&stats->counter, 1);

// 或使用 CAS 操作
__u64 old_val, new_val;
do {
    old_val = stats->counter;
    new_val = old_val + 1;
} while (!__sync_bool_compare_and_swap(&stats->counter, old_val, new_val));

3. 監控 Map 使用情況

sudo bpftool map show
sudo bpftool map dump id <MAP_ID>

# 檢查 Map 統計
cat /proc/kallsyms | grep bpf_map

總結

今天我們深入探討了 eBPF Maps 的各個方面：

1. Map 類型：Array、Hash、Ring Buffer 等不同類型的特點和用途
2. 效能考量：Per-CPU Maps、記憶體管理、併發處理
3. 實際應用：統計系統、配置管理、事件聚合
4. 最佳實務：類型選擇、大小設定、錯誤處理

Maps 是 eBPF 程式設計的核心，掌握了 Maps 的使用，就掌握了 eBPF 開發的精髓。在下一篇文章中，我們將探索 XDP 網路程式設計，學習如何在最底層處理網路封包。