Remove the separate io_uring trace stream; keep fs mirror

io_uring file read/writes are already mirrored into the fs/VFS trace
(_mirror_io_uring_to_fs), so the standalone io_uring/ output duplicated that
data. Drop the separate stream entirely:

- IOTracer._print_event_io_uring now only resolves file identity and mirrors
  completed READ/WRITE ops into the fs trace; it no longer builds the full
  io_uring row or calls append_io_uring_log. The BPF io_uring probe/perf
  buffer stay (they feed the mirror).
- WriterManager: remove the io_uring output file/dir, buffer, handle, lock,
  stream-registry entry, thresholds, dynamic_limits, adaptive-sizing branch,
  should_flush/append/flush methods, write_to_disk thread, force_flush
  compress, and close_handles entry.

Note: non-file io_uring ops (network/poll), the submit lifecycle, and
read/writes whose inode didn't resolve are no longer recorded, matching the
intent that io_uring is represented via the fs trace.

Author: claude

src/tracer/IOTracer.py

@@ -792,14 +792,14 @@ def _print_event_pagefault(self, cpu, data, size):
 
     def _print_event_io_uring(self, cpu, data, size):
         """
-        Callback for processing io_uring events from the perf buffer.
-        
-        Captures io_uring async I/O operations including:
-        - ENTER: io_uring_enter syscall
-        - SUBMIT: Individual SQE submissions
-        - COMPLETE: Request completions with latency
-        - WORKER: Async worker executions
-        
+        Callback for io_uring perf events.
+
+        The standalone io_uring trace stream has been removed. io_uring file
+        activity is surfaced by mirroring completed READ/WRITE operations into
+        the fs/VFS trace (they call ->read_iter/->write_iter directly, bypass
+        vfs_read/vfs_write, and would otherwise be invisible). Other io_uring
+        events (submit lifecycle, network/poll ops) are no longer recorded.
+
         Args:
             cpu: CPU number where the event was captured
             data: Raw event data pointer
@@ -808,58 +808,23 @@ def _print_event_io_uring(self, cpu, data, size):
         self._tick_maintenance()
 
         e = self.b["io_uring_events"].event(data)
+
+        # Only completed ops (event_type == 2) carry a result/latency and feed
+        # the fs mirror; nothing else needs recording now that the separate
+        # io_uring stream is gone.
+        if e.event_type != 2:
+            return
+
         ts = datetime.today()
-        
         comm = e.comm.decode("utf-8", errors="replace").strip("\x00")
-        
+
         if self._should_filter_process(comm):
             return
-            
-        event_type = self.flag_mapper.format_io_uring_event_type(e.event_type)
-        opcode = self.flag_mapper.format_io_uring_opcode(e.opcode) if e.opcode else ""
-        enter_flags = self.flag_mapper.format_io_uring_enter_flags(e.enter_flags) if e.enter_flags else ""
-        sqe_flags = self.flag_mapper.format_io_uring_sqe_flags(e.sqe_flags) if e.sqe_flags else ""
-        
-        # Format fields based on event type
-        ring_fd = str(e.ring_fd) if e.ring_fd else ""
-        ring_ptr = hex(e.ring_ptr) if e.ring_ptr else ""
-        to_submit = str(e.to_submit) if e.to_submit else ""
-        min_complete = str(e.min_complete) if e.min_complete else ""
-        
-        req_ptr = hex(e.req_ptr) if e.req_ptr else ""
-        user_data = str(e.user_data) if e.user_data else ""
-        fd = str(e.fd) if e.fd != 0 and e.fd != -1 else ""
-        length = str(e.len) if e.len else ""
-        offset = str(e.offset) if e.offset else ""
-        ioprio = str(e.ioprio) if e.ioprio else ""
-        buf_index = str(e.buf_index) if e.buf_index else ""
-        personality = str(e.personality) if e.personality else ""
-        
-        result = str(e.result) if e.result != 0 or e.event_type == 2 else ""
-        is_error = "1" if e.is_error else ""
-        cqe_errno = str(e.cqe_errno) if e.cqe_errno else ""
-        
-        submit_ts = str(e.submit_ts_ns) if e.submit_ts_ns else ""
-        complete_ts = str(e.complete_ts_ns) if e.complete_ts_ns else ""
-        latency_ns = str(e.latency_ns) if e.latency_ns else ""
-        
-        worker_pid = str(e.worker_pid) if e.worker_pid else ""
-        worker_tid = str(e.worker_tid) if e.worker_tid else ""
-        worker_cpu = str(e.worker_cpu) if e.worker_cpu else ""
-        is_async = "1" if e.is_async else ""
-        
-        sq_head = str(e.sq_head) if e.sq_head else ""
-        sq_tail = str(e.sq_tail) if e.sq_tail else ""
-        cq_head = str(e.cq_head) if e.cq_head else ""
-        cq_tail = str(e.cq_tail) if e.cq_tail else ""
-        sq_depth = str(e.sq_depth) if e.sq_depth else ""
-        cq_depth = str(e.cq_depth) if e.cq_depth else ""
 
         # File correlation — the prep probe records the backing file's inode,
-        # device and filesystem for file-backed ops. Resolve the path from the
-        # inode→path cache populated by OPEN events (same strategy as VFS events)
-        # so io_uring I/O carries the same file identity as the fs trace.
-        inode_val = e.inode if getattr(e, "inode", 0) else ""
+        # device and filesystem. Resolve the path from the inode→path cache
+        # (same strategy as VFS events) so mirrored io_uring I/O carries the
+        # same file identity as the fs trace.
         dev_val = self._format_dev(e.dev) if getattr(e, "dev", 0) else ""
         fs_type_val = (
             self.flag_mapper.format_fs_type(e.fs_magic)
@@ -873,59 +838,9 @@ def _print_event_io_uring(self, cpu, data, size):
             if self.anonymous and filename:
                 filename = hash_filename_in_path(Path(filename))
 
-        # Surface io_uring file READ/WRITE in the main fs/VFS trace stream so
-        # async I/O appears alongside syscall reads/writes. Mirror only on
-        # COMPLETE (result/latency known) and only read/write opcodes, which
-        # bypass vfs_read/vfs_write and would otherwise be invisible there.
-        if e.event_type == 2:
-            self._mirror_io_uring_to_fs(e, comm, filename, ts, dev_val, fs_type_val)
+        # Mirror completed file READ/WRITE into the main fs/VFS trace stream.
+        self._mirror_io_uring_to_fs(e, comm, filename, ts, dev_val, fs_type_val)
 
-        # Build CSV row matching the unified schema from the guide
-        output = format_csv_row(
-            ts.strftime("%Y-%m-%d %H:%M:%S.%f"),
-            str(e.timestamp_ns),
-            event_type,
-            str(e.pid),
-            str(e.tid),
-            comm,
-            str(e.cpu),
-            ring_fd,
-            ring_ptr,
-            to_submit,
-            min_complete,
-            enter_flags,
-            req_ptr,
-            user_data,
-            opcode,
-            fd,
-            length,
-            offset,
-            sqe_flags,
-            ioprio,
-            buf_index,
-            personality,
-            result,
-            is_error,
-            cqe_errno,
-            submit_ts,
-            complete_ts,
-            latency_ns,
-            worker_pid,
-            worker_tid,
-            worker_cpu,
-            is_async,
-            sq_head,
-            sq_tail,
-            cq_head,
-            cq_tail,
-            sq_depth,
-            cq_depth,
-            inode_val,
-            filename,
-            dev_val,
-            fs_type_val,
-        )
-        self.writer.append_io_uring_log(output)
 
     def _mirror_io_uring_to_fs(self, e, comm, filename, ts, dev_val, fs_type_val):
         """

