|
| 1 | +# Copyright (c) 2025 Agentspan |
| 2 | +# Licensed under the MIT License. See LICENSE file in the project root for details. |
| 3 | + |
| 4 | +"""Worker isolation mode — ``process`` (default) or ``thread``. |
| 5 | +
|
| 6 | +``CONDUCTOR_WORKER_ISOLATION=thread`` replaces :class:`TaskHandler`'s |
| 7 | +multiprocessing primitives (``Process``, ``Queue``) with thread equivalents. |
| 8 | +For environments where multiprocessing's fork+exec bootstraps (spawn |
| 9 | +children, ``resource_tracker``) fail — e.g. minimal microVM guests. Threads |
| 10 | +share the parent's memory, so the log relay needs no IPC and worker payloads |
| 11 | +need no pickling. |
| 12 | +
|
| 13 | +The default (``process``, or the variable unset) leaves every existing code |
| 14 | +path untouched: ``multiprocessing.Process`` under the ``spawn`` start method. |
| 15 | +
|
| 16 | +Tradeoffs (thread mode only): no per-worker force-kill (``terminate``/``kill`` |
| 17 | +are no-ops; shutdown is cooperative), CPU-bound workers share the GIL, and |
| 18 | +``signal.signal`` becomes a no-op off the main thread instead of raising |
| 19 | +``ValueError``. |
| 20 | +""" |
| 21 | + |
| 22 | +from __future__ import annotations |
| 23 | + |
| 24 | +import logging |
| 25 | +import os |
| 26 | +import queue |
| 27 | +import threading |
| 28 | +from typing import Any |
| 29 | + |
| 30 | +logger = logging.getLogger(__name__) |
| 31 | + |
| 32 | +WORKER_ISOLATION_ENV = "CONDUCTOR_WORKER_ISOLATION" |
| 33 | +ISOLATION_PROCESS = "process" |
| 34 | +ISOLATION_THREAD = "thread" |
| 35 | +_VALID_MODES = frozenset({ISOLATION_PROCESS, ISOLATION_THREAD}) |
| 36 | + |
| 37 | +_invalid_mode_warned = False |
| 38 | + |
| 39 | + |
| 40 | +def isolation_mode() -> str: |
| 41 | + """Return the configured worker isolation mode. |
| 42 | +
|
| 43 | + Reads ``CONDUCTOR_WORKER_ISOLATION`` (case-insensitive, whitespace |
| 44 | + trimmed). Invalid values warn once and fall back to ``process`` — failing |
| 45 | + open to the default behavior rather than bricking every worker on a typo. |
| 46 | + """ |
| 47 | + global _invalid_mode_warned |
| 48 | + raw = os.environ.get(WORKER_ISOLATION_ENV, "") |
| 49 | + mode = raw.strip().lower() or ISOLATION_PROCESS |
| 50 | + if mode not in _VALID_MODES: |
| 51 | + if not _invalid_mode_warned: |
| 52 | + logger.warning( |
| 53 | + "Ignoring invalid %s=%r; valid values are %s; using %r", |
| 54 | + WORKER_ISOLATION_ENV, |
| 55 | + raw, |
| 56 | + sorted(_VALID_MODES), |
| 57 | + ISOLATION_PROCESS, |
| 58 | + ) |
| 59 | + _invalid_mode_warned = True |
| 60 | + mode = ISOLATION_PROCESS |
| 61 | + return mode |
| 62 | + |
| 63 | + |
| 64 | +class _ThreadAsProcess(threading.Thread): |
| 65 | + """threading.Thread shim that satisfies the multiprocessing.Process interface.""" |
| 66 | + |
| 67 | + def __init__( |
| 68 | + self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None |
| 69 | + ): |
| 70 | + super().__init__( |
| 71 | + group=group, target=target, name=name, args=args, kwargs=kwargs or {}, daemon=daemon |
| 72 | + ) |
| 73 | + self.exitcode: Any = None |
| 74 | + |
| 75 | + def terminate(self) -> None: |
| 76 | + pass |
| 77 | + |
| 78 | + def kill(self) -> None: |
| 79 | + pass |
| 80 | + |
| 81 | + @property |
| 82 | + def pid(self) -> None: |
| 83 | + return None |
| 84 | + |
| 85 | + |
| 86 | +def apply_thread_isolation() -> None: |
| 87 | + """Swap ``task_handler``'s multiprocessing primitives for thread equivalents. |
| 88 | +
|
| 89 | + Idempotent. Must run before the first :class:`TaskHandler` is constructed: |
| 90 | + its ``__init__`` creates the logging ``Queue`` and starts the logger |
| 91 | + ``Process`` immediately. |
| 92 | + """ |
| 93 | + from conductor.client.automator import task_handler as _th_module # lazy: no import cycle |
| 94 | + |
| 95 | + if getattr(_th_module, "_thread_isolation_applied", False): |
| 96 | + return |
| 97 | + |
| 98 | + _th_module.Process = _ThreadAsProcess # type: ignore[attr-defined] |
| 99 | + |
| 100 | + # With every worker a thread, the log relay is same-process: a plain |
| 101 | + # thread-safe queue replaces the IPC Queue, whose SemLock would drag in |
| 102 | + # the resource_tracker bootstrap this mode exists to avoid. |
| 103 | + _th_module.Queue = queue.Queue # type: ignore[attr-defined] |
| 104 | + |
| 105 | + # The worker/logger process targets call signal.signal(SIGINT, SIG_IGN) |
| 106 | + # at startup — valid in a child process but a ValueError in a non-main |
| 107 | + # thread. Make signal.signal skip silently off the main thread. |
| 108 | + import signal as _signal_mod |
| 109 | + |
| 110 | + _orig_signal_fn = _signal_mod.signal |
| 111 | + |
| 112 | + def _thread_safe_signal(signalnum, handler): |
| 113 | + if threading.current_thread() is threading.main_thread(): |
| 114 | + return _orig_signal_fn(signalnum, handler) |
| 115 | + # Non-main thread: skip silently |
| 116 | + |
| 117 | + _signal_mod.signal = _thread_safe_signal # type: ignore[assignment] |
| 118 | + |
| 119 | + _th_module._thread_isolation_applied = True # type: ignore[attr-defined] |
0 commit comments