Fix concurrent StreamablePipe writes

[nalves599]

Summary

• serialize StreamablePipe.write() calls with a per-pipe lock
• recreate the lock after unpickling to support spawned worker processes
• add regression coverage for concurrent writes and pickling

Connection.send_bytes() splits large frames into separate header and payload writes. Without serialization, concurrent writers sharing one StreamablePipe can interleave those writes and corrupt the parent log watcher stream.

Fixes #888.

Validation

• focused StreamablePipe regression tests
• flake8 dramatiq/compat.py tests/test_compat.py
• five-writer, 128 KiB frame reproduction: all 500 frames are received with this patch