fifo doesn't work

Author: nad2040

executor/executor.py

@@ -3,7 +3,6 @@
 import logging
 import subprocess
 import os
-import io
 
 from dataclasses import dataclass
 from executor_util import ptempfile, ptempdir, create_sandbox, copy, PASH_SPEC_TOP
@@ -12,9 +11,7 @@
 @dataclass
 class ExecCtxt:
     process: subprocess.Popen
-    trace_file: str       # base path; .r and .w are FIFOs, .missed is written at exit
-    read_fifo: io.IOBase  # open read-end of trace_file + ".r"
-    write_fifo: io.IOBase # open read-end of trace_file + ".w"
+    trace_file: str   # base path; .r/.w are streaming files, .missed written at exit
     outfds: str
     stderr: str
     pre_env_file: str
@@ -58,17 +55,6 @@ def run_trace_sandboxed(args: ExecArgs):
     logging.debug(f'Scheduler: Stdout file for: {args.concrete_node_id} is: {outfiles_dir}')
     logging.debug(f'Scheduler: Stderr file for: {args.concrete_node_id} is: {stderr_file}')
 
-    # Create FIFOs on the host before the sandbox starts so they land in the
-    # OverlayFS lower layer and trace_v3 (running inside) can open them for writing.
-    # Open the read ends now with O_NONBLOCK so trace_v3's open() doesn't block.
-    read_fifo_path  = trace_file + '.r'
-    write_fifo_path = trace_file + '.w'
-    os.mkfifo(read_fifo_path)
-    os.mkfifo(write_fifo_path)
-    _nonblock_open = lambda path, flags: os.open(path, flags | os.O_NONBLOCK)
-    read_fifo  = open(read_fifo_path,  'r', opener=_nonblock_open)
-    write_fifo = open(write_fifo_path, 'r', opener=_nonblock_open)
-
     sandbox_dir, tmp_dir = create_sandbox()
     post_execution_env_file = ptempfile(prefix='hs_post_env')
     lower_dirs_str = ':'.join(args.lower_sandboxes)
@@ -78,7 +64,7 @@ def run_trace_sandboxed(args: ExecArgs):
     logging.debug(cmd)
     process = subprocess.Popen(cmd, stdout=None, stderr=None, preexec_fn=set_pgid)
 
-    return ExecCtxt(process, trace_file, read_fifo, write_fifo, outfiles_dir, stderr_file, args.pre_execution_env_file, post_execution_env_file, sandbox_dir)
+    return ExecCtxt(process, trace_file, outfiles_dir, stderr_file, args.pre_execution_env_file, post_execution_env_file, sandbox_dir)
 
 def commit_workspace(workspace_path):
     run_script = f'{PASH_SPEC_TOP}/deps/try/try'

scheduler/node.py

@@ -292,9 +292,12 @@ class ConcreteNode:
     # Exists when node is in READY
     assignments: "list[NodeId]"
 
-    # Leftover partial line from each FIFO during incremental reads
-    read_fifo_buf: str
-    write_fifo_buf: str
+    # 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
 
     def __init__(self, cnid: ConcreteNodeId, node: Node, loop_list_context: HSLoopListContext,
                  spec_pre_env=None):
@@ -309,8 +312,7 @@ def __init__(self, cnid: ConcreteNodeId, node: Node, loop_list_context: HSLoopLi
         self.spec_pre_env = spec_pre_env
         self.loop_list_context = loop_list_context
         self.init_loop_list_context = loop_list_context
-        self.read_fifo_buf = ''
-        self.write_fifo_buf = ''
+        self._reset_stream_state()
 
     def __str__(self):
         return f'Node(id:{self.id_}, cmd:{self.cmd}, state:{self.state}, wait_env_file:{self.wait_env_file}, exec_ctxt:{self.exec_ctxt})'
@@ -451,12 +453,23 @@ def kill(self):
         assert self.state in [NodeState.EXECUTING, NodeState.SPEC_EXECUTING]
         self.exec_ctxt.process.kill()
 
-    def _drain_fifo(self, fifo, buf: str) -> tuple[set, str]:
-        """Non-blocking read from a FIFO; return (complete paths, leftover partial line)."""
+    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)
         try:
-            chunk = fifo.read()
-        except BlockingIOError:
-            return set(), buf
+            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
@@ -466,8 +479,15 @@ def _drain_fifo(self, fifo, buf: str) -> tuple[set, str]:
 
     def gather_fs_actions(self):
         assert self.state in [NodeState.EXECUTING, NodeState.SPEC_EXECUTING]
-        read_paths, self.read_fifo_buf  = self._drain_fifo(self.exec_ctxt.read_fifo,  self.read_fifo_buf)
-        write_paths, self.write_fifo_buf = self._drain_fifo(self.exec_ctxt.write_fifo, self.write_fifo_buf)
+        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)
 
     def get_rw_set(self):
@@ -652,8 +672,8 @@ def reset_to_ready(self, spec_pre_env: str = None, loop_list_context: HSLoopList
         self.loop_list_context = self.init_loop_list_context
         if spec_pre_env is not None:
             self.spec_pre_env = spec_pre_env
-        self.read_fifo_buf = ''
-        self.write_fifo_buf = ''
+        self._reset_stream_state()
+        
         self.state = NodeState.READY
         self.trace_state()
 
@@ -662,17 +682,17 @@ def start_executing(self, env_file):
 
         self.start_command(env_file)
         self.state = NodeState.EXECUTING
-        self.read_fifo_buf = ''
-        self.write_fifo_buf = ''
+        self._reset_stream_state()
+        
         self.rwset = RWSet(set(), set())
         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.read_fifo_buf = ''
-        self.write_fifo_buf = ''
+        self._reset_stream_state()
+        
         self.rwset = RWSet(set(), set())
         self.state = NodeState.SPEC_EXECUTING
         self.trace_state()
@@ -686,8 +706,10 @@ def collect_result(self):
     def commit_frontier_execution(self):
         assert self.state == NodeState.EXECUTING
         self.gather_fs_actions()
-        self.exec_ctxt.read_fifo.close()
-        self.exec_ctxt.write_fifo.close()
+        if self.read_stream_fd:
+            self.read_stream_fd.close()
+        if self.write_stream_fd:
+            self.write_stream_fd.close()
         self.update_loop_list_context()
         util.overhead_log(f"COMMIT|{self.cnid}")
         executor.commit_workspace(self.exec_ctxt.sandbox_dir)
@@ -699,12 +721,14 @@ def commit_frontier_execution(self):
         self.trace_state()
 
     def finish_spec_execution(self) -> bool:
-        """Drain FIFOs, close them, and return True if trace_v3 missed events."""
+        """Drain stream files, close them, and return True if trace_v3 missed events."""
         assert self.state == NodeState.SPEC_EXECUTING
         self.update_loop_list_context()
         self.gather_fs_actions()
-        self.exec_ctxt.read_fifo.close()
-        self.exec_ctxt.write_fifo.close()
+        if self.read_stream_fd:
+            self.read_stream_fd.close()
+        if self.write_stream_fd:
+            self.write_stream_fd.close()
         missed = dep_util.read_missed(self.exec_ctxt.sandbox_dir, self.exec_ctxt.trace_file)
         self.fixup_fds()
         self.state = NodeState.SPECULATED