tvalentyn opened a new issue, #40673:
URL: https://github.com/apache/beam/issues/40673
### What happened?
Beam's MulitProcessShared and allows creating an object in a dedicated
process, and interact with the object in other processes through proxy objects.
Beam uses Python multiprocessing package to orchestrate this.
It appears that when an SDK harness process interacts with the shared handle
owned by the same process, a deadlock can happen if
- there is a cyclic reference that includes a proxy object.
- GC happens while a thread running a portion of `multiprocessing` inner
machinery is active.
- The nature of the deadlock is: a lifecycle method on the proxy is
initiating a synchronous action that has to be served by a thread responsible
for some portion of `multiprocessing` machinery, but that thread won't schedule
until GC is finished, hence deadlock.
Still figuring the details out, but 1) I encountered this on an inference
pipeline running on Dataflow and 2) there is a repro.
```
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
- [x] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
