Optimize pipelining by parsing and prefetching multiple commands

[zuiderkwast]

Instead of parsing only one command per client before executing it, parse multiple commands from the query buffer and batch-prefetch the keys accessed by the commands in the queue before executing them.

This is an optimization for pipelined commands, both with and without I/O threads. The optimization is currently disabled for the replication stream, due to failures (probably caused by how the replication offset is calculated based on the query buffer offset).

• When parsing commands from the input buffer, multiple commands are parsed and stored in a command queue per client.
• In single-threaded mode (I/O threads off) keys are batch-prefetched before the commands in the queue are executed. Multi-key commands like MGET, MSET and DEL benefit from this even if pipelining is not used.
• Prefetching when I/O threads are used does prefetching for multiple clients in parallel. This code takes client command queues into account, improving prefetching when pipelining is used. The batch size is controlled by the existing config prefetch-batch-max-size (default 16), which so far only was used together with I/O threads. The config is moved to a different section in valkey.conf.
• When I/O threads are used and the maximum number of keys are prefetched, a client's command is executed, then the next one in the queue, etc. If there are more commands in the queue for which the keys have not been prefetched (say the client sends 16 pipelined MGET with 16 keys in each) keys for the next few commands in the queue are prefetched before the commands is executed if prefetching has not been done for the next command. (This utilizes the code path used in single-threaded mode.)

Code improvements:

• Decoupling of command parser state and command execution state:
  • The variables reqtype, multibulklen and bulklen refer to the current position in the query buffer. These are no longer reset in resetClient (which runs after each command being executed). Instead, they are reset in the parser code after each completely parsed command.
  • The command parser code is partially decoupled from the client struct. The query buffer is still one per client, but the resulting argument vector is stored in caller-defined variables.

Benchmarking with and without I/O threads, see below.

Fixes #2044

[codecov]

Codecov Report

❌ Patch coverage is 86.91099% with 25 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.21%. Comparing base (ac51515) to head (a4ad52e).
⚠️ Report is 1 commits behind head on unstable.

Files with missing lines	Patch %	Lines
src/memory_prefetch.c	0.00%	18 Missing ⚠️
src/networking.c	96.73%	5 Missing ⚠️
src/server.c	90.00%	2 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##           unstable    #2092      +/-   ##
============================================
+ Coverage     72.02%   72.21%   +0.19%     
============================================
  Files           126      126              
  Lines         70493    70616     +123     
============================================
+ Hits          50771    50994     +223     
+ Misses        19722    19622     -100     

Files with missing lines	Coverage Δ
src/kvstore.c	96.01% <ø> (ø)
src/module.c	9.52% <ø> (ø)
src/server.h	100.00% <ø> (ø)
src/server.c	88.44% <90.00%> (-0.03%)	⬇️
src/networking.c	88.28% <96.73%> (+0.36%)	⬆️
src/memory_prefetch.c	11.26% <0.00%> (-0.77%)	⬇️

... and 15 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[dvkashapov]

Did some benchmarks on AMD EPYC 7713 32 CPUs

• config:

databases 1
save ""
appendonly no
rdbcompression no
activedefrag no
maxclients 1000
io-threads *
io-threads-do-reads yes
protected-mode no
maxmemory 2500mb
maxmemory-policy noeviction
hz 1
latency-monitor-threshold 0

• cset shield
• cpu bind
• cache population
  valkey-benchmark -h <ip> -c 96 --threads 96 -d 256 -t set -n 3000000 -r 3000000 --sequential
• benchmark itself
  valkey-benchmark -h <ip> -c 192 --threads 96 -d 256 -t get -n 10000000 -r 3000000 -P 10

io-threads 1 unstable result:

Summary:
  throughput summary: 352795.91 requests per second
  latency summary (msec):
          avg       min       p50       p95       p99       max
        5.396     0.048     5.279     5.727    10.455   124.287

io-threads 1 this pr result:

Summary:
  throughput summary: 604375.69 requests per second
  latency summary (msec):
          avg       min       p50       p95       p99       max
        3.100     0.040     2.935     3.431     5.791   132.991

io-threads 9 unstable result:

Summary:
  throughput summary: 383435.59 requests per second
  latency summary (msec):
          avg       min       p50       p95       p99       max
        4.981     0.056     4.959     5.255     5.503    79.039

io-threads 9 this pr result:

Summary:
  throughput summary: 738334.31 requests per second
  latency summary (msec):
          avg       min       p50       p95       p99       max
        2.553     0.032     2.567     2.751     2.919    36.703

Seems like a solid 2x improvement in pipelining workload! Also did not find any memory leaks or instabilities during testing, LGTM!

[PingXie]

@zuiderkwast, I think our changes are indeed very close, especially around the command queue concept. that being said (and I know I am super biased here :)), I think my PR is closer to the finishing line. Do you want to hop onto my branch and collaborate on mine next?

https://github.com/PingXie/valkey/tree/pipeline

[madolson]

