Admit blocked InsertTx calls in FIFO order and bound the waiter queue - #4184
Conversation
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
|
The latest Buf updates on your PR. Results from workflow Buf / buf (pull_request).
|
PR SummaryMedium Risk Overview Adds Reviewed by Cursor Bugbot for commit 7aec8f7. Bugbot is set up for automated code reviews on this repo. Configure here. |
![seidroid[bot]](https://avatars.githubusercontent.com/in/2745135?s=60&v=4){kind=small}
There was a problem hiding this comment.
The FIFO admission queue is correct as far as I can trace it — every full→non-full transition (prune, dequeue, close) signals the head, and all admitted stores happen under the mempool lock, so there is no lost-wakeup or spin path. Four non-blocking points: the new MaxPendingInserts default is applied per-caller rather than at a choke point, TryInsertTx can still jump the queue, dequeue is O(N) under the hot lock, and pendingInserts is production state read only by tests.
Findings: 0 blocking | 4 non-blocking | 4 posted inline
Blockers
- None at the file/PR level.
Non-blocking
- None at the file/PR level.
- 4 suggestion(s)/nit(s) flagged inline on specific lines.
| nonce, ok := m.evmNonces[addr] | ||
| if !ok { | ||
| nonce = s.app.EvmNonce(addr) | ||
| if m.IsFull() && !waitIfFull { |
There was a problem hiding this comment.
[suggestion] TryInsertTx is checked against IsFull() only and never consults isHead, so it takes capacity ahead of every queued waiter. With the old code all N waiters woke on each ctrl.Updated() and raced the TryInsertTx caller for the lock; now exactly one waiter is signalled per free slot and must re-acquire the lock, so a sustained TryInsertTx stream can win that race repeatedly, driving the head back through the t.admitted.Store(false) path each time.
These are both live RPC paths — broadcast_tx_async calls TryInsertTx and broadcast_tx_sync/broadcast_tx calls InsertTx (internal/rpc/core/mempool.go:61,90) — so under async load-generator traffic (the exact scenario in the PR description) sync broadcasters can be starved for an unbounded time, then eventually rejected once the queue hits MaxPendingInserts. The inline comment at line 340 acknowledges the interaction but the InsertTx godoc ("admitting blocked calls in arrival order") reads as a stronger guarantee than holds.
At minimum, scope the godoc to arrival order among blocked InsertTx calls; if sync broadcasters should not be starvable, TryInsertTx needs to respect a non-empty queue too.
There was a problem hiding this comment.
Scoped the InsertTx godoc in 99d78dc: ordering is among blocked InsertTx calls, and TryInsertTx does not queue and may take freed capacity ahead of them. Keeping TryInsertTx semantics unchanged is a stated requirement of this PR; making it respect a non-empty queue is a behaviour change that should be decided separately.
| // dequeue removes t from the admission queue and passes the turn to the next waiter. | ||
| func (mp *mempool) dequeue(m *mempoolInner, t *insertTicket) { | ||
| if i := slices.Index(m.waiters, t); i >= 0 { | ||
| m.waiters = slices.Delete(m.waiters, i, i+1) |
There was a problem hiding this comment.
[suggestion] slices.Delete(m.waiters, 0, 1) on the common (head) path memmoves the remaining len(m.waiters)-1 pointers while holding the mempool lock, so admission stays O(N) in queue depth — at the 4096 default that is a ~32 KiB copy on the hot lock for every admitted tx. It is far cheaper than the N goroutine wake-ups it replaces, but it is avoidable: utils.RingBuf (libs/utils/ringbuf.go) already exists in-tree, and a head index or linked list makes both isHead and the head dequeue O(1). The cancelled-waiter path can stay O(N) since it is rare.
There was a problem hiding this comment.
Agreed it is O(N) in queue depth, bounded by MaxPendingInserts (a ≤32 KiB memmove at the default) and already amortised against the per-tx CheckTx work. Leaving it as a slice for now to keep this PR minimal; happy to switch to utils.RingBuf or a head index if the memmove shows up in profiles.
| type mempool struct { | ||
| inner utils.Watch[*mempoolInner] | ||
| // pendingInserts is the number of InsertTx calls blocked in the admission queue. | ||
| pendingInserts utils.AtomicSend[uint64] |
There was a problem hiding this comment.
[suggestion] pendingInserts is production state whose only readers are mempool_test.go — nothing outside tests calls Load or Wait on it — and each Store allocates a fresh version plus a channel (utils.AtomicSend.Store), so it adds two allocations per enqueue and per dequeue to the hot path.
Given the PR is motivated by an operational incident (goroutine and CPU excess on an ingesting validator), queue depth and the over-bound rejection count are precisely what an operator would want to see. Publishing it through the existing autobahn metrics facility (internal/autobahn/data/metrics, which already has gauge/counter helpers) would make the field earn its place in the production struct rather than existing solely as a test synchronization handle.
There was a problem hiding this comment.
Metrics for this path are being added in the sibling PR #4179, which this PR must not depend on; exposing queue depth and over-bound rejections there (or as a follow-up once both land) seems the right place. pendingInserts is kept here as the deterministic synchronization handle the tests need to avoid sleeps.
…fo-insert-admission # Conflicts: # sei-tendermint/internal/autobahn/producer/mempool.go # sei-tendermint/internal/autobahn/producer/mempool_test.go
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## giga-1 #4184 +/- ##
==========================================
- Coverage 65.55% 65.54% -0.01%
==========================================
Files 2081 2076 -5
Lines 157460 157131 -329
==========================================
- Hits 103222 102999 -223
+ Misses 54097 53991 -106
Partials 141 141
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
2 participants
When the producer mempool is full, every InsertTx call blocked in producer.State.insertTx waited on the mempool Watch, so each ctrl.Updated() (a sealed block, a prune, a new block starting) woke all blocked inserters, each of which re-took the lock and re-evaluated IsFull. Under load-generator traffic the ingesting validator accumulates thousands of such inserters, which makes every mempool update O(N) in lock contention and wake-ups and is consistent with the goroutine and CPU excess seen on that node.
Blocked inserters now take a ticket in a FIFO queue on mempoolInner and wait on a per-ticket AtomicSend. Only the head of the queue is signalled, and only when the mempool has capacity: after a prune, after the head itself is admitted or leaves, and on close. A waiter whose context is cancelled removes its ticket and passes the turn on, and closing the mempool releases every waiter with ErrNotProducing as before. A new producer.Config field MaxPendingInserts (default 4096, wired wherever producer.Config is built) caps the number of blocked inserters; past that InsertTx fails immediately with the mempool-full error so clients back off instead of piling up goroutines. TryInsertTx is unchanged.
New tests cover FIFO order under concurrent inserters, one admission per freed slot, a cancelled waiter not blocking the queue, release on close, and the bound. The producer package passes under -race, and the touched packages pass go vet, golangci-lint v2.13.2 and the formatters.