This document is written for an AI assistant picking up this work in a new session. Read every section before doing anything. The user (Luke Bakken, @lukebakken) will confirm when to start.
Before any code or commit work, read these two files. They define mandatory workflow rules that Luke enforces strictly:
/home/lrbakken/genai/AGENTS.md- commit workflow (one 👍 per commit, never commit without approval, no em-dashes, etc.)/home/lrbakken/genai/git/GIT.md- commit message formatting, PR body conventions, markdown rules, use--body-filefor gh commands with backticks
The single most important rule: never commit anything without an explicit 👍 from Luke. Present the change, announce it is ready, wait for the thumbs-up, then commit. One logical fix = one commit = one 👍.
Working directory: /home/lrbakken/development/pika/pika
Branch: feature/thread-safe-connection
Remote tracking: gh-1582/feature/thread-safe-connection
PR: pika/pika#1582 (currently DRAFT, milestone 1.5.0)
Original PR author: Dheeraj4103. Luke is the assignee and has been doing all the review and fixes in this session.
pika is in the top 5% of all PyPI downloads. Correctness matters enormously.
Run tests:
# Unit tests only (no broker needed)
python -m pytest tests/unit/thread_safe_connection_tests.py -q
# Acceptance tests (need RabbitMQ - see section 3)
python -m pytest tests/acceptance/thread_safe_connection_test.py -q
# Type checking
hatch run typecheck
The acceptance tests require a broker on 127.0.0.1:5672 with guest/guest credentials. Start it with:
docker run --pull always --rm --publish 5672:5672 --publish 15672:15672 rabbitmq:4-management-alpine
The broker may not be running when you start the session. The acceptance test failures
will say ConnectionRefusedError if it is down - that is not a code bug.
pika/adapters/thread_safe_connection.py - the main implementation
tests/unit/thread_safe_connection_tests.py - 51 unit tests (no broker)
tests/acceptance/thread_safe_connection_test.py - 7 integration tests (broker required)
examples/basic_publisher_threaded.py - usage example
ThreadSafeConnection wraps SelectConnection and runs its IOLoop in a single
dedicated background thread (the IOLoop thread). All channel write operations are
routed through connection.ioloop.add_callback_threadsafe so that _tx_buffers is
only ever touched from the IOLoop thread. This eliminates the
IndexError: pop from an empty deque race that occurred when multiple threads called
basic_publish on a shared raw connection (issues #1144 and #511).
ThreadSafeChannel wraps pika.channel.Channel:
- Fire-and-forget methods (
basic_publish,basic_ack,basic_nack,basic_reject): schedule viaadd_callback_threadsafe, return immediately, safe from any thread simultaneously. - Blocking methods (
queue_declare,basic_qos,basic_consume,basic_cancel,close): block the calling thread on athreading.Eventuntil the broker responds or the connection closes.
ThreadSafeConnection public API:
channel()- blocks until channel is open, returnsThreadSafeChannelclose(timeout=10)- blocks until IOLoop thread exits (or force-stops after timeout)add_callback_threadsafe(callback)- schedule arbitrary work on the IOLoop threadis_open,is_closedproperties- Context manager support (
with ThreadSafeConnection(...) as conn:)
The _blocking_waiters mechanism: every blocking call appends a
(threading.Event, list[BaseException | None]) tuple to _blocking_waiters while
holding _channel_waiters_lock. When the connection closes (_on_connection_closed),
when the IOLoop crashes (_run_ioloop except handler), or when the IOLoop is
force-stopped (close() timeout path), all waiters are woken by setting their events
with the error reason. This is how blocked threads are guaranteed to unblock even when
things go wrong.
add_callback_threadsafe lives on connection.ioloop, not on SelectConnection.
This was the critical production bug found by integration tests - all unit tests missed
it because MagicMock auto-invents any attribute access. Every call site in the
wrapper is self._connection.ioloop.add_callback_threadsafe(...).
_on_chan_close callbacks: each blocking channel method registers an
add_on_close_callback inside the scheduled closure so that a channel-level broker
error (e.g. 404 on passive declare) wakes the waiter. Pika provides no API to
deregister individual close callbacks, so they accumulate for the lifetime of the
channel. Each is guarded with if not ready.is_set() to make stale callbacks
(from already-completed calls) unconditional no-ops.
on_message_callback wrapping: basic_consume wraps the user's callback so
that the channel argument delivered to it is always the ThreadSafeChannel (self),
never the raw pika.channel.Channel. This gives users a consistent API throughout.
All of these are committed to the branch. They are listed in the order fixed.
Root cause: Every call site used self._connection.add_callback_threadsafe(...).
The method lives on self._connection.ioloop, not on SelectConnection. The wrapper
was entirely non-functional in production. MagicMock silently invented the attribute
in unit tests, masking the bug completely.
Fix: replaced all occurrences with self._connection.ioloop.add_callback_threadsafe(...).
Updated unit tests to assert on mock_conn.ioloop.add_callback_threadsafe.
Commit: 67feebc
Root cause: _run_ioloop's except handler woke _blocking_waiters but never set
_connected_event. If the IOLoop crashed before _on_connection_open fired (e.g.
during TCP setup), the constructor blocked on _connected_event.wait() indefinitely.
Fix: added to the except block:
if not self._connected_event.is_set():
self._connect_error = exc
self._connected_event.set()Commit: 283a481
Root cause: queue_declare, basic_qos, basic_consume, basic_cancel blocked
on a threading.Event only woken by the success callback or a full connection close.
A broker-side Channel.Close (404 on passive declare, parameter conflict, etc.) causes
pika to drop all pending callbacks for that channel without firing them.
_on_connection_closed is NOT called - only the channel closes. The calling thread
waited forever.
Fix: registered add_on_close_callback(_on_chan_close) inside each scheduled
closure. The close callback sets error[0] and calls ready.set().
Commit: a081f22
Root cause: _open (in channel()), _declare, _qos, _consume, _cancel,
_close all called into the connection or channel without error handling. A wrong-state
exception (e.g. channel() racing with close()) propagated out of ioloop.start()
and killed the IOLoop instead of surfacing the error to the caller.
Fix: wrapped each scheduled closure body in try/except that sets error[0] and
calls ready.set().
Commit: a081f22 (same commit as Bug 2)
Root cause: the IOLoop-thread fast path called self._connection.close() with no
error handling. If the connection was already closing, ConnectionWrongStateError
propagated from the callback and crashed the IOLoop.
Fix: wrapped in try/except matching the _safe_close pattern.
Commit: a081f22 (same commit as Bugs 2 and 3)
Root cause: when close(timeout=N) expired and force-stopped the IOLoop via
ioloop.stop(), _on_connection_closed was never called. _closed_reason stayed
None; threads in _blocking_waiters at that moment hung forever; subsequent
channel() calls also hung because the _closed_reason is None guard passed.
Fix: after the second join() in the force-stop path, if _closed_reason is None,
set it to a forced-close exception and wake/clear all remaining _blocking_waiters.
Commit: aee948f
Root cause: each completed blocking call left a registered _on_chan_close
closure that would still execute work (setting error[0], calling ready.set()) when
the channel eventually closed, even though the call had long since returned.
Fix: guarded each _on_chan_close with if not ready.is_set().
Commit: 8e35435
Root cause: the warning about heartbeat starvation and deadlock from calling
blocking methods was buried in basic_consume's docstring, implying the danger was
specific to message delivery. Any callback scheduled via add_callback_threadsafe has
the same constraints.
Fix: moved the full warning to a dedicated "IOLoop-thread callbacks" section in the
ThreadSafeChannel class docstring. Reduced basic_consume's docstring to a one-line
cross-reference.
Commit: c2a4afd
Root cause: basic_consume passed on_message_callback directly to the underlying
raw channel, so the broker-delivered channel argument was always a raw
pika.channel.Channel. Users who called channel.basic_ack(...) were bypassing the
ThreadSafeChannel API, which was undocumented and inconsistent.
Fix: introduced _wrapped_callback inside _consume that substitutes self (the
ThreadSafeChannel) for the raw channel argument before forwarding. No allocation
needed - self is already the correct wrapper.
Commit: a83aad3
- 51 unit tests in
tests/unit/thread_safe_connection_tests.py- all pass without a broker - 7 acceptance tests in
tests/acceptance/thread_safe_connection_test.py- require broker hatch run typecheck- clean on all 31 source files
The 7 acceptance tests are:
TestBasicLifecycle- connect, channel, declare, publish, passive declare checkTestConcurrentPublishing- 10 threads behind Barrier; regression for original _tx_buffers raceTestConcurrentPublishAndConsume- basic_qos, basic_consume, 4 publisher threads, 20 messages all ackedTestBrokerDropBlockedInChannel- ForwardServer intercept; blockedchannel()unblocks within 10s after TCP dropTestBrokerDropBlockedInQueueDeclare- same forqueue_declare()TestConcurrentClose- 5 threads callclose()simultaneously; no crash or hangTestContextManager- connection closed on__exit__
This is the topic that was being discussed when the session ended. Luke said he has ideas and wanted to hear the AI's thoughts first before sharing his own. The AI gave its thoughts; Luke said "yes, let's" to the general direction but had not yet shared his specific ideas. This thread needs to be picked up.
Any callback running on the IOLoop thread - on_message_callback and anything
scheduled via add_callback_threadsafe - faces two dangers:
- Heartbeat starvation: slow work blocks the IOLoop, broker teardown follows
- Deadlock: calling any blocking
ThreadSafeChannelmethod (queue_declare,basic_qos,basic_consume,basic_cancel,close) blocks the calling thread waiting for a broker response the IOLoop must process - since the IOLoop is blocked in the callback, it never processes the response
Currently the class docstring warns about this. That is necessary but not sufficient.
Option A - Fail fast on IOLoop-thread calls to blocking methods
Detect threading.current_thread() is self._wrapper._ioloop_thread at the top of
each blocking method and raise RuntimeError immediately. Turns a silent permanent
hang into a clear error. Zero API change, easy to implement. Should happen regardless
of any other option chosen.
Option B - Executor-based dispatch for on_message_callback
Wrap the delivery callback to submit it to a concurrent.futures.ThreadPoolExecutor
with a single worker (to preserve message ordering). The IOLoop hands off each
delivery immediately and returns to I/O. The callback runs in the worker thread where
blocking methods and slow operations are both safe. basic_ack etc. from the worker
thread are already safe via add_callback_threadsafe.
Trade-off: changes the threading context of callbacks (shared state in callbacks now needs locks, but that is arguably correct). Adds a worker thread per consumer (or shared per channel/connection).
Option C - Optional executor parameter on basic_consume
basic_consume accepts an optional executor argument defaulting to None
(current IOLoop-thread behaviour). Users who need blocking work pass
executor=ThreadPoolExecutor(max_workers=1). The wrapper does
executor.submit(_wrapped_callback, ...) instead of calling it directly. Opt-in,
backwards-compatible, gives each consumer its own concurrency model.
AI's recommendation (stated at end of session): Option A (fail fast) as an
immediate fix regardless, plus Option C (optional executor on basic_consume) as the
proper solution. The executor approach is opt-in and does not impose overhead or
threading-model changes on users who keep callbacks fast and non-blocking.
Luke said he has ideas - he was about to share them when the session ended. The next session should ask Luke to share his ideas before proceeding with any implementation.
These were identified during review but not acted on:
_channel_waiters_lockis a slightly misleading name (it guards both_blocking_waitersand_closed_reason, not just channel waiters). Low priority.- No type annotations on
ThreadSafeConnection.__init__parameters (parameters,on_open_error_callback,on_close_callback). Low priority. - The PR is currently DRAFT. At some point Luke will mark it ready. Do not do this without being asked.
This is critical. Luke is strict about this.
- Make a change
- Run tests + typecheck to verify
- Check for trailing whitespace:
grep -Pn ' +$' <file> - Announce the change is ready
- Wait for an explicit 👍
- Only then commit
One logical fix = one commit = one 👍. Never batch multiple fixes into one commit unless Luke explicitly says to.
For commit messages with backticks or apostrophes, write to a temp file:
cat > /tmp/msg.txt << 'EOF'
Subject line
Body text with `backticks` and apostrophes.
EOF
git commit -F /tmp/msg.txtTitle: Add ThreadSafeConnection: fix _tx_buffers race and thread-safety bugs
The PR body was rewritten in this session and covers all the bugs, tests, and checklist items. It is up to date as of the last commit (a83aad3). If further significant changes are made, the PR body should be updated using:
cat > /tmp/pr_body.md << 'EOF'
... content ...
EOF
gh pr edit 1582 --repo pika/pika --body-file /tmp/pr_body.mdNever pass PR body text via inline --body with backticks - use --body-file.
The normal Claude Code memory system is session-specific and will not be available in a new environment. All relevant preferences have been copied to:
/home/lrbakken/development/lukebakken/gist/pika-gh-1582/user-preferences.md
Read that file. Key points:
- No hard-wrapping of prose in .md files (only commit message bodies wrap at 72 chars)
- Read local source files with Read tool, not
gh pr diff - Skip
EnterWorktreeif cwd is already under.claude/worktrees/ - Git/markdown rules are in
~/genai/git/GIT.md- read at session start - Commit workflow rules are in
~/genai/AGENTS.md- read at session start