How to Use eBPF for Monitoring Linux Thread Contention#

1
#include <vmlinux.h>
2
#include <bpf/bpf_helpers.h>
3
#include <bpf/bpf_tracing.h>
4
#include <bpf/bpf_core_read.h>
5

6
// Data structures for tracking contention
7
struct contention_event {
8
    u32 pid;
9
    u32 tid;
10
    u64 timestamp;
11
    u64 duration;
12
    u32 contention_type;
13
    char comm[16];
14
};
15

16
struct {
17
    __uint(type, BPF_MAP_TYPE_HASH);
18
    __uint(max_entries, 10240);
19
    __type(key, u32);
20
    __type(value, u64);
21
} thread_start_time SEC(".maps");
22

23
struct {
24
    __uint(type, BPF_MAP_TYPE_RINGBUF);
25
    __uint(max_entries, 256 * 1024);
26
} events SEC(".maps");
27

28
// Track runqueue latency
29
SEC("tp/sched/sched_wakeup")
30
int trace_sched_wakeup(struct trace_event_raw_sched_wakeup *ctx) {
31
    u32 pid = ctx->pid;
32
    u64 ts = bpf_ktime_get_ns();
33

34
    bpf_map_update_elem(&thread_start_time, &pid, &ts, BPF_ANY);
35
    return 0;
36
}
37

38
SEC("tp/sched/sched_switch")
39
int trace_sched_switch(struct trace_event_raw_sched_switch *ctx) {
40
    u32 prev_pid = ctx->prev_pid;
41
    u32 next_pid = ctx->next_pid;
42
    u64 ts = bpf_ktime_get_ns();
43

44
    // Handle runqueue latency for incoming thread
45
    u64 *start_ts = bpf_map_lookup_elem(&thread_start_time, &next_pid);
46
    if (start_ts) {
47
        u64 latency = ts - *start_ts;
48

49
        // Only report significant latencies (> 1ms)
50
        if (latency > 1000000) {
51
            struct contention_event *event =
52
                bpf_ringbuf_reserve(&events, sizeof(*event), 0);
53
            if (event) {
54
                event->pid = ctx->next_pid >> 32;
55
                event->tid = next_pid;
56
                event->timestamp = ts;
57
                event->duration = latency;
58
                event->contention_type = 1; // Runqueue contention
59
                bpf_get_current_comm(event->comm, sizeof(event->comm));
60

61
                bpf_ringbuf_submit(event, 0);
62
            }
63
        }
64

65
        bpf_map_delete_elem(&thread_start_time, &next_pid);
66
    }
67

68
    return 0;
69
}
70

71
// Track mutex contention
72
SEC("kprobe/mutex_lock_slowpath")
73
int trace_mutex_lock_slowpath(struct pt_regs *ctx) {
74
    u32 tid = bpf_get_current_pid_tgid();
75
    u64 ts = bpf_ktime_get_ns();
76

77
    bpf_map_update_elem(&thread_start_time, &tid, &ts, BPF_ANY);
78
    return 0;
79
}
80

81
SEC("kretprobe/mutex_lock_slowpath")
82
int trace_mutex_lock_slowpath_ret(struct pt_regs *ctx) {
83
    u32 tid = bpf_get_current_pid_tgid();
84
    u64 ts = bpf_ktime_get_ns();
85

86
    u64 *start_ts = bpf_map_lookup_elem(&thread_start_time, &tid);
87
    if (start_ts) {
88
        u64 duration = ts - *start_ts;
89

90
        struct contention_event *event =
91
            bpf_ringbuf_reserve(&events, sizeof(*event), 0);
92
        if (event) {
93
            event->pid = tid >> 32;
94
            event->tid = tid;
95
            event->timestamp = ts;
96
            event->duration = duration;
97
            event->contention_type = 2; // Mutex contention
98
            bpf_get_current_comm(event->comm, sizeof(event->comm));
99

100
            bpf_ringbuf_submit(event, 0);
101
        }
102

103
        bpf_map_delete_elem(&thread_start_time, &tid);
104
    }
105

106
    return 0;
107
}
108

109
char _license[] SEC("license") = "GPL";

1
#include <stdio.h>
2
#include <stdlib.h>
3
#include <unistd.h>
4
#include <signal.h>
5
#include <time.h>
6
#include <bpf/libbpf.h>
7
#include <bpf/bpf.h>
8

9
struct contention_event {
10
    u32 pid;
11
    u32 tid;
12
    u64 timestamp;
13
    u64 duration;
14
    u32 contention_type;
15
    char comm[16];
16
};
17

18
static volatile bool running = true;
19

20
static void sig_handler(int sig) {
21
    running = false;
22
}
23

