Stream tool output through BenchExec for truncation prevention

benchexec/baseexecutor.py

@@ -11,6 +11,7 @@
 import subprocess
 import sys
 import threading
+import select
 
 from benchexec import __version__
 from benchexec import util
@@ -114,15 +115,19 @@ def pre_subprocess():
         p = subprocess.Popen(
             args,
             stdin=stdin,
-            stdout=stdout,
-            stderr=stderr,
+            stdout=None if stdout is None else subprocess.PIPE,
+            stderr=None if stderr is None else subprocess.PIPE,
             env=env,
             cwd=cwd,
             close_fds=True,
             preexec_fn=pre_subprocess,
         )
 
         def wait_and_get_result():
+            # Wait until all output has been processed before considering the process
+            # as terminated.
+            if stdout is not None or stderr is not None:
+                self._stream_output_with_selector(p, stdout, stderr)
             exitcode, ru_child = self._wait_for_process(p.pid, args[0])
             p.poll()
 
@@ -133,6 +138,70 @@ def wait_and_get_result():
 
         return p.pid, cgroups, wait_and_get_result
 
+    def _stream_output_with_selector(self, proc, stdout, stderr):
+        """Asynchronously read from stdout and stderr of the given process
+        and write to the given output target.
+        @param proc subprocess.Popen object whose output to monitor
+        @param stdout fileobj to send stdout
+        @param stderr fileobj to send stderr
+        """
+        # To address https://github.com/sosy-lab/benchexec/issues/535,
+        # the stdout and stderr of the executed process are monitored
+        # using select.
+        # This allows stdout and stderr to be processed asynchronously.
+
+        # Prepare file descriptors to monitor
+        fd_to_file = {}
+        if stdout is not None:
+            fd_to_file[proc.stdout.fileno()] = stdout
+        if stderr is not None:
+            fd_to_file[proc.stderr.fileno()] = stderr
+
+        # Use pidfd to monitor whether proc terminates
+        try:
+            pidfd = util.pidfd_open(proc.pid)
+        except OSError as e:
+            print(f"Failed to open pidfd: {e}")
+
+        try:
+            while True:
+                # fds to monitor
+                read_fds = list(fd_to_file.keys()) + [pidfd]
+                readables, _wlist, _xlist = select.select(read_fds, [], [])
+
+                for fd in readables:
+                    if fd == pidfd:
+                        continue
+                    out_stream = fd_to_file[fd]
+                    try:
+                        output = os.read(fd, 4096)
+                    except OSError:
+                        # fd might closed
+                        del fd_to_file[fd]
+                        continue
+                    if not output:
+                        # Reach EOF
+                        out_stream.flush()
+                        del fd_to_file[fd]
+                    else:
+                        print(output.decode().strip(), file=out_stream)
+
+                if pidfd in readables:
+                    # Flush all the output and exit if proc receives a signal
+                    for out_stream in fd_to_file.values():
+                        out_stream.flush()
+                    break
+
+                if not fd_to_file:
+                    # Exit if both stdout and stderr send EOF
+                    break
+        finally:
+            if proc.stdout is not None:
+                proc.stdout.close()
+            if proc.stderr is not None:
+                proc.stderr.close()
+            os.close(pidfd)
+
     def _wait_for_process(self, pid, name):
         """Wait for the given process to terminate.
         @return tuple of exit code and resource usage

benchexec/containerexecutor.py

@@ -773,12 +773,16 @@ def child():
                     grandchild_proc = subprocess.Popen(
                         args,
                         stdin=stdin,
-                        stdout=stdout,
-                        stderr=stderr,
+                        stdout=None if stdout is None else subprocess.PIPE,
+                        stderr=None if stderr is None else subprocess.PIPE,
                         env=env,
                         close_fds=False,
                         preexec_fn=grandchild,
                     )
+                    if stdout is not None or stderr is not None:
+                        self._stream_output_with_selector(
+                            grandchild_proc, stdout, stderr
+                        )
                 except (OSError, RuntimeError) as e:
                     logging.critical("Cannot start process: %s", e)
                     return CHILD_OSERROR

benchexec/util.py

@@ -842,3 +842,31 @@ def is_child_process_of_us(pid: int) -> bool:
         return is_child_process_of_us(ppid)
     else:
         return False
+
+
+def pidfd_open(pid, flags=0):
+    """
+    Return a file descriptor that refers to the process whose PID is specified in pid
+    @param pid a process id which the function refers to
+    @param flags either has the value 0, or contains the PIDFD_NONBLOCK (Linux >= 5.10)
+    @return: a file descriptor that referes to the `pid` process
+    """
+    if hasattr(os, "pidfd_open"):
+        # Use the built-in function if available (Python 3.9+ on Linux 5.3+)
+        try:
+            return os.pidfd_open(pid, flags)
+        except OSError as e:
+            raise RuntimeError(f"os.pidfd_open failed: {e}") from e
+    else:
+        libc = ctypes.CDLL("libc.so.6", use_errno=True)
+
+        # syscall number for pidfd_open on x86_64
+        SYS_pidfd_open = 434
+
+        fd = libc.syscall(SYS_pidfd_open, pid, flags)
+
+        if fd == -1:
+            err = ctypes.get_errno()
+            raise OSError(err, os.strerror(err))
+
+        return fd