StreamablePipe corrupts concurrent large writes and can stall the log watcher
- Dominant language
- Python
- Stars
- 5.3k
- Forks
- 383
- Avg merge
- 9h 41m
- Merged PRs (30d)
- 2
Description
## Summary
`StreamablePipe.write()` calls `multiprocessing.connection.Connection.send_bytes()` without serializing access. A Dramatiq child redirects both `sys.stdout` and `sys.stderr` to one `StreamablePipe`, so independent concurrent writers can interleave pipe frames. This corrupts the parent log watcher stream.
## Impact
For payloads larger than 16 KiB, CPython emits the frame header and payload in separate writes. Interleaving can make `watch_logs()` decode a mixed payload or treat arbitrary bytes as a payload length. In the latter case, `recv_bytes()` waits for that false length and the parent stops forwarding worker logs.
## Affected versions
Reproduced with:
- Dramatiq 1.17.1
- Dramatiq 2.2.0 (current latest release)
## Reproduction
Use five concurrent writers sharing one `StreamablePipe`, each writing 128 KiB JSON records. With 500 records per writer, the unmodified 2.2.0 implementation consistently reports frame corruption, for example:
```text
frame corruption after 0 of 2500 frames: UnicodeDecodeError
```
Serializing the same writes with a process-local lock delivers all frames:
```text
received all 2500 frames
```
The relevant current implementation is:
```python
def write(self, s):
self.pipe.send_bytes(s.encode(self.encoding, errors="replace"))
```
## Proposed fix
Add a process-local lock to `StreamablePipe` and hold it around the complete `send_bytes()` call. The lock must be recreated on unpickle so spawned worker processes remain supported.
```python
with self._write_lock:
self.pipe.send_bytes(s.encode(self.encoding, errors="replace"))
```
A regression test can use two threads and a fake `send_bytes()` implementation that detects overlapping calls, plus a pickle test for `StreamablePipe`.
I have a local patch and focused regression test available if a PR would be useful.
Contributor guide
Research direction
Look at the StreamablePipe class in the codebase, likely in a file related to logging or multiprocessing. Understand how write() currently calls send_bytes. The fix involves adding a process-local lock around the send_bytes call and ensuring it works after unpickling. Write a regression test using threads to simulate concurrent writes and verify no corruption occurs.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 65/100