24
static const char* contention_type_str(u32 type) {
25
    switch (type) {
26
        case 1: return "RUNQUEUE";
27
        case 2: return "MUTEX";
28
        default: return "UNKNOWN";
29
    }
30
}
31

32
static int handle_event(void *ctx, void *data, size_t data_sz) {
33
    const struct contention_event *e = data;
34
    struct tm *tm;
35
    char ts[32];
36
    time_t t;
37

38
    t = e->timestamp / 1000000000;
39
    tm = localtime(&t);
40
    strftime(ts, sizeof(ts), "%H:%M:%S", tm);
41

42
    printf("%s.%03llu %-15s PID: %u TID: %u DURATION: %llu us TYPE: %s\n",
43
           ts, (e->timestamp % 1000000000) / 1000000,
44
           e->comm, e->pid, e->tid, e->duration / 1000,
45
           contention_type_str(e->contention_type));
46

47
    return 0;
48
}
49

50
int main(int argc, char **argv) {
51
    struct bpf_object *obj;
52
    struct bpf_link *links[4];
53
    struct ring_buffer *rb = NULL;
54
    int err;
55

56
    // Set up signal handlers
57
    signal(SIGINT, sig_handler);
58
    signal(SIGTERM, sig_handler);
59

60
    // Load eBPF program
61
    obj = bpf_object__open_file("thread_contention.bpf.o", NULL);
62
    if (libbpf_get_error(obj)) {
63
        fprintf(stderr, "Failed to open BPF object\n");
64
        return 1;
65
    }
66

67
    err = bpf_object__load(obj);
68
    if (err) {
69
        fprintf(stderr, "Failed to load BPF object: %d\n", err);
70
        goto cleanup;
71
    }
72

73
    // Attach programs to tracepoints and kprobes
74
    links[0] = bpf_program__attach(bpf_object__find_program_by_name(obj, "trace_sched_wakeup"));
75
    links[1] = bpf_program__attach(bpf_object__find_program_by_name(obj, "trace_sched_switch"));
76
    links[2] = bpf_program__attach(bpf_object__find_program_by_name(obj, "trace_mutex_lock_slowpath"));
77
    links[3] = bpf_program__attach(bpf_object__find_program_by_name(obj, "trace_mutex_lock_slowpath_ret"));
78

79
    // Set up ring buffer
80
    rb = ring_buffer__new(bpf_object__find_map_fd_by_name(obj, "events"),
81
                         handle_event, NULL, NULL);
82
    if (!rb) {
83
        fprintf(stderr, "Failed to create ring buffer\n");
84
        goto cleanup;
85
    }
86

87
    printf("Monitoring thread contention... Press Ctrl-C to exit.\n");
88
    printf("TIME      COMM            PID    TID    DURATION TYPE\n");
89

90
    // Poll for events
91
    while (running) {
92
        err = ring_buffer__poll(rb, 100);
93
        if (err < 0 && err != -EINTR) {
94
            fprintf(stderr, "Error polling ring buffer: %d\n", err);
95
            break;
96
        }
97
    }
98

99
cleanup:
100
    ring_buffer__free(rb);
101
    for (int i = 0; i < 4; i++) {
102
        if (links[i]) bpf_link__destroy(links[i]);
103
    }
104
    bpf_object__close(obj);
105

106
    return err < 0 ? 1 : 0;
107
}

1
#!/bin/bash
2
DURATION=${1:-60}
3
OUTPUT_DIR="thread_monitoring_$(date +%Y%m%d_%H%M%S)"
4

5
mkdir -p "$OUTPUT_DIR"
6

7
echo "Starting comprehensive thread contention monitoring for ${DURATION} seconds..."
8

9
# Start multiple monitoring tools in background
10
sudo /usr/share/bcc/tools/offcputime.py -d $DURATION > "$OUTPUT_DIR/offcpu.txt" &
11
sudo /usr/share/bcc/tools/runqlat.py -d $DURATION > "$OUTPUT_DIR/runqlat.txt" &
12
sudo /usr/share/bcc/tools/wakeuptime.py -d $DURATION > "$OUTPUT_DIR/wakeup.txt" &
13

14
# Custom bpftrace script for lock contention
15
sudo bpftrace -e '
16
kprobe:mutex_lock { @lock_attempts[comm] = count(); }
17
kprobe:mutex_lock_slowpath { @lock_contentions[comm] = count(); }
18
END { printf("\nLock Contention Summary:\n"); print(@lock_attempts); print(@lock_contentions); }
19
' > "$OUTPUT_DIR/locks.txt" &
20

21
# Wait for all background jobs
22
wait
23

24
echo "Monitoring complete. Results saved in $OUTPUT_DIR/"
25

