Improve multiprocessing: instance-specific logging, stdout/stderr capture, and proper Ctrl+C cleanup

[xingyaoww]

Summary

This PR improves the multiprocessing evaluation system with better logging, output capture, and cleanup handling.

Key Changes

1. Instance-Specific Logging

• Replaced PID-based logs with instance ID logs (instance_{id}.log instead of pid_{pid}.log)
• Added reset_logger_for_multiprocessing() function that:
  • Creates separate log files per instance
  • Shows one informative console message per instance with tail -f hint
  • Only logs WARNING+ to console after initial message
  • Logs INFO+ to instance-specific file
• Removed redundant _child_init() function

2. Auto-Save Metadata

• Added model_post_init() to Evaluation class
• Automatically saves metadata.json on initialization
• Ensures metadata is available even if evaluation is interrupted early

3. Complete Output Capture

• Added redirect_stdout_stderr() context manager
• Captures ALL output to instance log files:
  • SDK visualizations and progress bars
  • Print statements from agent code
  • Any console output from subprocesses
• Modified _process_one_mp() to wrap execution with redirection

4. Proper KeyboardInterrupt (Ctrl+C) Handling

• Explicitly terminates worker processes on interrupt
• Accesses ProcessPoolExecutor._processes to call terminate() on each worker
• Prevents zombie processes from lingering
• Clean shutdown on both KeyboardInterrupt and general exceptions

5. Comprehensive Test Suite

• Added tests/test_keyboard_interrupt.py with two test cases:
  1. test_keyboard_interrupt_cleanup: Validates all workers terminate after SIGINT during normal operation
  2. test_keyboard_interrupt_immediate: Validates cleanup works when interrupt happens early
• Tests use psutil to verify no orphaned processes remain
• All tests passing ✅

Benefits

• Better debugging: Each instance has its own complete log file with ALL output
• Cleaner console: No interleaved output from multiple workers
• No zombies: Ctrl+C properly terminates all child processes
• Better UX: Clear hints on how to monitor individual instances
• More reliable: Metadata always saved, even on early interruption

Testing

# Run the KeyboardInterrupt cleanup tests
uv run pytest tests/test_keyboard_interrupt.py -v

# Both tests pass:
# ✅ test_keyboard_interrupt_cleanup
# ✅ test_keyboard_interrupt_immediate

Example Output

When running evaluation, each instance now shows:

Instance test_instance_1 - 2025-01-13 20:00:00 - INFO - Starting evaluation for instance test_instance_1.
Hint: run "tail -f /path/to/logs/instance_test_instance_1.log" to see live logs in a separate shell

The instance log file contains all output including SDK visualizations:

$ tail -f logs/instance_test_instance_1.log
2025-01-13 20:00:01 - INFO - Running with runtime: modal-0.1.0
2025-01-13 20:00:02 - INFO - Running action: MessageAction(content='Let me check the repository structure', ...)
[Progress bars, visualizations, and all output captured here]

[xingyaoww]

@OpenHands /codereview-roasted help me review this PR

[openhands-ai]

I'm on it! xingyaoww can track my progress at all-hands.dev

[openhands-ai]

🔴 Code Review: PR #93 - Multiprocessing Improvements

Taste Rating: 🟡 Acceptable - Solves real problems but with some compromises on elegance

---

[CRITICAL ISSUES] (Must address or document)

[benchmarks/utils/evaluation.py, Line 341-342] Breaking Encapsulation

if not wait and hasattr(pool, "_processes") and pool._processes:
    for process in pool._processes.values():

Problem: You're reaching into ProcessPoolExecutor._processes, a private attribute. This is fragile and could break in any Python version update.

Why this matters: The underscore prefix means "this is an implementation detail, don't touch it." You're making your code dependent on CPython internals. Tomorrow Python 3.13 could change this and your code breaks.

The fundamental issue: You're using the wrong abstraction. If you need to forcefully terminate worker processes, ProcessPoolExecutor isn't giving you what you need through its public API.

Options:

1. Document the risk: Add a big fat comment explaining this is intentionally breaking encapsulation because ProcessPoolExecutor doesn't expose process termination, and include the Python version you tested against
2. Use multiprocessing.Pool instead: It gives you explicit control over worker processes with .terminate() as a public method
3. Accept that cleanup might not be immediate: Use pool.shutdown(wait=False, cancel_futures=True) and let the OS clean up eventually

Recommendation: At minimum, add defensive code and a comment:

# HACK: ProcessPoolExecutor doesn't expose worker termination in its public API.
# We need to forcefully terminate workers on Ctrl+C to prevent zombie processes.
# This accesses private _processes attribute. Tested on Python 3.9-3.11.
if not wait:
    try:
        # Attempt to access private _processes (may break in future Python versions)
        if hasattr(pool, "_processes") and pool._processes:
            for process in pool._processes.values():
                try:
                    if process.is_alive():
                        process.terminate()
                except (AttributeError, OSError):
                    pass  # Process may have already exited
    except Exception as e:
        logger.warning(f"Could not terminate workers directly: {e}")

---

[benchmarks/utils/evaluation.py, Lines 430-480] Logger State Mutation

def reset_logger_for_multiprocessing(log_dir: str, instance_id: str) -> None:
    # Get root logger and remove all existing handlers
    root_logger = logging.getLogger()
    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)