src/tracer/WriterManager.py

@@ -83,7 +83,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self.output_process_file = f"{self.output_dir}/process/process_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
         self.output_fs_snapshot_file = f"{self.output_dir}/filesystem_snapshot/filesystem_snapshot_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
         self.output_pagefault_file = f"{self.output_dir}/pagefault/pagefault_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
-        self.output_io_uring_file = f"{self.output_dir}/io_uring/io_uring_{self.current_datetime.strftime('%Y%m%d_%H%M%S_%f')[:-3]}.csv"
 
         # Create output directories
         os.makedirs(f"{self.output_dir}/system_spec", exist_ok=True)
@@ -93,7 +92,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         os.makedirs(f"{self.output_dir}/process", exist_ok=True)
         os.makedirs(f"{self.output_dir}/filesystem_snapshot", exist_ok=True)
         os.makedirs(f"{self.output_dir}/pagefault", exist_ok=True)
-        os.makedirs(f"{self.output_dir}/io_uring", exist_ok=True)
 
         self.upload_manager = upload_manager
         self.automatic_upload = automatic_upload
@@ -105,7 +103,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self.process_buffer = deque()
         self.fs_snap_buffer = deque()
         self.pagefault_buffer = deque()
-        self.io_uring_buffer = deque()
 
         # Event rate tracking
         self.event_timestamps = {
@@ -115,7 +112,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
             'fs_state': deque(maxlen=1000),
             'proc_state': deque(maxlen=1000),
             'pagefault': deque(maxlen=1000),
-            'io_uring': deque(maxlen=1000),
         }
 
         # Dynamic thresholds (min, max). Raised roughly 10x from the original