26
# Generate summary report
27
python3 << EOF
28
import os
29
import glob
30

31
print("=== Thread Contention Analysis Summary ===")
32
print(f"Analysis period: {$DURATION} seconds")
33
print(f"Output directory: {$OUTPUT_DIR}")
34

35
# Process results and generate insights
36
for file in glob.glob("$OUTPUT_DIR/*.txt"):
37
    print(f"\n--- {os.path.basename(file)} ---")
38
    with open(file, 'r') as f:
39
        lines = f.readlines()
40
        print(f"Total lines: {len(lines)}")
41
        if lines:
42
            print("Sample output:")
43
            for line in lines[:5]:
44
                print(f"  {line.strip()}")
45
            if len(lines) > 5:
46
                print("  ...")
47
EOF

1
#!/usr/bin/env python3
2
import time
3
import subprocess
4
import curses
5
from collections import defaultdict, deque
6
import json
7

8
class ThreadContentionDashboard:
9
    def __init__(self):
10
        self.stats = defaultdict(lambda: defaultdict(int))
11
        self.history = defaultdict(lambda: deque(maxlen=60))
12

13
    def collect_stats(self):
14
        """Collect thread contention statistics"""
15
        try:
16
            # Collect runqueue latency
17
            result = subprocess.run([
18
                'sudo', '/usr/share/bcc/tools/runqlat.py', '-d', '1'
19
            ], capture_output=True, text=True, timeout=2)
20

21
            if result.returncode == 0:
22
                self.parse_runqlat_output(result.stdout)
23

24
        except subprocess.TimeoutExpired:
25
            pass
26
        except Exception as e:
27
            print(f"Error collecting stats: {e}")
28

29
    def parse_runqlat_output(self, output):
30
        """Parse runqlat output and extract statistics"""
31
        lines = output.strip().split('\n')
32
        for line in lines:
33
            if 'usecs' in line and ':' in line:
34
                # Parse histogram data
35
                parts = line.split()
36
                if len(parts) >= 3:
37
                    range_str = parts[0]
38
                    count = int(parts[-1])
39
                    self.stats['runqlat'][range_str] = count
40

41
    def display_dashboard(self, stdscr):
42
        """Display real-time dashboard"""
43
        stdscr.clear()
44
        stdscr.nodelay(True)
45

46
        while True:
47
            stdscr.clear()
48

49
            # Header
50
            stdscr.addstr(0, 0, "Thread Contention Monitor", curses.A_BOLD)
51
            stdscr.addstr(1, 0, f"Updated: {time.strftime('%H:%M:%S')}")
52
            stdscr.addstr(2, 0, "-" * 60)
53

54
            # Runqueue latency stats
55
            row = 4
56
            stdscr.addstr(row, 0, "Runqueue Latency Distribution:", curses.A_BOLD)
57
            row += 1
58

59
            for range_str, count in sorted(self.stats['runqlat'].items()):
60
                if count > 0:
61
                    bar = "#" * min(count // 10, 50)
62
                    stdscr.addstr(row, 0, f"{range_str:>15}: {count:>6} {bar}")
63
                    row += 1
64

65
            # Instructions
66
            stdscr.addstr(row + 2, 0, "Press 'q' to quit, 'r' to reset stats")
67

68
            stdscr.refresh()
69

70
            # Handle keyboard input
71
            key = stdscr.getch()
72
            if key == ord('q'):
73
                break
74
            elif key == ord('r'):
75
                self.stats.clear()
76
                self.history.clear()
77

78
            # Collect new data
79
            self.collect_stats()
80
            time.sleep(1)
81

82
def main():
83
    dashboard = ThreadContentionDashboard()
84
    curses.wrapper(dashboard.display_dashboard)
85

86
if __name__ == "__main__":
87
    main()

1
#!/bin/bash
2
THRESHOLD_MS=50  # Alert if contention > 50ms
3

4
while true; do
5
    MAX_CONTENTION=$(sudo bpftrace -e '
6
    tracepoint:sched:sched_switch {
7
        if (@start[args->next_pid]) {
8
            $latency = (nsecs - @start[args->next_pid]) / 1000000;
9
            if ($latency > '$THRESHOLD_MS') {
10
                printf("ALERT: High contention %d ms for PID %d\n",
11
                       $latency, args->next_pid);
12
            }
13
        }
14
        @start[args->next_pid] = nsecs;
15
    }' 2>/dev/null | head -1)
16

17
    if [[ -n "$MAX_CONTENTION" ]]; then
18
        # Send alert (email, Slack, etc.)
19
        echo "$MAX_CONTENTION" | mail -s "Thread Contention Alert" admin@company.com
20
    fi
21

22
    sleep 60
23
done