Problem: You're mutating global state (the root logger) from worker processes. This is dangerous - if anything else in the process tries to log, their handlers are gone.

Better approach: Create a new logger instance instead of hijacking the root logger:

def get_instance_logger(log_dir: str, instance_id: str) -> logging.Logger:
    """Get a logger configured for this specific instance."""
    logger = logging.getLogger(f"openhands.instance.{instance_id}")
    logger.handlers.clear()
    logger.propagate = False  # Don't propagate to root logger
    # ... rest of setup
    return logger

Why this matters: Mucking with the root logger is like mucking with global variables. It creates hidden dependencies and makes debugging harder.

---

[IMPROVEMENT OPPORTUNITIES] (Should fix - violates good taste)

[benchmarks/utils/evaluation.py, Lines 362-368] Redundant Log Files

log_file = os.path.join(log_dir, f"instance_{instance_id}.output.log")
# Redirect stdout/stderr to capture all output (SDK visualizations, etc.)
with redirect_stdout_stderr(log_file):

But then in reset_logger_for_multiprocessing:

log_file = os.path.join(log_dir, f"instance_{instance_id}.log")

Problem: You're creating TWO log files per instance:

• instance_{id}.log - for logger output
• instance_{id}.output.log - for stdout/stderr

Better: Use ONE log file for both. Redirect stdout/stderr to the same file handler the logger uses. This eliminates the special case and makes debugging simpler.

---

[benchmarks/utils/evaluation.py, Lines 263-296] Context Manager Pattern Violation

pool = ProcessPoolExecutor(max_workers=self.num_workers)
futures = []
try:
    # ... work ...
    pool.shutdown(wait=True)
except KeyboardInterrupt:
    self._cleanup_pool(pool, futures, wait=False)
    raise
except Exception:
    self._cleanup_pool(pool, futures, wait=False)
    raise

Problem: You abandoned the with statement for manual try/finally. This adds a special case (the normal completion path needs explicit pool.shutdown(wait=True)).

Good taste solution: Eliminate the special case:

pool = ProcessPoolExecutor(max_workers=self.num_workers)
try:
    futures = [pool.submit(...) for inst in instances_to_process]
    # ... process results ...
except (KeyboardInterrupt, Exception):
    self._cleanup_pool(pool, futures, wait=False)
    raise
finally:
    if not hasattr(self, '_cleanup_called'):
        pool.shutdown(wait=True)

Or better yet, create a custom context manager that handles cleanup properly:

@contextmanager
def interruptible_pool_executor(num_workers):
    pool = ProcessPoolExecutor(max_workers=num_workers)
    futures = []
    try:
        yield pool, futures
        pool.shutdown(wait=True)
    except (KeyboardInterrupt, Exception):
        cleanup_pool(pool, futures, wait=False)
        raise

---

[benchmarks/utils/evaluation.py, Lines 483-515] File Handle Leaks Possible

try:
    log_file = open(log_file_path, "a", buffering=1, encoding="utf-8")
    sys.stdout = log_file
    sys.stderr = log_file
    yield
finally:
    sys.stdout = original_stdout
    sys.stderr = original_stderr
    if log_file is not None and not log_file.closed:
        log_file.close()

Problem: If something still holds a reference to sys.stdout (like a thread or cached formatter), closing the file can cause issues. Also, if an exception happens during open(), you restore stdout/stderr before they were changed.

Better pattern:

log_file = None
try:
    log_file = open(log_file_path, "a", buffering=1, encoding="utf-8")
    sys.stdout = log_file
    sys.stderr = log_file
    try:
        yield
    finally:
        sys.stdout = original_stdout
        sys.stderr = original_stderr
finally:
    if log_file is not None:
        try:
            log_file.close()
        except Exception:
            pass  # Already closing, don't mask original exception

---

[STYLE NOTES] (Minor)

[benchmarks/swe_bench/eval_infer.py, Line 215] Inconsistent Naming

Changed default from "OpenHands" to "openhands". This is fine but make sure all downstream consumers expect lowercase. Case inconsistencies cause bugs.

---

VERDICT: ✅ Worth merging with documentation

Core logic is sound - You're solving real problems (zombie processes, output interleaving, debugging difficulty). The implementation works and has tests.

BUT: You're building on shaky foundation (accessing private attributes). At minimum, document the risks and add defensive code.

---

KEY INSIGHT:

You're fighting ProcessPoolExecutor's abstraction instead of working with it. When you need to hasattr(pool, "_processes"), that's the abstraction telling you "I don't want to give you this control."

The pragmatic solution is what you did (reach in and grab _processes), but the "good taste" solution would be to either:

1. Use multiprocessing.Pool which exposes .terminate() publicly
2. Wrap worker functions to handle signals properly and allow graceful shutdown
3. Accept that immediate termination isn't available and document the limitation

What Linus would say: "If you're reaching into private attributes, either the API is wrong or you're using the wrong API. In this case, it's both - ProcessPoolExecutor doesn't expose what you need, and you should probably use multiprocessing.Pool instead."

But he'd also say: "Ship it with a big comment explaining why you broke the rules. Real bugs trump theoretical purity." ✅

_{View full conversation}