[ClusterPipeline] ExecutorService/thread is created and destroyed too frequently in ClusterPipeline

[xrayw]

Title

Allow externally managed ExecutorService for multi-node pipeline operations

Description

Problem:

Multi-node pipelines create a new ExecutorService on every sync() call (line 93), causing significant overhead in high-throughput applications:

@Override
public final void sync() {
  // ...
  if (multiNode) {
    executorService = Executors.newFixedThreadPool(MULTI_NODE_PIPELINE_SYNC_WORKERS);
    executor = executorService;
  }
  // ...
}

Thread creation is expensive:

• Allocates large memory blocks for thread stacks
• Requires system calls to register native threads
• Adds latency to each pipeline operation

In high-throughput scenarios (e.g., 100k ops/sec), this overhead can consume 40% of CPU.

Proposed Solution:

Allow users to provide an externally managed ExecutorService that can be reused across pipeline operations

Benefits:

• Eliminates thread creation overhead
• Allows tuning executor to specific workload patterns
• Maintains backward compatibility with current behavior as default

Related:

• Approach 1 Provide executor on pipeline creation - Initial suggested approach with providing
• Approach 2 Pipeline executor provided on client level & shared by pipelines
• Current workaround: Adjust MULTI_NODE_PIPELINE_SYNC_WORKERS (doesn't solve the core issue)

[ggivo]

Thanks for reporting it. Multiple sources have raised this valid concern.
Link to PR #4070 trying to address the same issue with ongoing discussions

[xrayw]

@ggivo thank you, and by the way

jedis/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java

Line 62 in 1d1b268

pipelinedResponses.putIfAbsent(nodeKey, new LinkedList<>());

is it should use computeIfAbsent? instead of putIfAbsent.

pipelinedResponses.putIfAbsent(nodeKey, new LinkedList<>());
queue = pipelinedResponses.get(nodeKey);

-> 
queue = pipelinedResponses.computeIfAbsent(nodeKey, (_) -> new LinkedList<>())

[ggivo]

is it should use computeIfAbsent? instead of putIfAbsent.

Do you have anything particular in mind?
Otherwise I would say, No.
We have already checked if (pipelinedResponses.containsKey(nodeKey)) and also not using the result.

[xrayw]

Because each putIfAbsent will create a list object, but if the key already exists in the map, this list doesn't need to be created.

[ggivo]

From a performance optimization perspective, I think there is not much of a difference, since we will execute putIfAbsent code only if the pipelinedResponses does not contain the nodeKey.

The current implementation of MultiNodePipelineBase is not thread-safe. And since we already check for if (pipelinedResponses.containsKey(nodeKey)), we will run trough pipelinedResponses.putIfAbsent(nodeKey, new LinkedList<>()); only the first time when we need to initialise it.

[xrayw]

@ggivo Yes, You're right. we can just even use put. because we checked it not contains.

[xrayw]

jedis/src/main/java/redis/clients/jedis/MultiNodePipelineBase.java

Line 98 in 1d1b268

executorService.submit(() -> {

And if pipelinedResponses.size() == 1, we don't need to submit it to ExecutorService. but run it on the work thread directly. to avoid the cost of thread context switch.

and we can run the last Map.Entry<HostAndPort, Queue<Response<?>>> on the work thread to save resources.

If you think so, I'd like to create a PR.
Thanks.

[ggivo]

@xrayw
Assuming there are pipelines with all commands ending up on the same node, it sounds reasonable. The only drawback I see is adding a bit of complexity to the code.

Note: Since it does not address the main concern described in this Issue, I suggest opening a dedicated Issue to track this optimization.

[xrayw]

Thank you ,i createed a new issue to track it. #4148