Add ringbuf type hinting.

Author: varun-r-mallya

pythonbpf/maps/__init__.py

@@ -1,4 +1,4 @@
-from .maps import HashMap, PerfEventArray
+from .maps import HashMap, PerfEventArray, RingBuf
 from .maps_pass import maps_proc
 
-__all__ = ["HashMap", "PerfEventArray", "maps_proc"]
+__all__ = ["HashMap", "PerfEventArray", "maps_proc", "RingBuf"]

pythonbpf/maps/maps.py

@@ -1,3 +1,4 @@
+# This file provides type  and function hints only and does not actually give any functionality.
 class HashMap:
     def __init__(self, key, value, max_entries):
         self.key = key
@@ -33,3 +34,18 @@ def __init__(self, key_size, value_size):
 
     def output(self, data):
         pass  # Placeholder for output method
+
+
+class RingBuf:
+    def __init__(self, max_entries):
+        self.max_entries = max_entries
+
+    def reserve(self, size: int, flags=0):
+        if size > self.max_entries:
+            raise ValueError("size cannot be greater than set maximum entries")
+        return 0
+
+    def submit(self, data, flags=0):
+        pass
+
+    # add discard, output and also give names to flags and stuff

pythonbpf/maps/maps_pass.py

@@ -51,7 +51,7 @@ def create_bpf_map(module, map_name, map_params):
 
 
 def create_map_debug_info(module, map_global, map_name, map_params):
-    """Generate debug information metadata for BPF map"""
+    """Generate debug information metadata for BPF maps HASH and PERF_EVENT_ARRAY"""
     generator = DebugInfoGenerator(module)
 
     uint_type = generator.get_uint32_type()

tests/c-form/ex8.bpf.c

@@ -0,0 +1,30 @@
+#include "vmlinux.h"
+#include <bpf/bpf_helpers.h>
+#include <bpf/bpf_tracing.h>
+#include <bpf/bpf_core_read.h>
+
+struct {
+    __uint(type, BPF_MAP_TYPE_RINGBUF);
+    __uint(max_entries, 1 << 24); // 16 MB
+} rb SEC(".maps");
+
+//struct msg {
+//    u32 pid;
+//    char comm[16];
+//};
+
+//SEC("tracepoint/syscalls/sys_enter_execve")
+//int handle_execve(struct trace_event_raw_sys_enter *ctx)
+//{
+//    struct msg *m;
+//    m = bpf_ringbuf_reserve(&rb, sizeof(*m), 0);
+//    if (!m)
+//        return 0;
+//
+//    m->pid = bpf_get_current_pid_tgid() >> 32;
+//    bpf_get_current_comm(&m->comm, sizeof(m->comm));
+//    bpf_ringbuf_submit(m, 0);
+//    return 0;
+//}
+
+//char LICENSE[] SEC("license") = "GPL";

tests/c-form/ringbuf.bpf.c

@@ -0,0 +1,45 @@
+from pythonbpf import bpf, map, struct, section, bpfglobal, compile
+from pythonbpf.helpers import ktime, pid
+from pythonbpf.maps import PerfEventArray
+
+from ctypes import c_void_p, c_int32, c_uint64
+
+
+# PLACEHOLDER EXAMPLE. THIS SHOULD TECHNICALLY STILL FAIL TESTS
+@bpf
+@struct
+class data_t:
+    pid: c_uint64
+    ts: c_uint64
+    comm: str(16)
+
+
+@bpf
+@map
+def events() -> PerfEventArray:
+    return PerfEventArray(key_size=c_int32, value_size=c_int32)
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_clone")
+def hello(ctx: c_void_p) -> c_int32:
+    dataobj = data_t()
+    ts = ktime()
+    strobj = "hellohellohello"
+    dataobj.pid = pid()
+    dataobj.ts = ktime()
+    # dataobj.comm = strobj
+    print(
+        f"clone called at {dataobj.ts} by pid {dataobj.pid}, comm {strobj} at time {ts}"
+    )
+    events.output(dataobj)
+    return c_int32(0)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()

tests/failing_tests/perf_buffer_map.py

@@ -0,0 +1,25 @@
+from pythonbpf import bpf, map, bpfglobal, section
+from pythonbpf.maps import RingBuf
+from ctypes import c_int32, c_void_p
+
+
+# Define a map
+@bpf
+@map
+def mymap() -> RingBuf:
+    return RingBuf(max_entries=(1 << 24))
+
+
+@bpf
+@section("tracepoint/syscalls/sys_enter_clone")
+def testing(ctx: c_void_p) -> c_int32:
+    return c_int32(0)
+
+
+@bpf
+@bpfglobal
+def LICENSE() -> str:
+    return "GPL"
+
+
+compile()