rk35xx-android14/external/bcc/tests/python/test_ringbuf.py

#!/usr/bin/env python3
# Copyright (c) PLUMgrid, Inc.
# Licensed under the Apache License, Version 2.0 (the "License")

from bcc import BPF
import os
import ctypes as ct
import random
import time
import subprocess
from unittest import main, TestCase, skipUnless
from utils import kernel_version_ge

class TestRingbuf(TestCase):
    @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8")
    def test_ringbuf_output(self):
        self.counter = 0

        class Data(ct.Structure):
            _fields_ = [("ts", ct.c_ulonglong)]

        def cb(ctx, data, size):
            self.assertEqual(size, ct.sizeof(Data))
            event = ct.cast(data, ct.POINTER(Data)).contents
            self.counter += 1

        text = """
BPF_RINGBUF_OUTPUT(events, 8);
struct data_t {
    u64 ts;
};
int do_sys_nanosleep(void *ctx) {
    struct data_t data = {bpf_ktime_get_ns()};
    events.ringbuf_output(&data, sizeof(data), 0);
    return 0;
}
"""
        b = BPF(text=text)
        b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"),
                        fn_name="do_sys_nanosleep")
        b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"),
                        fn_name="do_sys_nanosleep")
        b["events"].open_ring_buffer(cb)
        subprocess.call(['sleep', '0.1'])
        b.ring_buffer_poll()
        self.assertGreater(self.counter, 0)
        b.cleanup()

    @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8")
    def test_ringbuf_consume(self):
        self.counter = 0

        class Data(ct.Structure):
            _fields_ = [("ts", ct.c_ulonglong)]

        def cb(ctx, data, size):
            self.assertEqual(size, ct.sizeof(Data))
            event = ct.cast(data, ct.POINTER(Data)).contents
            self.counter += 1

        text = """
BPF_RINGBUF_OUTPUT(events, 8);
struct data_t {
    u64 ts;
};
int do_sys_nanosleep(void *ctx) {
    struct data_t data = {bpf_ktime_get_ns()};
    events.ringbuf_output(&data, sizeof(data), 0);
    return 0;
}
"""
        b = BPF(text=text)
        b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"),
                        fn_name="do_sys_nanosleep")
        b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"),
                        fn_name="do_sys_nanosleep")
        b["events"].open_ring_buffer(cb)
        subprocess.call(['sleep', '0.1'])
        b.ring_buffer_consume()
        self.assertGreater(self.counter, 0)
        b.cleanup()

    @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8")
    def test_ringbuf_submit(self):
        self.counter = 0

        class Data(ct.Structure):
            _fields_ = [("ts", ct.c_ulonglong)]

        def cb(ctx, data, size):
            self.assertEqual(size, ct.sizeof(Data))
            event = ct.cast(data, ct.POINTER(Data)).contents
            self.counter += 1

        text = """
BPF_RINGBUF_OUTPUT(events, 8);
struct data_t {
    u64 ts;
};
int do_sys_nanosleep(void *ctx) {
    struct data_t *data = events.ringbuf_reserve(sizeof(struct data_t));
    if (!data)
        return 1;
    data->ts = bpf_ktime_get_ns();
    events.ringbuf_submit(data, 0);
    return 0;
}
"""
        b = BPF(text=text)
        b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"),
                        fn_name="do_sys_nanosleep")
        b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"),
                        fn_name="do_sys_nanosleep")
        b["events"].open_ring_buffer(cb)
        subprocess.call(['sleep', '0.1'])
        b.ring_buffer_poll()
        self.assertGreater(self.counter, 0)
        b.cleanup()

    @skipUnless(kernel_version_ge(5,8), "requires kernel >= 5.8")
    def test_ringbuf_discard(self):
        self.counter = 0

        class Data(ct.Structure):
            _fields_ = [("ts", ct.c_ulonglong)]

        def cb(ctx, data, size):
            self.assertEqual(size, ct.sizeof(Data))
            event = ct.cast(data, ct.POINTER(Data)).contents
            self.counter += 1

        text = """
BPF_RINGBUF_OUTPUT(events, 8);
struct data_t {
    u64 ts;
};
int do_sys_nanosleep(void *ctx) {
    struct data_t *data = events.ringbuf_reserve(sizeof(struct data_t));
    if (!data)
        return 1;
    data->ts = bpf_ktime_get_ns();
    events.ringbuf_discard(data, 0);
    return 0;
}
"""
        b = BPF(text=text)
        b.attach_kprobe(event=b.get_syscall_fnname("nanosleep"),
                        fn_name="do_sys_nanosleep")
        b.attach_kprobe(event=b.get_syscall_fnname("clock_nanosleep"),
                        fn_name="do_sys_nanosleep")
        b["events"].open_ring_buffer(cb)
        subprocess.call(['sleep', '0.1'])
        b.ring_buffer_poll()
        self.assertEqual(self.counter, 0)
        b.cleanup()

if __name__ == "__main__":
    main()