apache / apache/pulsar-client-python
Client is highly fork-unsafe: any usage of multiprocessing deadlocks if an at-fork hook triggers garbage collection of a Pulsar client
- Dominant language
- Python
- Stars
- 75
- Forks
- 53
- PR merge metrics
- No merged PRs in 30d
Description
## Context:
- Pulsar 2.10.2 (StreamNative)
- Pulsar client Python 2.10.1 and 3.1.0 both repro'd.
- Python 3.10 and 3.7
- Linux (Ubuntu 20) and MacOS.
## Bug description:
It's common practice to clear out global connection cached objects after a `fork` takes place. That's typically done via the `os.register_at_fork` method in Python.
Callbacks to that method are similar to signal handlers: they shouldn't do anything complicated or IO-ful. However something very common to do in an atfork callback is to wipe out a global connection cache, setting it to `None` in the child process to indicate that a new connection should be opened if needed.
We do that with our Pulsar client connections. Since the Pulsar client isn't fork safe at all, we wipe out our global references to it whenever we fork via `register_at_fork` handlers (and our code runs in tons of our internal services that make _heavy_ use of Python forking; this is an extremely common practice in large Python installs: gunicorn, Celery, multiprocessing, billiard, airflow ... all heavily use fork-based concurrency to encapsulate workloads, share expensive-to-load-and-huge pre-forked memory, and evade the GIL).
The most common forking system is `multiprocessing`, which provides a "run some work concurrency then wrap up" interface.
We observed that arbitrary `multiprocessing` workloads across our fleet occasionally hang. Workloads that very frequently invoke `multiprocessing` pools after publishing to Pulsar were more likely to hang.
After some painful production incidents, we arrived at the below reproduction, which indicates an internal Pulsar client deadlock that occurs when a global connection cache is cleared out, triggering destruction inside the Pulsar-client C++ code.
**Note that this occurs on *any* invocation of `multiprocessing` in the presence of a Pulsar connection that is cleared by an atfork hook, regardless of whether the multiprocessing code, or the parent process, will ever use Pulsar again**. This is highly unfortunate.
## Steps to Reproduce
1. Run a standalone pulsar broker on localhost:6650.
2. Put the below Python code in a `repro.py` file. Optionally, edit the test topic name to use a topic/namespace/tenant that exist for you locally.
3. Do `while true; do python repro.py; done` to repeatedly invoke it.
4. Wait ten minutes.
Snippet:
```python
import time
from multiprocessing import Pool, set_start_method
import pulsar
import os
CLIENT = None
PROD = None
def do_send():
global CLIENT
global PROD
if PROD is None:
CLIENT = pulsar.Client(service_url='pulsar://localhost:6650')
PROD = CLIENT.create_producer(topic='persistent://chariot1/chariot_ns_sre--kms_test/chariot_topic_kms_test')
PROD.send(b"123")
def clear_connection_cache():
global CLIENT
global PROD
print("Destroying connections", flush=True)
CLIENT = PROD = None
print("Destroyed connections", flush=True)
def main():
os.register_at_fork(after_in_child=clear_connection_cache)
do_send()
processes = 20
set_start_method('fork')
pool = Pool(
processes=processes,
)
try:
pool.map_async(
time.sleep, [0.1 for _ in range(processes)]
).get(1)
finally:
pool.close()
print("Joining pool", flush=True)
pool.join()
print("Joined pool", flush=True)
if __name__ == '__main__':
main()
```
## Expected behavior
The code should continue successfully being invoked over and over after ten minutes.
More to the point **all invocations of Pulsar client functionality, including destruction, should be "safe" to perform in all contexts: forks, threads, asyncio coroutines, trio coroutines, multiprocessing, Billiard, `pthread_atfork` hooks, signal handlers, etc**. "Safe" might mean "works" or it might mean "errors immediately saying 'this is not allowed'". But hanging forever is not safe.
## Observed behavior
After some number of iterations, usually <1min, the code hangs in a pool-join timeout.
When that happens, the number of `Destroying connections` log lines and `Destroyed connections` log lines will not match up, indicating that one call to `CLIENT = PROD = None` is blocking.
## Root cause
When the test code hangs, attaching a `py-spy` stack printer (with `--threads --native`) to the hung subprocess indicates that it is stuck in what appears to be code related to the client's destructor. The code in `_after_fork_child` in the below stack is exactly analogous to the `clear_connection_cache` code in the above snippet:
Contributor guide
Assessment
This issue has not been assessed yet.