Skip to content

Stream tool output through BenchExec for truncation prevention #1170

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
t0hsumi wants to merge 12 commits into
base: main
Choose a base branch
from
Draft

Stream tool output through BenchExec for truncation prevention #1170

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
6526b5d
Add the pipe functionality to BaseExecutor class.
t0hsumi Jun 19, 2025
3d02db1
Incorporate the resource cleanups in `wait_for_read()`
t0hsumi Jun 19, 2025
ed72686
Make `_stream_output_with_selector()`
t0hsumi Jun 19, 2025
a6fde54
Flush the output after finishing reading
t0hsumi Jun 20, 2025
8e1044b
Pipe the output in `ContainerExecutor()`
t0hsumi Jun 20, 2025
dbdd8de
Delete non-blocking update to process stdout/stderr
t0hsumi Jun 20, 2025
6c953c1
Disable pipe if the output file is not specified
t0hsumi Jun 20, 2025
8a34a3e
Remove unused import
t0hsumi Jun 20, 2025
65d4b7c
Enable to exit if there is no event
t0hsumi Jun 25, 2025
7b3c030
Continue when `selector.select()` returns empty list
t0hsumi Jun 30, 2025
eab95e4
Add the explanation of the argments
t0hsumi Jun 30, 2025
07d7036
Use pidfd_open to monitor whether the process terminates
t0hsumi Jun 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 46 additions & 2 deletions benchexec/baseexecutor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import subprocess
import sys
import threading
import selectors

from benchexec import __version__
from benchexec import util
Expand Down Expand Up @@ -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()

Expand All @@ -133,6 +138,45 @@ 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
"""
# Set output selector
# To address https://github.com/sosy-lab/benchexec/issues/535,
# the stdout and stderr of the executed process are connected to
# selectors.DefaultSelector().
# This allows stdout and stderr to be processed asynchronously.
selector = selectors.DefaultSelector()

# Register the executed command output with the selector,
# associating them with their respective file objects
if stdout is not None:
selector.register(proc.stdout, selectors.EVENT_READ, data=stdout)
if stderr is not None:
selector.register(proc.stderr, selectors.EVENT_READ, data=stderr)

try:
while True:
for key, _unused_event in selector.select():
# Block until output is ready
output = os.read(key.fileobj.fileno(), 4096)
Copy link
Contributor Author

@t0hsumi t0hsumi Jun 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’ve currently set the read size to 4096 bytes. However, we may need to evaluate whether this is optimal in terms of performance.

Copy link
Member

@PhilippWendler PhilippWendler Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess it is fine. I wouldn't know a better number either.

if not output:
key.data.flush()
selector.unregister(key.fileobj)
else:
print(output.decode().strip(), file=key.data)
if not selector.get_map():
# Exit if both stdout and stderr are closed
break
finally:
if proc.stdout is not None:
proc.stdout.close()
if proc.stderr is not None:
proc.stderr.close()
selector.close()

def _wait_for_process(self, pid, name):
"""Wait for the given process to terminate.
@return tuple of exit code and resource usage
Expand Down
8 changes: 6 additions & 2 deletions benchexec/containerexecutor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Comment on lines +782 to +785
Copy link
Contributor Author

@t0hsumi t0hsumi Jun 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we call self._stream_output_with_selector() here and the tool does not send an EOF but is instead terminated by a signal, the procedure gets blocked in this function (more precisely, in selector.select() within self._stream_output_with_selector()), and runexec does not terminate. (See this pytest error case.)

To handle this properly, we need to process stdout and stderr while also checking for walltime limits and signals. In the implementation, we need to integrate self._stream_output_with_selector() with wait_for_child_and_forward_signals(). However, as far as I can tell, there are two possible solutions—neither of which seems ideal:

  1. Set a timeout in selector.select()
  2. Use an additional thread

Copy link
Member

