[Bug]: Using MultiProcessShared handle can deadlock when always_proxy=True

What happened?

Beam's MultiProcessShared allows creating an object in one process and interacting with it from other processes through proxy objects. Beam uses Python's multiprocessing.managers to orchestrate this. By default, the object lives in the SDK harness process that created it first, which serves it from a background thread; spawn_process=True uses a dedicated process instead.

When the SDK harness process that hosts the object also accesses it through proxies (always_proxy=True, as RunInference does), a deadlock can happen if:

a proxy to the object is only reachable from a reference cycle, so only the cyclic GC can free it, and
the GC runs on one of the server's threads.

The GC then runs the proxy's finalizer, BaseProxy._decref, on that thread. The finalizer sends a decref request to the server over a new connection and waits for the reply. To serve it, the server's accepter thread has to accept the connection and start a request thread. Since GC ran on the accepter thread itself, the finalizer waited for a request thread that only the accepter could start, and the server never accepted another connection.

After the deadlock, the shared handle becomes unresponsive. Since _get_manager connects while holding a process-wide lock, acquiring other MultiProcessShared handles in the affected processes can hang too.

This was observed in an inference pipeline running on Dataflow (2026-10-16_17_11_27-1377094448160546253).

Process bundle threads both in the SDK holding the model and the sibling are stuck in File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 769, in answer_challenge:

========== ACTIVE PROCESSING BUNDLES ==========
--- instruction process_bundle-5846309219931912825-6 ---
ProcessBundleDescriptorId: s12-14
tracked thread: <_Worker(Thread-18, started daemon 137720324150976)>
time since transition: 599.94 seconds

