Merge pull request #414 from conductor-oss/fix/spawn-default-and-worker-pickling

Make @worker_task workers spawn-safe; default to spawn on all platforms

Author: bradyyie

CHANGELOG.md

@@ -15,5 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Multiprocessing start method now defaults to `spawn` on all platforms** (was `fork` everywhere except Windows). `fork` caused silently-restarting `SIGSEGV` worker subprocesses on macOS (exitcode `-11`) and fork-with-held-lock deadlocks on POSIX. Opt back in with `CONDUCTOR_MP_START_METHOD=fork`. Entrypoint scripts must use the standard `if __name__ == "__main__":` guard, and workers must be defined at module level
 - Legacy metrics emit unchanged by default; no env var required
 - `metrics_collector.py` is now a compatibility shim; `from conductor.client.telemetry.metrics_collector import MetricsCollector` continues to work
+
+### Fixed
+
+- `@worker_task` workers are now picklable, making the decorator path work with the `spawn`/`forkserver` start methods (fixes `TypeError: cannot pickle '_thread.lock' object` and `PicklingError: it's not the same object as ...`; issues #264, #271): `Worker.api_client` is created lazily in the worker process, runtime state (locks, pending async tasks, background loop) is excluded from pickling and rebuilt in the child, and decorated functions are pickled as importable references resolved in the child
+- `TaskHandler.start_processes()` no longer hangs the interpreter when a worker fails to start (e.g., unpicklable state under `spawn`); it now cleans up already-started subprocesses and raises with actionable guidance
+- Worker processes killed by a signal now log a diagnostic hint (signal number, `PYTHONFAULTHANDLER=1` guidance) instead of restarting silently

docs/WORKER.md

@@ -229,11 +229,6 @@ from conductor.client.configuration.configuration import Configuration
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.worker.worker import Worker
 
-#### Add these lines if running on a mac####
-from multiprocessing import set_start_method
-set_start_method('fork')
-############################################
-
 SERVER_API_URL = 'http://localhost:8080/api'
 KEY_ID = '<KEY_ID>'
 KEY_SECRET = '<KEY_SECRET>'
@@ -261,10 +256,39 @@ workers = [
 
 # TaskHandler scans for @worker_task decorated workers by default.
 # Set scan_for_annotated_workers=False if you want to disable auto-discovery.
-with TaskHandler(workers, configuration, scan_for_annotated_workers=True) as task_handler:
-    task_handler.start_processes()
+# The __main__ guard is required: worker processes use the 'spawn'
+# multiprocessing start method by default, which re-imports this script.
+if __name__ == '__main__':
+    with TaskHandler(workers, configuration, scan_for_annotated_workers=True) as task_handler:
+        task_handler.start_processes()
 ```
 
+### Multiprocessing start method
+
+`TaskHandler` runs each worker in its own process using Python's `multiprocessing`
+with the **`spawn`** start method by default on all platforms. `spawn` starts each
+worker as a fresh interpreter, which avoids the crashes and deadlocks `fork`
+causes in processes holding native state — most visibly on macOS, where forked
+workers can die with `SIGSEGV` (exitcode `-11`) inside Apple system frameworks
+and get silently restarted by the supervisor.
+
+Requirements under `spawn` (standard Python multiprocessing rules):
+
+- Guard your entrypoint with `if __name__ == '__main__':`
+- Define workers at module level (not nested inside functions or lambdas)
+- `configuration`, `metrics_settings`, and `event_listeners` passed to
+  `TaskHandler` must be picklable
+
+To opt back into the previous behavior (e.g., a Linux deployment relying on
+fork's copy-on-write memory sharing):
+
+```shell
+export CONDUCTOR_MP_START_METHOD=fork   # spawn (default) | fork | forkserver
+```
+
+If a worker process dies from a signal repeatedly at startup, set
+`PYTHONFAULTHANDLER=1` to capture the crashing stack trace from the child.
+
 ### Resilience: auto-restart and health checks
 
 If you run workers as a long-lived service (e.g., alongside FastAPI/Uvicorn), you can optionally enable process

examples/spawn_safety_probe.py

@@ -0,0 +1,125 @@
+"""
+spawn_safety_probe.py - @worker_task spawn-safety probe (GitHub issues #264/#271).
+
+Run from the repo root:
+    PYTHONPATH=src python3 examples/spawn_safety_probe.py
+
+Forces the 'spawn' multiprocessing start method (the SDK default; also the
+only safe method on macOS) and then checks, in order:
+
+  1. Which Worker attributes survive pickling (attribute-level diagnostic)
+  2. Whole-Worker pickle round-trip + task execution on the restored worker
+  3. Worker transferred into a REAL 'spawn' child process and executed there
+     (exactly what TaskHandler.start_processes() does)
+
+Broken SDK output:  FAIL lines for api_client / _pending_tasks_lock /
+                    _execute_function; round-trip and spawn-child checks fail.
+Fixed SDK output:   everything PASS.
+"""
+import os
+import sys
+
+# Force spawn BEFORE importing any conductor module: task_handler pins the
+# start method at import time from this env var.
+os.environ["CONDUCTOR_MP_START_METHOD"] = "spawn"
+
+import multiprocessing
+import pickle
+
+from conductor.client.worker.worker_task import worker_task
+from conductor.client.http.models.task import Task
+
+
+@worker_task(task_definition_name="pickle_probe_task")
+def pickle_probe_task(x: int) -> dict:
+    return {"doubled": x * 2}
+
+
+def _make_task() -> Task:
+    return Task(
+        task_id="probe-task-id",
+        workflow_instance_id="probe-wf-id",
+        task_def_name="pickle_probe_task",
+        input_data={"x": 21},
+        status="IN_PROGRESS",
+    )
+
+
+def run_in_child(worker, result_queue) -> None:
+    """Spawn-child entry point: unpickling `worker` here IS the test."""
+    result = worker.execute(_make_task())
+    result_queue.put((str(result.status), result.output_data))
+
+
+def get_probe_worker():
+    from conductor.client.automator.task_handler import get_registered_workers
+    return next(w for w in get_registered_workers()
+                if w.task_definition_name == "pickle_probe_task")
+
+
+def main() -> int:
+    failures = 0
+    print(f"python={sys.version.split()[0]}  platform={sys.platform}")
+
+    worker = get_probe_worker()  # importing task_handler pins the start method
+    print(f"multiprocessing start method: {multiprocessing.get_start_method(allow_none=True)}\n")
+
+    # -- 1. attribute-level diagnostic --------------------------------------
+    # Pickle what pickling actually serializes: the __getstate__ output.
+    # (Raw __dict__ values legitimately contain process-local objects - a
+    # threading.Lock, the original decorated function - which __getstate__
+    # excludes or substitutes. If a future change adds unpicklable state and
+    # forgets to handle it in __getstate__, this section catches it.)
+    print("[1] Worker pickled-state (__getstate__) picklability:")
+    state = worker.__getstate__() if hasattr(worker, "__getstate__") else dict(worker.__dict__)
+    for k, v in state.items():
+        try:
+            pickle.dumps(v)
+            print(f"    PASS  {k}")
+        except Exception as e:
+            failures += 1
+            print(f"    FAIL  {k}: {type(e).__name__}: {str(e)[:70]}")
+
+    # -- 2. whole-worker round-trip + execute --------------------------------
+    print("\n[2] Whole-Worker pickle round-trip:")
+    try:
+        restored = pickle.loads(pickle.dumps(worker))
+        same_fn = restored.execute_function is worker.execute_function
+        result = restored.execute(_make_task())
+        if str(result.status) == "COMPLETED" and result.output_data == {"doubled": 42}:
+            print(f"    PASS  round-trip (same function object: {same_fn}, "
+                  f"result: {result.output_data})")
+        else:
+            failures += 1
+            print(f"    FAIL  unexpected result: {result.status} {result.output_data}")
+    except Exception as e:
+        failures += 1
+        print(f"    FAIL  {type(e).__name__}: {str(e)[:100]}")
+
+    # -- 3. real spawn child --------------------------------------------------
+    print("\n[3] Execute in a real 'spawn' child process:")
+    ctx = multiprocessing.get_context("spawn")
+    q = ctx.Queue()
+    p = ctx.Process(target=run_in_child, args=(get_probe_worker(), q))
+    try:
+        p.start()
+        status, output = q.get(timeout=60)
+        p.join(timeout=10)
+        if status == "COMPLETED" and output == {"doubled": 42}:
+            print(f"    PASS  child exitcode={p.exitcode}, result={output}")
+        else:
+            failures += 1
+            print(f"    FAIL  status={status} output={output}")
+    except Exception as e:
+        failures += 1
+        print(f"    FAIL  {type(e).__name__}: {str(e)[:100]}")
+        if p.is_alive():
+            p.terminate()
+            p.join(timeout=5)
+
+    print(f"\n{'ALL CHECKS PASSED' if failures == 0 else f'{failures} CHECK(S) FAILED'}")
+    return 0 if failures else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/conductor/client/automator/task_handler.py

@@ -4,10 +4,11 @@
 import inspect
 import logging
 import os
+import pickle
 import signal
 import threading
 import time
-from multiprocessing import Process, freeze_support, Queue, set_start_method
+from multiprocessing import Process, freeze_support, Queue, set_start_method, get_start_method
 from sys import platform
 from typing import List, Optional, Any, Dict
 
@@ -37,12 +38,25 @@
 _mp_fork_set = False
 if not _mp_fork_set:
     try:
-        # The prometheus_client library holds a module-level threading lock;
-        # forking while that lock is held causes a deadlock in child processes.
-        # Set CONDUCTOR_MP_START_METHOD=spawn to avoid this if you hit the
-        # deadlock.  Default is fork for backward compatibility (spawn requires
-        # all Process arguments to be picklable).
-        _default_method = "spawn" if platform == "win32" else "fork"
+        # Default start method: "spawn" on every platform.
+        #
+        # fork() is fundamentally unsafe in a process that holds native
+        # framework state or non-main threads:
+        #   - macOS: Apple frameworks (CFNetwork, Security, libdispatch) may
+        #     SIGSEGV in forked children, producing silently-restarting worker
+        #     processes with exitcode=-11. CPython switched its own macOS
+        #     default to spawn in 3.8 for the same reason (bpo-33725).
+        #   - all POSIX: forking while another thread holds a lock (e.g. the
+        #     prometheus_client module-level lock, or the TaskHandler monitor
+        #     thread restarting a worker) leaves that lock permanently held in
+        #     the child -> silent deadlock.
+        # Workers (including the @worker_task decorator path) are pickle-safe,
+        # so spawn works everywhere. Deployments that rely on fork's
+        # copy-on-write inheritance can opt back in with
+        # CONDUCTOR_MP_START_METHOD=fork (re-exposing the hazards above).
+        # NOTE: spawn requires the standard `if __name__ == "__main__":` guard
+        # in the entrypoint script.
+        _default_method = "spawn"
         _method = os.environ.get("CONDUCTOR_MP_START_METHOD", "").strip().lower() or _default_method
         if _method not in _VALID_MP_START_METHODS:
             logger.warning(
@@ -341,8 +355,26 @@ def start_processes(self) -> None:
         logger.info("Starting worker processes...")
         freeze_support()
         self._monitor_stop_event.clear()
-        self.__start_task_runner_processes()
-        self.__start_metrics_provider_process()
+        try:
+            self.__start_task_runner_processes()
+            self.__start_metrics_provider_process()
+        except (TypeError, pickle.PicklingError, AttributeError) as e:
+            # Under the 'spawn'/'forkserver' start methods, Process arguments
+            # are pickled at start(); unpicklable state fails here. Clean up
+            # everything already started - otherwise the non-daemon logger
+            # subprocess keeps the interpreter alive forever after the
+            # exception (observed as a hang after "cannot pickle
+            # '_thread.lock' object").
+            logger.error(
+                "Failed to start worker processes with start method %r: %s. "
+                "Workers must be defined at module level (not nested inside "
+                "functions), and configuration, metrics_settings and "
+                "event_listeners must be picklable. "
+                "See https://github.com/conductor-oss/conductor-python/issues/264",
+                get_start_method(allow_none=True), e,
+            )
+            self.stop_processes()
+            raise
         self.__start_monitor_thread()
         logger.info("Started all processes")
 
@@ -445,6 +477,21 @@ def __check_and_restart_processes(self) -> None:
                 worker = self.workers[i] if i < len(self.workers) else None
                 worker_name = worker.get_task_definition_name() if worker is not None else f"worker[{i}]"
                 logger.warning("Worker process exited (worker=%s, pid=%s, exitcode=%s)", worker_name, process.pid, exitcode)
+                if exitcode < 0:
+                    # Negative exitcode == killed by signal (e.g. -11 is
+                    # SIGSEGV). Under the 'fork' start method this is the
+                    # classic symptom of fork-unsafe native libraries in the
+                    # parent (Apple frameworks on macOS, numpy/Accelerate,
+                    # grpc, ...).
+                    logger.warning(
+                        "Worker process %s was killed by signal %s. If this "
+                        "repeats on every restart, the process is likely "
+                        "crashing at startup: use the 'spawn' start method "
+                        "(CONDUCTOR_MP_START_METHOD=spawn, the default) and "
+                        "set PYTHONFAULTHANDLER=1 to capture the crashing "
+                        "stack trace.",
+                        worker_name, -exitcode,
+                    )
                 if not self.restart_on_failure:
                     continue
                 self.__restart_worker_process(i)