@@ -130,7 +126,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
             'fs_state': (80000, 200000),
             'proc_state': (80000, 100000),
             'pagefault': (80000, 400000),
-            'io_uring': (80000, 400000),
         }
 
         # Start adaptive sizing thread
@@ -152,7 +147,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self.process_max_events = 80000  # Large enough to fit entire hourly snapshot
         self.fs_snap_max_events = 80000
         self.pagefault_max_events = 80000
-        self.io_uring_max_events = 80000
 
         # Per-stream locks. Buffer flushes are triggered both from the
         # perf-callback (polling) thread via append_*_log -> flush_*_only and
@@ -166,7 +160,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
             'process':   threading.Lock(),
             'fs_snap':   threading.Lock(),
             'pagefault': threading.Lock(),
-            'io_uring':  threading.Lock(),
         }
 
         # File handles for each output
@@ -176,7 +169,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
         self._process_handle = None
         self._pagefault_handle = None
         self._fs_snap_handle = None
-        self._io_uring_handle = None
 
         # Registry of the continuous event streams that support generic
         # rotation. Snapshots (process, fs_snap) are intentionally excluded:
@@ -188,7 +180,6 @@ def __init__(self, output_dir: str, upload_manager: ObjectStorageManager, automa
             'block':     {'subdir': 'ds',        'prefix': 'ds',        'buf': 'block_buffer',     'handle': '_block_handle',     'file': 'output_block_file',     'log': 'Block'},
             'cache':     {'subdir': 'cache',     'prefix': 'cache',     'buf': 'cache_buffer',     'handle': '_cache_handle',     'file': 'output_cache_file',     'log': 'Cache'},
             'pagefault': {'subdir': 'pagefault', 'prefix': 'pagefault', 'buf': 'pagefault_buffer', 'handle': '_pagefault_handle', 'file': 'output_pagefault_file', 'log': 'PageFault'},
-            'io_uring':  {'subdir': 'io_uring',  'prefix': 'io_uring',  'buf': 'io_uring_buffer',  'handle': '_io_uring_handle',  'file': 'output_io_uring_file',  'log': 'IO_Uring'},
         }
 
         # Time/size based rotation so a slow stream's log doesn't wait until
@@ -247,7 +238,7 @@ def _adaptive_sizing(self):
         while True:
             time.sleep(10)  
 
-            for event_type in ['vfs', 'block', 'cache', 'fs_state','proc_state', 'pagefault', 'io_uring']:
+            for event_type in ['vfs', 'block', 'cache', 'fs_state','proc_state', 'pagefault']:
                 rate = self._calculate_event_rate(event_type)
                 min_limit, max_limit = self.dynamic_limits[event_type]
 
@@ -272,8 +263,6 @@ def _adaptive_sizing(self):
                     self.process_max_events = new_limit
                 elif event_type == 'pagefault':
                     self.pagefault_max_events = new_limit
-                elif event_type == 'io_uring':
-                    self.io_uring_max_events = new_limit
 
     def _periodic_flush(self):
         """
@@ -330,9 +319,6 @@ def _log_status(self):
             buffer_info.append(f"Cache:{len(self.cache_buffer)}")
         if len(self.pagefault_buffer) > 0:
             buffer_info.append(f"PgFault:{len(self.pagefault_buffer)}")
-        if len(self.io_uring_buffer) > 0:
-            buffer_info.append(f"IO_Uring:{len(self.io_uring_buffer)}")
-        
         if buffer_info:
             status_parts.append(f"Buffers: {', '.join(buffer_info)}")
 
@@ -391,10 +377,6 @@ def should_flush_pagefault(self) -> bool:
         """Check if pagefault buffer should be flushed."""
         return (len(self.pagefault_buffer) >= self.pagefault_max_events)
 
-    def should_flush_io_uring(self) -> bool:
-        """Check if io_uring buffer should be flushed."""
-        return (len(self.io_uring_buffer) >= self.io_uring_max_events)
-
     def append_fs_snap_log(self, log_output: str):
         """
         Add a filesystem snapshot log entry.