The overall approach makes sense to me, I have some minor thoughts but I haven't thought to much how to think about this vs Ping's PR yet.

Looking back at the original IO threading, we batch N keys, maybe we should change it to be batch N commands instead, since that will make sure we are fully parsing a command. That way we can probably more tightly unify the code.

[zuiderkwast]

The overall approach makes sense to me, I have some minor thoughts but I haven't thought to much how to think about this vs Ping's PR yet.

Thanks @madolson for taking the time to look at this.

Looking back at the original IO threading, we batch N keys, maybe we should change it to be batch N commands instead, since that will make sure we are fully parsing a command. That way we can probably more tightly unify the code.

I think we shall prefetch full commands, but we can still configure it as the number of keys. If we go over the number N when adding a command to the batch, then we can allow prefetching a few more, let's say up to N*2. That way we prefetch by default 16 keys, but allow prefetching up to 32 keys to stick to full commands.

We should definitely unify the IO-threaded and single-threaded code paths for this code and more of the parse-lookup-prefetch-execute. I avoided refactoring to keep the diff small but I think it's necessary to at least rename some functions to keep this comprehensive.

[PingXie]

Looking back at the original IO threading, we batch N keys, maybe we should change it to be batch N commands instead, since that will make sure we are fully parsing a command. That way we can probably more tightly unify the code.

I think we shall prefetch full commands, but we can still configure it as the number of keys. If we go over the number N when adding a command to the batch, then we can allow prefetching a few more, let's say up to N*2. That way we prefetch by default 16 keys, but allow prefetching up to 32 keys to stick to full commands.

my preference, for the sake of simplicity, is that we continue to count keys. as soon as we use up max_pretech_size we should just execute the command, irrespective of the number of the keys referenced by the command (e.g., mget) or the number of commands batached.

ultimately max_prefetch_size is determined by the L1 cache size. There is only so much we can prefetch. over-prefetching will lead to L1 cache thrashing, nullifying the effect.

the L1 data cache on Intel cpus is 32 KiB or 48 KiB, typically, with a cache line size of 64 bytes, which gives us 512 or 768 cache lines, respectively. I haven't counted carefully the number of memory accesses needed to prefetch a key-value pair of strings in the new hashtable but I guess it would be 5 (including both keys and hash table)? Assuming 5, prefetching 32 keys would consume 160 cache lines, which is not a trivial number compared to 512 already.

[secwall]

Hello, it seems that this pull request has some conflicts with the unstable branch. Do you need any assistance working on it? Merging it into 9.0 would be beneficial for many users (a lot of clients use pipelines automatically).

[zuiderkwast]

After discussion with @PingXie and others, I wanted to rewrite it a bit and unify the prefetching code paths for io-threaded and single-threaded modes. I probably can't make it to 9.0 though.

How would folks (@PingXie, @madolson) feel about merging this more or less as is for 9.0 and improving it later?

[PingXie]

How would folks (@PingXie, @madolson) feel about merging this more or less as is for 9.0 and improving it later?

works for me. Will review this pr next.

[madolson]

We're getting de-sync errors in the test:

24786:S 19 Aug 2025 09:52:35.066 # Protocol error (Master using the inline protocol. Desync?) from client: id=32 addr=127.0.0.1:21116 laddr=127.0.0.1:61112 fd=25 name= age=11 idle=0 flags=M capa= db=0 sub=0 psub=0 ssub=0 multi=-1 watch=0 qbuf=16010 qbuf-free=49520 argv-mem=0 multi-mem=0 rbs=1024 rbp=0 obl=0 oll=0 omem=0 tot-mem=67136 events=r cmd=set user=(superuser) redir=-1 resp=2 lib-name= lib-ver= tot-net-in=12800433 tot-net-out=451 tot-cmds=20579. Query buffer: '6ZJ}:226..*2..$6..UNLINK..$8..{6ZJ}:22..*2..$6..UNLINK..$8..{6ZJ' (... more 14880 bytes ...) '..*2..$6..UNLINK..$9..{6ZJ}:207..*2..$6..UNLINK..$9..{6ZJ}:280..'

Later in the day update:
The target eventually stops processing incoming requests, but it is not consistent. There doesn't seem to be anything non-deterministic about it that I can see outside of the speed of processing. @murphyjacob4 Since your back, if you have any thoughts as well for the test failures.

Just to comment since Viktor messaged me on slack. Apparently there is something getting incorrectly updated for replication offset. This will have impact for other deeply pipelined replication streams, but it looks like only the slot migration tests are triggering that.

[murphyjacob4]

Apparently there is something getting incorrectly updated for replication offset. This will have impact for other deeply pipelined replication streams, but it looks like only the slot migration tests are triggering that.

Since slot migration uses chaining replication, if the replication offset is wrong, what is sent to the replicas of the target would be wrong:

https://github.com/valkey-io/valkey/blob/unstable/src/networking.c#L3677-L3689

So that would make sense why we see the desync errors