streaming support attempt v2

  executor/executor.py
  - Imported sandboxed_path from executor_util
  - After create_sandbox(), pre-creates two FIFOs (.r, .w) in {sandbox_dir}/upperdir/{trace_file}.{r,w} — visible inside the sandbox at the same
  /tmp/... path because try's overlayfs upperdir preserves pre-created files

  scheduler/node.py
  - Added import threading
  - Replaced read_stream_fd/write_stream_fd/read_stream_buf/write_stream_buf fields with _reader_r/_reader_w thread fields
  - _reset_stream_state() — now joins any live threads with 0.5s timeout before clearing
  - _run_stream_reader(path, is_write) — opens FIFO blocking (waits for trace_v3), reads line-by-line, updates rwset.read_set or write_set
  directly (GIL ensures atomicity)
  - _start_reader_threads() — spawns two daemon threads immediately after start_command()
  - _join_reader_threads() — joins with 1.0s timeout; called before dep_util.read_missed()
  - Removed _open_stream_fd, _drain_stream, gather_fs_actions
  - commit_frontier_execution and finish_spec_execution now call _join_reader_threads() instead of the old drain+close pattern

  scheduler/partial_program_order.py
  - Removed fetch_fs_actions() method entirely
  - has_fs_deps() now calls _has_fs_deps() directly — rwsets are always current via reader threads
  - eager_fs_killing() no longer needs to poll before checking

Author: nad2040

executor/executor.py

@@ -5,7 +5,7 @@
 import os
 
 from dataclasses import dataclass
-from executor_util import ptempfile, ptempdir, create_sandbox, copy, PASH_SPEC_TOP
+from executor_util import ptempfile, ptempdir, create_sandbox, copy, PASH_SPEC_TOP, sandboxed_path
 
 
 @dataclass
@@ -56,6 +56,10 @@ def run_trace_sandboxed(args: ExecArgs):
     logging.debug(f'Scheduler: Stderr file for: {args.concrete_node_id} is: {stderr_file}')
 
     sandbox_dir, tmp_dir = create_sandbox()
+    for suffix in ('.r', '.w'):
+        fifo_path = sandboxed_path(sandbox_dir, trace_file + suffix)
+        os.makedirs(os.path.dirname(fifo_path), exist_ok=True)
+        os.mkfifo(fifo_path)
     post_execution_env_file = ptempfile(prefix='hs_post_env')
     lower_dirs_str = ':'.join(args.lower_sandboxes)
     speculate_mode = "speculate" if args.speculate_mode else "standard"

executor/executor_util.py

@@ -35,5 +35,16 @@ def create_sandbox():
     return sdir, tdir
 
 
+def sandboxed_path(sandbox_dir: str, path: str) -> str:
+    """Return the host-side path to a file written by a process inside the sandbox.
+
+    try(1) mounts an overlayfs for each root-level directory; writes to /foo/bar
+    inside the sandbox land in {sandbox_dir}/upperdir/foo/bar on the host.
+    """
+    if sandbox_dir:
+        return f"{sandbox_dir}/upperdir/{path}"
+    return path
+
+
 def copy(path_from, path_to):
     shutil.copy(path_from, path_to)

scheduler/node.py

@@ -1,17 +1,15 @@
 from enum import Enum, auto
-import logging
 import os
 import re
 import executor
 from executor import ExecCtxt, ExecResult, ExecArgs
 import dep_util
 import util
 import signal
+import threading
 from dataclasses import dataclass
 from typing import Tuple
-from enum import Enum, auto
 from pathlib import Path
-import util
 import analysis
 
 STATE_LOG = '[STATE_LOG] '
@@ -292,12 +290,9 @@ class ConcreteNode:
     # Exists when node is in READY
     assignments: "list[NodeId]"
 
-    # Open file descriptors for the streaming dep files (opened lazily in gather_fs_actions)
-    read_stream_fd: object
-    write_stream_fd: object
-    # Leftover partial line from each stream file during incremental reads
-    read_stream_buf: str
-    write_stream_buf: str
+    # Daemon reader threads — one per FIFO; started after exec_ctxt is set
+    _reader_r: "threading.Thread"
+    _reader_w: "threading.Thread"
 
     def __init__(self, cnid: ConcreteNodeId, node: Node, loop_list_context: HSLoopListContext,
                  spec_pre_env=None):
@@ -454,45 +449,44 @@ def kill(self):
         self.exec_ctxt.process.kill()
 
     def _reset_stream_state(self):
