concurrency: fix meter, executor, work-queue and transport races - #23
Merged
Conversation
- Meter.update/reset guard sum/count/max/min/last with a lock, fixing lost updates under concurrent task execution - MemoryTransport.send raises TransportFullError instead of silently dropping messages when the bounded queue is full - Executor.start()/submit() make the running-flag check and pool construction atomic, preventing double pool construction - WorkQueue.remove_consumer keys handlers by consumer_id instead of popping the last-registered one - reaper-driven requeue now calls _try_dispatch so push-only consumers aren't starved after a visibility timeout - add locks around the _CLASS_SUBSCRIBERS registry and RPCServer._methods Adds tests/test_concurrency.py: producer/consumer exactly-once and failure-handling stress tests, remove_consumer correctness, reaper starvation regression, meter race, and executor pool race.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes the thread-safety defects found in a concurrency audit. The library ships concurrent execution modes (ConcurrentDispatcher, Executor THREAD/PROCESS, per-message consumer threads), so these are hit under real use.
update/reset/value/statsread-modify-wrote sum/count/max/min/last with no lock -> concurrent task completions dropped updates silently. Added_state_lock; dispatch stays outside it.put_nowait+ bareexcept: passmeant messages vanished fromreceive()under load with no signal. Now raises a newTransportFullError(mirrorsQueueFullError) after notifying live subscribers; no silent drop.submit()s each build a pool and leak one. The running-flag check + pool construction is now one atomic block under the lock.pop()removed the last-registered handler, not the one for the givenconsumer_id. Groups now carry(consumer_id, handler)._try_dispatchso push consumers get the redelivery._CLASS_SUBSCRIBERS,RPCServer._methods) now have explicit locks.New
tests/test_concurrency.py(6 tests). The straight-line races (Meter, Executor) barely interleave under CPython 3.14's specializing interpreter, so those tests inject a small delay to widen the check-then-act window; each fails on the pre-fix code with the exact symptom and passes after. Full suite 188 passed, mypy strict + ruff clean.Not included: a callback-group / mutually-exclusive-subscriber abstraction (so a shared-state handler can declare it must not run concurrently) -- that's a Dispatcher public-contract change and belongs as a separate designed feature.