@@ -497,16 +479,6 @@ def append_pagefault_log(self, log_output: str):
         else:
             logger("error", "Invalid pagefault log output format. Expected a string.")
 
-    def append_io_uring_log(self, log_output: str):
-        """Add an io_uring event log entry."""
-        if isinstance(log_output, str):
-            self.io_uring_buffer.append(log_output)
-            self.event_timestamps['io_uring'].append(time.time())
-            if self.should_flush_io_uring():
-                self.flush_io_uring_only()
-        else:
-            logger("error", "Invalid io_uring log output format. Expected a string.")
-
     def direct_write(self, output_path: str, spec_str: str):
         """
         Write a system specification file directly.
@@ -708,10 +680,6 @@ def flush_pagefault_only(self):
         """Flush pagefault buffer to file (rotate + compress + upload)."""
         self._rotate_stream('pagefault')
 
-    def flush_io_uring_only(self):
-        """Flush io_uring buffer to file (rotate + compress + upload)."""
-        self._rotate_stream('io_uring')
-
     def _rotate_stream(self, key: str):
         """Rotate one continuous stream's current log and queue it for upload.
 
@@ -827,7 +795,6 @@ def force_flush(self):
             self.fs_snap_buffer.clear()
 
         self.compress_log(self.output_pagefault_file)
-        self.compress_log(self.output_io_uring_file)
         self.compress_dir(self.output_dir)
 
 
@@ -840,7 +807,6 @@ def clear_events(self):
         self.process_buffer.clear()
         self.fs_snap_buffer.clear()
         self.pagefault_buffer.clear()
-        self.io_uring_buffer.clear()
 
     def _write_buffer_to_file(self, buffer, file_handle, buffer_name: str):
         """
@@ -915,13 +881,6 @@ def write_pagefault():
                         self._pagefault_handle = open(self.output_pagefault_file, 'a', buffering=8192)
                     self._write_buffer_to_file(self.pagefault_buffer, self._pagefault_handle, "PageFault")
 
-        def write_io_uring():
-            with self._stream_locks['io_uring']:
-                if self.io_uring_buffer:
-                    if self._io_uring_handle is None:
-                        self._io_uring_handle = open(self.output_io_uring_file, 'a', buffering=8192)
-                    self._write_buffer_to_file(self.io_uring_buffer, self._io_uring_handle, "IO_Uring")
-
         threads = []
 
         # Start parallel write threads for each buffer
@@ -955,11 +914,6 @@ def write_io_uring():
             threads.append(t7)
             t7.start()
 
-        if self.io_uring_buffer:
-            t13 = threading.Thread(target=write_io_uring)
-            threads.append(t13)
-            t13.start()
-
         # Wait for all threads to complete
         for thread in threads:
             thread.join()
@@ -1032,7 +986,6 @@ def close_handles(self):
             (self._process_handle, "Process State"),
             (self._fs_snap_handle, "Filesystem Snapshot"),
             (self._pagefault_handle, "PageFault"),
-            (self._io_uring_handle, "IO_Uring"),
         ]
 
         for handle, name in handles:
@@ -1050,4 +1003,3 @@ def close_handles(self):
         self._process_handle = None
         self._fs_snap_handle = None
         self._pagefault_handle = None
-        self._io_uring_handle = None

tests/test_writer_upload.py

@@ -117,7 +117,6 @@ def test_thresholds_are_enlarged(self):
         self.assertGreaterEqual(self.wm.block_max_events, 80000)
         self.assertGreaterEqual(self.wm.cache_max_events, 100000)
         self.assertGreaterEqual(self.wm.pagefault_max_events, 80000)
-        self.assertGreaterEqual(self.wm.io_uring_max_events, 80000)
         # Dynamic minimums must stay consistent (min <= max) after enlargement.
         for name, (lo, hi) in self.wm.dynamic_limits.items():
             self.assertGreaterEqual(lo, 80000, name)