-        self.read_stream_fd = None
-        self.write_stream_fd = None
-        self.read_stream_buf = ''
-        self.write_stream_buf = ''
-
-    def _open_stream_fd(self, suffix):
-        path = util.sandboxed_path(self.exec_ctxt.sandbox_dir,
-                                   self.exec_ctxt.trace_file + suffix)
+        for t in (getattr(self, '_reader_r', None), getattr(self, '_reader_w', None)):
+            if t is not None and t.is_alive():
+                t.join(timeout=0.5)
+        self._reader_r = None
+        self._reader_w = None
+
+    def _run_stream_reader(self, path: str, is_write: bool):
         try:
-            return open(path, 'r')
-        except FileNotFoundError:
-            return None
-
-    def _drain_stream(self, fd, buf: str) -> tuple[set, str]:
-        """Read new content from fd; return (complete paths, leftover partial line)."""
-        chunk = fd.read()
-        if not chunk:
-            return set(), buf
-        buf += chunk
-        *complete, buf = buf.split('\n')
-        paths = {p for line in complete if (p := line.strip()) and not dep_util.should_filter(p)}
-        return paths, buf
-
-    def gather_fs_actions(self):
-        assert self.state in [NodeState.EXECUTING, NodeState.SPEC_EXECUTING]
-        if self.read_stream_fd is None:
-            self.read_stream_fd  = self._open_stream_fd('.r')
-        if self.write_stream_fd is None:
-            self.write_stream_fd = self._open_stream_fd('.w')
-        read_paths, write_paths = set(), set()
-        if self.read_stream_fd:
-            read_paths,  self.read_stream_buf  = self._drain_stream(self.read_stream_fd,  self.read_stream_buf)
-        if self.write_stream_fd:
-            write_paths, self.write_stream_buf = self._drain_stream(self.write_stream_fd, self.write_stream_buf)
-        self.update_rw_set(read_paths, write_paths)
+            fd = open(path, 'r')
+        except OSError:
+            return
+        try:
+            for line in fd:
+                p = line.rstrip('\n')
+                if p and not dep_util.should_filter(p):
+                    if is_write:
+                        self.rwset.write_set.add(p)
+                    else:
+                        self.rwset.read_set.add(p)
+        finally:
+            fd.close()
+
+    def _start_reader_threads(self):
+        sandbox_dir = self.exec_ctxt.sandbox_dir
+        trace_file  = self.exec_ctxt.trace_file
+        for suffix, is_write, attr in (('.r', False, '_reader_r'), ('.w', True, '_reader_w')):
+            path = util.sandboxed_path(sandbox_dir, trace_file + suffix)
+            t = threading.Thread(target=self._run_stream_reader, args=(path, is_write),
+                                 daemon=True)
+            t.start()
+            setattr(self, attr, t)
+
+    def _join_reader_threads(self):
+        for t in (getattr(self, '_reader_r', None), getattr(self, '_reader_w', None)):
+            if t is not None:
+                t.join(timeout=1.0)
 
     def get_rw_set(self):
-        # if self.state in [NodeState.EXECUTING, NodeState.SPEC_EXECUTING]:
-        #     self.gather_fs_actions()
         return self.rwset
 
     fd_line = re.compile(r'(\d+) ([rwd]) (\d+) (.+)')
@@ -550,7 +544,7 @@ def parse_env(content):
                     function_body_lines.append(line)
                     if line == '}':
                         inside_function = False
-                        if not current_function in ignore_vars:
+                        if current_function not in ignore_vars:
                             env_vars[current_function] = '\n'.join(function_body_lines)
                         function_body_lines = []
             return env_vars
@@ -683,17 +677,16 @@ def start_executing(self, env_file):
         self.start_command(env_file)
         self.state = NodeState.EXECUTING
         self._reset_stream_state()
-        
         self.rwset = RWSet(set(), set())
+        self._start_reader_threads()
         self.trace_state()
 
     def start_spec_executing(self, env_file, speculated_nodes):
-        # raise NotImplementedError
         assert self.state == NodeState.READY
         self.start_command(env_file, speculate=True, speculated_nodes=speculated_nodes)
         self._reset_stream_state()
-        
         self.rwset = RWSet(set(), set())
+        self._start_reader_threads()
         self.state = NodeState.SPEC_EXECUTING
         self.trace_state()
 
@@ -703,13 +696,11 @@ def collect_result(self):
         self.exec_result = ExecResult(self.exec_ctxt.process.returncode, self.exec_ctxt.process.pid)
         return self.exec_result.exit_code == 137, self.runtime_finished()
 
-    def commit_frontier_execution(self):
+    def commit_frontier_execution(self) -> int:
+        """Commit the frontier node and return the missed-event count from trace_v3."""
         assert self.state == NodeState.EXECUTING