--- Thread #137720324150976 name: Thread-18 ---
  File "/usr/local/lib/python3.11/threading.py", line 1002, in _bootstrap
    self._bootstrap_inner()
  File "/usr/local/lib/python3.11/threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/utils/thread_pool_executor.py", line 53, in run
    self._work_item.run()
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/utils/thread_pool_executor.py", line 37, in run
    self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 392, in task
    self._execute(
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 311, in _execute
    response = task()
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 393, in 
    lambda: self.create_worker().do_instruction(request), request)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 680, in do_instruction
    return getattr(self, request_type)(
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/sdk_worker.py", line 718, in process_bundle
    bundle_processor.process_bundle(
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1350, in process_bundle
    input_op_by_transform_id[element.transform_id].process_encoded(
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/runners/worker/bundle_processor.py", line 263, in process_encoded
    self.output(decoded_value)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/ml/inference/base.py", line 2162, in process
    self.update_model(si_model_metadata)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/ml/inference/base.py", line 2095, in update_model
    self._model = self._load_model(side_input_model_path=side_input_model_path)
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/ml/inference/base.py", line 2052, in _load_model
    load, tag=copy_tag, always_proxy=True).acquire())
  File "/opt/apache/beam-venv/beam-venv-worker-sdk-0-0/lib/python3.11/site-packages/apache_beam/utils/multi_process_shared.py", line 422, in acquire
    singleton = self._get_manager().acquire_singleton(self._tag)
  File "/usr/local/lib/python3.11/multiprocessing/managers.py", line 732, in temp
    conn = self._Client(token.address, authkey=self._authkey)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 525, in Client
    answer_challenge(c, authkey)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 769, in answer_challenge
    message = connection.recv_bytes(256)         # reject large message
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 430, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 395, in _recv
    chunk = read(handle, remaining)
...

and we can see the corresponding accepter thread also stuck in answer_challenge after calling _decref . Garbage collection is not evident since this doesn't include C frames, but the _decref call strongly suggests that GC started while accepter thread was active.

--- Thread #137719921497792 name: Thread-32 (accepter) ---
  File "/usr/local/lib/python3.11/threading.py", line 1002, in _bootstrap
    self._bootstrap_inner()
  File "/usr/local/lib/python3.11/threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.11/threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.11/multiprocessing/managers.py", line 192, in accepter
    t = threading.Thread(target=self.handle_request, args=(c,))
  File "/usr/local/lib/python3.11/threading.py", line 905, in __init__
    self._started = Event()
  File "/usr/local/lib/python3.11/threading.py", line 563, in __init__
    self._cond = Condition(Lock())
  File "/usr/local/lib/python3.11/threading.py", line 254, in __init__
    self._release_save = lock._release_save
  File "/usr/local/lib/python3.11/multiprocessing/util.py", line 227, in __call__
    res = self._callback(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.11/multiprocessing/managers.py", line 874, in _decref
    conn = _Client(token.address, authkey=authkey)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 525, in Client
    answer_challenge(c, authkey)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 769, in answer_challenge
    message = connection.recv_bytes(256)         # reject large message
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 430, in _recv_bytes
    buf = self._recv(4)
  File "/usr/local/lib/python3.11/multiprocessing/connection.py", line 395, in _recv
    chunk = read(handle, remaining)

There is a repro without DF.

import gc
import multiprocessing.connection
import threading
import uuid

from apache_beam.utils import multi_process_shared


class Counter:
  def __init__(self):
    self._n = 0

  def incr(self):
    self._n += 1
    return self._n


_original_accept = multiprocessing.connection.Listener.accept


def _accept_then_gc(self):
  conn = _original_accept(self)
  if threading.current_thread().name.endswith('(accepter)'):
    gc.collect()  # Stand-in for a GC pass triggered by an allocation here.
  return conn


def acquire_and_call(tag, result):
  shared = multi_process_shared.MultiProcessShared(
      Counter, tag=tag, always_proxy=True)
  result.append(shared.acquire().incr())


def main():
  multiprocessing.connection.Listener.accept = _accept_then_gc
  tag = 'mps_gc_deadlock_' + uuid.uuid4().hex

  result = []
  acquire_and_call(tag, result)  # Starts the in-process server.
  print('before: acquire()+incr() ->', result)

  # A proxy to the server becomes cyclic garbage, so only the cyclic GC can
  # reclaim it, on whatever thread the GC happens to run.
  gc.disable()
  victim = multi_process_shared.MultiProcessShared(
      Counter, tag=tag, always_proxy=True).acquire()
  cycle = {'proxy': victim}
  cycle['self'] = cycle
  del victim, cycle

  result = []
  t = threading.Thread(target=acquire_and_call, args=(tag, result), daemon=True)
  t.start()
  t.join(timeout=15)
  if t.is_alive():
    print('after: acquire() still blocked after 15s -> accepter deadlocked')
  else:
    print('after: acquire()+incr() ->', result)


if __name__ == '__main__':
  main()

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

repro of a similar deadlock without Beam:

import faulthandler
import gc
import sys
import threading
from multiprocessing.managers import BaseManager


class Counter:
    def __init__(self):
        self.n = 0

    def incr(self):
        self.n += 1
        return self.n


class MyManager(BaseManager):
    pass


MyManager.register('Counter', Counter)


if __name__ == '__main__':
    # Serve the manager from a thread of this process.
    server = MyManager(address=('127.0.0.1', 0), authkey=b'secret').get_server()
    threading.Thread(target=server.serve_forever, daemon=True).start()

    manager = MyManager(address=server.address, authkey=b'secret')
    manager.connect()

    # Run the cyclic GC often, on whichever thread happens to allocate.
    gc.set_threshold(10)

    def work():
        for _ in range(1000):
            proxy = manager.Counter()
            proxy.incr()
            cycle = [proxy]
            cycle.append(cycle)  # Only the cyclic GC can free the proxy.

    t = threading.Thread(target=work, daemon=True)
    t.start()
    t.join(timeout=30)
    if t.is_alive():
        print('DEADLOCKED')
        if '--stacks' in sys.argv:
            faulthandler.dump_traceback(all_threads=True)
    else:
        print('ok')

I think reasons contributing to this deadlock are

hosting the model in the same process that operates it via proxies.
multiprocessing internals, specifically the fact that finalizers on proxy objects require a blocking call to the server, and they run on whichever thread the GC happens to run on, which can be one of the server's own threads. I am not sure if multiprocessing necessarily forbids using proxies in the same process, so this might warrant a CPython issue.

Ideas to remediate this (with each of the below, the issue is not reproducible):
A) use always_proxy=False.
B) use spawn_process=True
C) monkey-patch multiprocessing internals to make finalizers not blocking