@PhilippWendler PhilippWendler Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you make select wait for termination of a process? I think that might be possible and should solve the problem.

Copy link
Contributor Author

@t0hsumi t0hsumi Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m trying to address the pytest error case. According to the selectors documentation, the latest code should detect when the events list is empty. An empty list indicates that "this method can return before any file object becomes ready or the timeout has elapsed if the current process receives a signal."

However, in the pytest error case, when the timeout limit is exceeded, a SIGKILL is sent to the grandchild process. Despite this, selector.select() does not return an empty list in response, which contradicts the expected behavior. I don’t yet understand the reason for this.

As far as I can tell from the documentation, there seems to be no other way within BenchExec to detect process termination using only the selectors module.

Copy link
Member

@PhilippWendler PhilippWendler Jun 25, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the documentation talks about the case if the select()ing process receives a signal. But this is not our case.

I would assume that the pipe is closed if the only listening process terminates, and that this causes select() to return. Can you check if there are any other processes that still have the same pipe open, for example BenchExec itself? BenchExec should only have the read ends of the pipe open, and the benchmarked process only the write ends.

Another way could be to add a file descriptor that signals if the process ends. Maybe we can do this with a pidfd. Or the part of the code that calls wait could close the pipes, such that select() returns?

Copy link
Contributor Author

@t0hsumi t0hsumi Jun 30, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry for the late reply.

I think the documentation talks about the case if the select()ing process receives a signal. But this is not our case.

Thanks for pointing that out — I misunderstood. I've now updated the code to retry select() when it returns an empty list.

Can you check if there are any other processes that still have the same pipe open, for example BenchExec itself?

Which pipe do you mean? I assume you're referring to proc.stdout and proc.stderr, but I couldn't find any indication that BenchExec is reading from those.

Another way could be to add a file descriptor that signals if the process ends. Maybe we can do this with a pidfd. Or the part of the code that calls wait could close the pipes, such that select() returns?

I’m now using pidfd_open() together with the select module. It works fine in my environment.

Copy link
Member

@PhilippWendler PhilippWendler Jul 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you check if there are any other processes that still have the same pipe open, for example BenchExec itself?

Which pipe do you mean? I assume you're referring to proc.stdout and proc.stderr, but I couldn't find any indication that BenchExec is reading from those.

Yes, I was referring to stdout and stderr. These should be pipes, like those returned by os.pipe(). Whether something is reading from them is not important, the question is whether there exists any process that still has an open file descriptor that is referring to their write end (after opening a pipe consists of two file descriptors, which can be passed around and closed individually). I would think (hope) that as soon as the last process terminates that has the write end open, the pipe is closed and select() should tell us.

Another way could be to add a file descriptor that signals if the process ends. Maybe we can do this with a pidfd. Or the part of the code that calls wait could close the pipes, such that select() returns?

I’m now using pidfd_open() together with the select module. It works fine in my environment.

This is nice! I general I think that PID fds are great and eventually BenchExec will use them anyway. However, they are available only since Linux 5.1, and this is still somewhat new. I would prefer if we do not introduce a hard dependency. So if we find a solution without PID fs, it would be nicer. I am pretty sure there is such a way, because what we need seems like a highly common problem for subprocess spawning and many tools have needed this for a long time even before the introduction of PID fs.

Copy link
Contributor Author

@t0hsumi t0hsumi Jul 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I was referring to stdout and stderr. These should be pipes, like those returned by os.pipe(). Whether something is reading from them is not important, the question is whether there exists any process that still has an open file descriptor that is referring to their write end (after opening a pipe consists of two file descriptors, which can be passed around and closed individually). I would think (hope) that as soon as the last process terminates that has the write end open, the pipe is closed and select() should tell us.

Alright, I’ll check which process is opening which file descriptors.

except (OSError, RuntimeError) as e:
logging.critical("Cannot start process: %s", e)
return CHILD_OSERROR
Expand Down