-        self.gather_fs_actions()
-        if self.read_stream_fd:
-            self.read_stream_fd.close()
-        if self.write_stream_fd:
-            self.write_stream_fd.close()
+        self._join_reader_threads()
+        missed = dep_util.read_missed(self.exec_ctxt.sandbox_dir, self.exec_ctxt.trace_file)
         self.update_loop_list_context()
         util.overhead_log(f"COMMIT|{self.cnid}")
         executor.commit_workspace(self.exec_ctxt.sandbox_dir)
@@ -719,16 +710,13 @@ def commit_frontier_execution(self):
         self.fixup_fds()
         self.state = NodeState.COMMITTED
         self.trace_state()
+        return missed
 
     def finish_spec_execution(self) -> bool:
-        """Drain stream files, close them, and return True if trace_v3 missed events."""
+        """Join reader threads and return True if trace_v3 missed events."""
         assert self.state == NodeState.SPEC_EXECUTING
         self.update_loop_list_context()
-        self.gather_fs_actions()
-        if self.read_stream_fd:
-            self.read_stream_fd.close()
-        if self.write_stream_fd:
-            self.write_stream_fd.close()
+        self._join_reader_threads()
         missed = dep_util.read_missed(self.exec_ctxt.sandbox_dir, self.exec_ctxt.trace_file)
         self.fixup_fds()
         self.state = NodeState.SPECULATED

scheduler/partial_program_order.py

@@ -1,9 +1,6 @@
-from enum import Enum
 from node import NodeId, Node, CFGEdgeType, ConcreteNodeId, ConcreteNode, HSProg, HSBasicBlock, HSLoopListContext, loop_iters_do_action, get_loop_list_from_env
 import logging
 import util
-from pathlib import Path
-from collections import deque
 from executor import run_assignment_and_return_env_file
 
 PROG_LOG =  '[PROG_LOG] '
@@ -403,12 +400,6 @@ def adjust_to_be_resolved_dict(self):
     def valid(self):
         return True
 
-    def fetch_fs_actions(self):
-        for node in self.get_executing_normal_and_spec_nodes():
-            util.overhead_log(f"TRACE_FETCHING|{node.cnid}")
-            node.gather_fs_actions()
-            util.overhead_log(f"TRACE_FETCHING_END|{node.cnid}")
-
     def _has_fs_deps(self, concrete_node_id: ConcreteNodeId):
         node_of_interest : ConcreteNode = self.get_concrete_node(concrete_node_id)
         for dep_entry in self.to_be_resolved[concrete_node_id]:
@@ -423,10 +414,7 @@ def _has_fs_deps(self, concrete_node_id: ConcreteNodeId):
                 return True
         return False
 
-    # TODO: It's currently designed this way to avoid reading trace file all the time
-    # When we have complex caching code for this we can make this go away
     def has_fs_deps(self, concrete_node_id: ConcreteNodeId):
-        self.fetch_fs_actions()
         return self._has_fs_deps(concrete_node_id)
 
     def schedule_spec_work(self, concrete_node_id: ConcreteNodeId):
@@ -444,7 +432,7 @@ def simulate_var_assignments(self, env, assignments: "list[NodeId]"):
         return env
 
     def reset_speculation(self):
-        event_log(f"reset speculation")
+        event_log("reset speculation")
         for cnid in self.spec_exec_order:
             self.concrete_nodes[cnid].try_reset_to_ready()
         self.spec_exec_order = []