A: We added always_proxy=True in #27989 with the motivation to reduce side-effects of an SDK process crash on other harnesses. The intent of that PR is not fully accomplished through always_proxy since we still don't launch a separate process with always_proxy, more recently we added spawn_process=True for that. In the latter case, the separate process is still attached to a parent process via a lifeline - IIUC if a process that created a shared MPS handle crashes, the process owning the handle self-destructs:

beam/sdks/python/apache_beam/utils/multi_process_shared.py

Line 238 in f014f5a

Includes a 'Suicide Pact' monitor: If parent dies, I die.

.

B: This is a more recent addition. We use it in Model Manager path, and there were several other issues we recently fixed with model manager, so that path is not as tested as the main, but still spawning separate processes might be working well, I haven't tested it beyond confirming that it solves the deadlock problem in the simple repro above. Likely, this is not a one-line change and might impact other things, like metrics. It might be better if multiprocessing makes some assumptions that each shared handle gets a server in a dedicated process. There are some signs that it does make such assumptions.

C: I'd rather not go this route. It looks like we are already messing with internals:

beam/sdks/python/apache_beam/utils/multi_process_shared.py

Line 57 in 7bdc6ff

manager_owned=True):

and that patch might increase the likelihood of such deadlocks, but doesn't seem to be the root cause for this particular instance I encountered. The LLM-generated patch doesn't look pretty, and I'd rather revert the existing patches of this kind if we no longer need them on currently supported Python versions.

A) could be the easiest remediation for now if it doesn't make things worse. Though a deadlock might still be possible in a scenario when two SDK processes each have proxies for objects owned by the opposite process, and then both proxies are garbage-collected at the exact same instant.

@damccorm

cc: @damccorm @AMOOOMA fyi.

This issue was discovered when testing a fix for another issue, and the pipeline used a fake MH where inferences were fast, so the worker spent a lot of time in inter-process communication. In real pipelines the chances that GC starts while one of multiprocessing threads is active might be less probable.

+1 to option (b), I think this is the desired behavior anyways (as you pointed, out that was the intent of my PR when I originally made it).

Thanks for looking into this

@tvalentyn

Hi @tvalentyn,

I’m interested in contributing to this issue. Based on the investigation and the discussion around spawn_process=True, I’d like to help evaluate the proposed approach and its implications for MultiProcessShared.

I can focus on the following areas:

Reviewing the implementation in apache_beam/utils/multi_process_shared.py.
Investigating the behavior of spawn_process=True, particularly its impact on model inference, lifecycle management, and metrics.
Developing regression tests to verify that the reported deadlock is prevented.
Checking for potential compatibility issues or unintended side effects.

Would you recommend starting with the regression tests or investigating the implementation and its interactions with the model manager first?

Thank you.

@damccorm

@damccorm is already looking into this issue. if any help is needed, @damccorm can get back to you.

Navigation Menu

Repository navigation

[Bug]: Using MultiProcessShared handle can deadlock when always_proxy=True #40673

What happened?

Issue Priority

Issue Components

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

[Bug]: Using MultiProcessShared handle can deadlock when always_proxy=True #40673

Description

What happened?

Issue Priority

Issue Components

Activity

tvalentyn commented on Oct 20, 2026

tvalentyn commented on Oct 20, 2026

tvalentyn commented on Oct 20, 2026

tvalentyn commented on Oct 20, 2026

damccorm commented on Oct 20, 2026

alisha-1000 commented on Oct 22, 2026

tvalentyn commented on Oct 22, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions