Cluster subscribe

[adam-fowler]

ValkeyClient has a subscribe command that allocates a single connection for all subscriptions. This PR adds something similar except it allocates a single connection from each primary node in the cluster (once we have reading from replicas we can extend to replicas).

To get this working

• I had to move the subscription connection management code from ValkeyClient to ValkeyNodeClient.
• I added an extra state shutdown to the subscription connection state machine.
• I added a shutdownSubscriptionConnection function that sets the shutdown subscription and resumes the release connection continuation.
• When subscribe is called cluster client chooses random node and uses subscription connection from that node.

[github-actions]

✅ Pull request no significant performance differences ✅

Summary

New baseline 'pull_request' is WITHIN the 'main' baseline thresholds.

Full Benchmark Comparison

Comparing results between 'main' and 'pull_request'

Host '6f7806e4514a' with 4 'x86_64' processors with 15 GB memory, running:
#18~24.04.1-Ubuntu SMP Sat Jun 28 04:46:03 UTC 2025

ValkeyBenchmarks

Client: GET benchmark metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	83	83	83	86	88	88	88	6
pull_request	81	82	83	87	87	87	87	6
Δ	-2	-1	0	1	-1	-1	-1	0
Improvement %	2	1	0	-1	1	1	1	0

Client: GET benchmark | parallel 20 | 20 concurrent connections metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	95	101	105	107	108	109	109	23
pull_request	98	100	105	107	110	113	113	23
Δ	3	-1	0	0	2	4	4	0
Improvement %	-3	1	0	0	-2	-4	-4	0

Connection: GET benchmark metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	4	4	4	4	4	4	4	8
pull_request	4	4	4	4	4	4	4	8
Δ	0	0	0	0	0	0	0	0
Improvement %	0	0	0	0	0	0	0	0

Connection: GET benchmark – NoOpTracer metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	7	8	8	10	10	10	10	8
pull_request	7	8	8	10	10	10	10	8
Δ	0	0	0	0	0	0	0	0
Improvement %	0	0	0	0	0	0	0	0

Connection: Pipeline array benchmark metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	37	37	38	38	43	43	43	4
pull_request	37	37	37	37	43	43	43	4
Δ	0	0	-1	-1	0	0	0	0
Improvement %	0	0	3	3	0	0	0	0

Connection: Pipeline benchmark metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	36	37	37	37	37	37	37	4
pull_request	37	37	37	37	37	37	37	4
Δ	1	0	0	0	0	0	0	0
Improvement %	-3	0	0	0	0	0	0	0

HashSlot – {user}.whatever metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	0	0	0	0	0	0	0	18
pull_request	0	0	0	0	0	0	0	18
Δ	0	0	0	0	0	0	0	0
Improvement %	0	0	0	0	0	0	0	0

ValkeyCommandEncoder – Command with 7 words metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	0	0	0	0	0	0	0	737
pull_request	0	0	0	0	0	0	0	738
Δ	0	0	0	0	0	0	0	1
Improvement %	0	0	0	0	0	0	0	1

ValkeyCommandEncoder – Simple GET metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	0	0	0	0	0	0	0	1875
pull_request	0	0	0	0	0	0	0	1866
Δ	0	0	0	0	0	0	0	-9
Improvement %	0	0	0	0	0	0	0	-9

ValkeyCommandEncoder – Simple MGET 15 keys metrics

Malloc (total): results within specified thresholds, fold down for details.

Malloc (total) *	p0	p25	p50	p75	p90	p99	p100	Samples
main	0	0	0	0	0	0	0	358
pull_request	0	0	0	0	0	0	0	352
Δ	0	0	0	0	0	0	0	-6
Improvement %	0	0	0	0	0	0	0	-6

[nilanshu-sharma]

Minor comment: Should be this called subscriptionConnectionStateMachineMutex or subscriptionConnectionStateMachineLock as it is a mutex around ConnectionStateMachine

[adam-fowler]

I'm happy with its name. It is a bit more than a mutex as it manages the leasing of the connection as well

[nilanshu-sharma]

Noticing there is a + in the filename, is that intentional, or did you mean to name the file like ValkeyClientSubscribe ?

[adam-fowler]

The + indicates the code in this file is an extension. In this case it is ValkeyClient being extended to include subscription functionality