@@ -471,9 +459,12 @@ def handle_complete(self, concrete_node_id: ConcreteNodeId, has_pending_wait: bo
             node.transition_from_ready_to_unsafe()
             return
         if node.is_executing():
-            node.commit_frontier_execution()
+            missed = node.commit_frontier_execution()
             self.current_loop_list = node.loop_list_context
             self.adjust_to_be_resolved_dict()
+            if missed > 0:
+                util.debug_log(f"{concrete_node_id} frontier missed {missed} trace events — resetting speculation")
+                self.reset_speculation()
         elif node.is_spec_executing():
             if self.has_fs_deps(concrete_node_id):
                 node.reset_to_ready()
@@ -483,7 +474,8 @@ def handle_complete(self, concrete_node_id: ConcreteNodeId, has_pending_wait: bo
             else:
                 had_missed = node.finish_spec_execution()
                 if had_missed:
-                    util.debug_log(f"{concrete_node_id} missed trace events — moving to frontier")
+                    util.debug_log(f"{concrete_node_id} spec missed trace events — resetting speculation and moving to frontier")
+                    self.reset_speculation()
                     node.reset_to_ready(loop_list_context=self.current_loop_list)
                     node.start_executing(current_env)
                 elif has_pending_wait:
@@ -529,7 +521,7 @@ def handle_wait(self, concrete_node_id: ConcreteNodeId, env_file: str):
         else:
             self.reset_speculation()
 
-        if not concrete_node_id in self.concrete_nodes:
+        if concrete_node_id not in self.concrete_nodes:
             abstract_node = self.hsprog.find_node(concrete_node_id.node_id)
             new_concrete_node = ConcreteNode(concrete_node_id, abstract_node, self.current_loop_list)
             new_concrete_node.transition_from_init_to_ready(env_file)
@@ -593,7 +585,6 @@ def handle_wait(self, concrete_node_id: ConcreteNodeId, env_file: str):
     def eager_fs_killing(self):
         event_log("try to eagerly kill conflicted speculation")
         to_be_killed: "list[ConcreteNode]" = []
-        self.fetch_fs_actions()
         for node in self.get_all_nodes():
             if ((node.is_speculated() or node.is_spec_executing())
                 and self._has_fs_deps(node.cnid)):

scheduler/scheduler_server.py

@@ -9,8 +9,8 @@
 
 import util
 import config
-from partial_program_order import PartialProgramOrder, NodeId
-from node import LoopStack, ConcreteNodeId
+from partial_program_order import NodeId
+from node import ConcreteNodeId
 
 ##
 ## A scheduler server
@@ -110,14 +110,20 @@ def handle_wait(self, input_cmd: str, connection):
             util.debug_log(f'ignoring var assignment {concrete_node_id}')
             self.respond_to_wait_on_unsafe(concrete_node_id)
 
-    def process_next_cmd(self):
-        connection, input_cmd = util.socket_get_next_cmd(self.socket)
+    def process_next_cmd(self, timeout=None):
+        if timeout is not None:
+            result = util.socket_try_get_next_cmd(self.socket, timeout)
+            if result is None:
+                return
+            connection, input_cmd = result
+        else:
+            connection, input_cmd = util.socket_get_next_cmd(self.socket)
 
         if(input_cmd.startswith("Init")):
             connection.close()
             self.handle_init(input_cmd)
         elif (input_cmd.startswith("Daemon Start") or input_cmd == ""):
-            util.debug_log(f'Scheduler: Received daemon start message.')
+            util.debug_log('Scheduler: Received daemon start message.')
             connection.close()
         elif (input_cmd.startswith("CommandExecComplete:")):
             node_id, exec_id, sandbox_dir, trace_file = self.__parse_command_exec_x(input_cmd)
@@ -195,6 +201,8 @@ def __parse_command_exec_x(self, input_cmd: str) -> "tuple[int, int]":
     def schedule_work(self):
         self.partial_program_order.try_schedule_spec_nodes(self.window)
 
+    _POLL_INTERVAL = 0.010  # 10 ms — drain trace streams even when no message arrives
+
     def run(self):
         ## The first command should be the daemon start
         self.process_next_cmd()
@@ -204,7 +212,7 @@ def run(self):
 
         self.partial_program_order.log_state()
         while not self.done:
-            self.process_next_cmd()
+            self.process_next_cmd(timeout=self._POLL_INTERVAL)
             self.partial_program_order.log_state()
             self.schedule_work()
             self.partial_program_order.log_state()

scheduler/util.py

@@ -1,18 +1,17 @@
 import config
 import logging
 import os
+import select
 import socket
-import subprocess
 import tempfile
 import time
 import re
 import psutil
 import signal
 import analysis
 import shutil
-from node import Node, NodeId, LoopStack, HSProg, HSBasicBlock
+from node import Node, NodeId, HSProg
 from partial_program_order import PartialProgramOrder
-from config import PASH_SPEC_TMP_PREFIX
 
 DEBUG_LOG = '[DEBUG_LOG] '
 ENV_LOG = '[ENV_LOG] '
@@ -110,6 +109,13 @@ def socket_get_next_cmd(sock: socket.socket) -> "tuple[socket.socket, str]" :
 
     return (connection, str_data)
 
+def socket_try_get_next_cmd(sock: socket.socket, timeout: float) -> "tuple[socket.socket, str] | None":
+    """Like socket_get_next_cmd but returns None if no connection arrives within timeout seconds."""
+    ready, _, _ = select.select([sock], [], [], timeout)
+    if not ready:
+        return None
+    return socket_get_next_cmd(sock)
+
 def socket_respond(connection: socket.socket, message: str):
     bytes_message = message.encode('utf-8')
     connection.sendall(bytes_message)