Bogdanp / Bogdanp/dramatiq

StreamablePipe corrupts concurrent large writes and can stall the log watcher

Đang mở
#888 1 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
Python
Star
5.3k
Fork
383
Merge trung bình
9 giờ 41 phút
Pull request đã merge (30 ngày)
2

Mô tả

## 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.

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
python
Lĩnh vực
backend, distributed-systems
Loại issue
Lỗi
Độ khó
3/5
Thời gian dự kiến
1-2 ngày
Mức độ hoạt động
Sôi nổi
Độ rõ ràng
Đặc tả rõ ràng
Mức phù hợp với người mới
65/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.