Don't re-run a blocked client's pending command in handleReadJobs() - #4370
Don't re-run a blocked client's pending command in handleReadJobs()#4370dgershko wants to merge 1 commit into
Conversation
blockForKeys() sets pending_command so a blocking command is re-executed
once the client is unblocked. handleReadJobs() calls
processPendingCommandAndInputBuffer() after processClientsCommandsBatch(),
which re-enters processCommand() on the client it just blocked, blocking it
a second time.
With io-threads active a single in-flight BLPOP reports blocked_clients:2,
and the counter leaks one reference per blocking command:
valkey-server --io-threads 2 --io-threads-always-active yes
# after 5 BLPOP timeouts, all those clients since disconnected:
blocked_clients:5
Guard the call on the states that mean the client already owns an
unprocessed command, matching processUnblockedClients(). Guarding the call
rather than the loop iteration keeps blocked clients reaching
connUpdateState().
Signed-off-by: Daniel Gershkovich <dgershko@gmail.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughThe IO-thread read completion path now skips blocked, unblocked, and immediately closing clients. Regression tests cover deferred ChangesIO-thread blocking commands
Estimated code review effort: 2 (Simple) | ~10 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
| if (!c->flag.blocked && !c->flag.unblocked && !c->flag.close_asap && | ||
| processPendingCommandAndInputBuffer(c) == C_OK) | ||
| beforeNextClient(c); |
There was a problem hiding this comment.
@valkey-review-bot can you review this PR, this also adds 2 more checks which might be more defensive rather then targeting to solve a real bug (!c->flag.unblocked && !c->flag.close_asap) - can you please verify that and if not please suggest extending the tests
|
The |
![valkey-review-bot[bot]](https://avatars.githubusercontent.com/in/3633497?s=60&v=4){kind=small}
| if (!c || !c->conn) continue; | ||
|
|
||
| if (processPendingCommandAndInputBuffer(c) == C_OK) beforeNextClient(c); | ||
| if (!c->flag.blocked && !c->flag.unblocked && !c->flag.close_asap && |
There was a problem hiding this comment.
unblocked and close_asap are valid guards, but neither is covered by the new BLPOP tests. queueClientForReprocessing() sets unblocked specifically so processUnblockedClients() owns the pending command (src/blocked.c:206-212), while freeClientAsync() sets close_asap before the client is actually freed (src/networking.c:2271-2275); running this drain in either state would bypass that ownership/teardown. Extend the IO-thread tests with a CLIENT PAUSE/CLIENT UNPAUSE case for unblocked, and a completion whose client is killed or hits a query/output limit for close_asap, then verify no queued command executes and the client is removed.
ranshid
added
the
run-extra-tests
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## unstable #4370 +/- ##
============================================
+ Coverage 76.98% 77.00% +0.01%
============================================
Files 162 162
Lines 81786 81788 +2
============================================
+ Hits 62965 62982 +17
+ Misses 18821 18806 -15
🚀 New features to boost your workflow:
|
2 participants
blockForKeys() sets pending_command so a blocking command is re-executed once the client is unblocked. handleReadJobs() calls processPendingCommandAndInputBuffer() after processClientsCommandsBatch(), which re-enters processCommand() on the client it just blocked, blocking it a second time.
With io-threads active a single in-flight BLPOP reports blocked_clients:2, and the counter leaks one reference per blocking command:
Reproduction:
./runtest --single unit/type/list --io-threads --only "Unblock fairness is kept while pipelining"Existing coverage only fails under --io-threads, which CI never passes; the new test in unit/io-threads.tcl pins
io-threads itself so a default run covers this path.
Guard the call on the states that mean the client already owns an unprocessed command, matching processUnblockedClients(). Guarding the call rather than the loop iteration keeps blocked clients reaching connUpdateState().