Skip to content

Commit bf1711b

Browse files
ggivoggivo
committed
backport: [ClusterPipeline] ExecutorService/thread is created and destroyed too frequently in ClusterPipeline #4479
(cherry picked from commit 7744cd4)
1 parent c37b259 commit bf1711b

5 files changed

Lines changed: 291 additions & 10 deletions

File tree

‎src/main/java/redis/clients/jedis/ClusterPipeline.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
import java.time.Duration;
44
import java.util.Set;
5+
import java.util.concurrent.ExecutorService;
6+
57
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
68
import redis.clients.jedis.providers.ClusterConnectionProvider;
79
import redis.clients.jedis.util.IOUtils;
@@ -40,6 +42,12 @@ public ClusterPipeline(ClusterConnectionProvider provider, ClusterCommandObjects
4042
this.provider = provider;
4143
}
4244

45+
ClusterPipeline(ClusterConnectionProvider provider, ClusterCommandObjects commandObjects,
46+
ExecutorService executorService) {
47+
super(commandObjects, executorService);
48+
this.provider = provider;
49+
}
50+
4351
private static ClusterCommandObjects createClusterCommandObjects(RedisProtocol protocol) {
4452
ClusterCommandObjects cco = new ClusterCommandObjects();
4553
if (protocol == RedisProtocol.RESP3) cco.setProtocol(protocol);

‎src/main/java/redis/clients/jedis/JedisCluster.java‎

Lines changed: 42 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import java.util.Collections;
55
import java.util.Map;
66
import java.util.Set;
7+
import java.util.concurrent.ExecutorService;
78

89
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
910

@@ -412,11 +413,49 @@ public void ssubscribe(BinaryJedisShardedPubSub jedisPubSub, final byte[]... cha
412413
}
413414
// commands
414415

416+
/**
417+
* Creates a new pipeline for executing commands in a Redis Cluster.
418+
*
419+
* <p>Pipelining allows batching multiple commands for more efficient execution
420+
* by reducing network round-trips. In a cluster environment, commands are routed
421+
* to the appropriate nodes based on key hash slots.</p>
422+
*
423+
* <p>If the pipeline spans multiple nodes, a dedicated {@link ExecutorService} is
424+
* created internally to execute requests in parallel and shutdown when the pipeline
425+
* is synced.</p>
426+
*
427+
* @return a new {@link ClusterPipeline} instance
428+
* @see #pipelined(ExecutorService)
429+
*/
415430
@Override
416431
public ClusterPipeline pipelined() {
417-
return new ClusterPipeline((ClusterConnectionProvider) provider, (ClusterCommandObjects) commandObjects);
418-
}
419-
432+
return pipelined(null);
433+
}
434+
435+
/**
436+
* Creates a new pipeline for executing commands in a Redis Cluster using the provided executor.
437+
*
438+
* <p>Pipelining allows batching multiple commands for more efficient execution
439+
* by reducing network round-trips. In a cluster environment, commands are routed
440+
* to the appropriate nodes based on key hash slots.</p>
441+
*
442+
* <p>If the pipeline spans multiple nodes, the provided {@link ExecutorService} is
443+
* used to execute requests in parallel. The caller is responsible for managing
444+
* the lifecycle of this executor (creation, shutdown, etc.).</p>
445+
*
446+
* <p>If {@code null} is provided, a dedicated executor is created and managed
447+
* internally, similar to {@link #pipelined()}.</p>
448+
*
449+
* @param executorService the executor to use for multi-node execution, or {@code null}
450+
* @return a new {@link ClusterPipeline} instance
451+
*/
452+
public ClusterPipeline pipelined(ExecutorService executorService) {
453+
return new ClusterPipeline(
454+
(ClusterConnectionProvider) provider,
455+
(ClusterCommandObjects) commandObjects,
456+
executorService
457+
);
458+
}
420459
/**
421460
* @param doMulti param
422461
* @return nothing

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

Lines changed: 58 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -32,12 +32,24 @@ public abstract class MultiNodePipelineBase extends AbstractPipeline {
3232
private final Map<HostAndPort, Connection> connections;
3333
private volatile boolean syncing = false;
3434

35+
/**
36+
* External executor service to use for {@code sync()}. If not set, a new executor service will be
37+
* created for each {@code sync()} call.
38+
*/
39+
private final ExecutorService sharedExecutorService;
40+
3541
public MultiNodePipelineBase(CommandObjects commandObjects) {
36-
super(commandObjects);
37-
pipelinedResponses = new LinkedHashMap<>();
38-
connections = new LinkedHashMap<>();
42+
this(commandObjects, null);
3943
}
4044

45+
MultiNodePipelineBase(CommandObjects commandObjects, ExecutorService executorService) {
46+
super(commandObjects);
47+
pipelinedResponses = new LinkedHashMap<>();
48+
connections = new LinkedHashMap<>();
49+
this.sharedExecutorService = executorService;
50+
}
51+
52+
4153
protected abstract HostAndPort getNodeKey(CommandArguments args);
4254

4355
protected abstract Connection getConnection(HostAndPort nodeKey);
@@ -90,7 +102,7 @@ public final void sync() {
90102
Executor executor;
91103
ExecutorService executorService = null;
92104
if (multiNode) {
93-
executorService = Executors.newFixedThreadPool(MULTI_NODE_PIPELINE_SYNC_WORKERS);
105+
executorService = getPipelineExecutor();
94106
executor = executorService;
95107
} else {
96108
executor = Runnable::run;
@@ -134,12 +146,53 @@ public final void sync() {
134146
log.error("Thread is interrupted during sync.", e);
135147
}
136148

137-
executorService.shutdownNow();
149+
releasePipelineExecutor(executorService);
138150
}
139151

140152
syncing = false;
141153
}
142154

155+
/**
156+
* Acquires the executor service to run multi-node pipeline commands.
157+
* <p>
158+
* If a shared executor is provided by the user, it is returned.
159+
* Otherwise, a new dedicated executor is created for this pipeline.
160+
* </p>
161+
*/
162+
private ExecutorService getPipelineExecutor() {
163+
return isUsingSharedExecutor()
164+
? this.sharedExecutorService
165+
: createDedicatedPipelineExecutor();
166+
}
167+
168+
/**
169+
* Releases the executor service used by the pipeline.
170+
* <p>
171+
* Dedicated executors are shut down after use.
172+
* Shared executors are managed externally and not shut down.
173+
* </p>
174+
*/
175+
private void releasePipelineExecutor(ExecutorService executorService) {
176+
if (!isUsingSharedExecutor()) {
177+
executorService.shutdownNow();
178+
}
179+
}
180+
181+
/**
182+
* Returns true if this pipeline is using a shared executor service
183+
* provided externally.
184+
*/
185+
private boolean isUsingSharedExecutor() {
186+
return this.sharedExecutorService != null;
187+
}
188+
189+
/**
190+
* Creates a new dedicated executor for multi-node pipeline execution.
191+
*/
192+
private ExecutorService createDedicatedPipelineExecutor() {
193+
return Executors.newFixedThreadPool(MULTI_NODE_PIPELINE_SYNC_WORKERS);
194+
}
195+
143196
@Deprecated
144197
public Response<Long> waitReplicas(int replicas, long timeout) {
145198
return appendCommand(commandObjects.waitReplicas(replicas, timeout));

‎src/main/java/redis/clients/jedis/RedisClusterClient.java‎

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import java.util.Collections;
55
import java.util.Map;
66
import java.util.Set;
7+
import java.util.concurrent.ExecutorService;
78

89
import redis.clients.jedis.builders.ClusterClientBuilder;
910
import redis.clients.jedis.executors.ClusterCommandExecutor;
@@ -184,12 +185,49 @@ public void ssubscribe(BinaryJedisShardedPubSub jedisPubSub, final byte[]... cha
184185
}
185186
// commands
186187

188+
/**
189+
* Creates a new pipeline for executing commands in a Redis Cluster.
190+
*
191+
* <p>Pipelining allows batching multiple commands for more efficient execution
192+
* by reducing network round-trips. In a cluster environment, commands are routed
193+
* to the appropriate nodes based on key hash slots.</p>
194+
*
195+
* <p>If the pipeline spans multiple nodes, a dedicated {@link ExecutorService} is
196+
* created internally to execute requests in parallel and shutdown when the pipeline
197+
* is synced.</p>
198+
*
199+
* @return a new {@link ClusterPipeline} instance
200+
* @see #pipelined(ExecutorService)
201+
*/
187202
@Override
188203
public ClusterPipeline pipelined() {
189-
return new ClusterPipeline((ClusterConnectionProvider) provider,
190-
(ClusterCommandObjects) commandObjects);
204+
return pipelined(null);
191205
}
192206

207+
/**
208+
* Creates a new pipeline for executing commands in a Redis Cluster using the provided executor.
209+
*
210+
* <p>Pipelining allows batching multiple commands for more efficient execution
211+
* by reducing network round-trips. In a cluster environment, commands are routed
212+
* to the appropriate nodes based on key hash slots.</p>
213+
*
214+
* <p>If the pipeline spans multiple nodes, the provided {@link ExecutorService} is
215+
* used to execute requests in parallel. The caller is responsible for managing
216+
* the lifecycle of this executor (creation, shutdown, etc.).</p>
217+
*
218+
* <p>If {@code null} is provided, a dedicated executor is created and managed
219+
* internally, similar to {@link #pipelined()}.</p>
220+
*
221+
* @param executorService the executor to use for multi-node execution, or {@code null}
222+
* @return a new {@link ClusterPipeline} instance
223+
*/
224+
public ClusterPipeline pipelined(ExecutorService executorService) {
225+
return new ClusterPipeline(
226+
(ClusterConnectionProvider) provider,
227+
(ClusterCommandObjects) commandObjects,
228+
executorService
229+
);
230+
}
193231
/**
194232
* @param doMulti param
195233
* @return nothing

‎src/test/java/redis/clients/jedis/ClusterPipeliningTest.java‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package redis.clients.jedis;
22

3+
import static java.util.concurrent.TimeUnit.SECONDS;
34
import static org.hamcrest.Matchers.contains;
45
import static org.hamcrest.MatcherAssert.assertThat;
56
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
@@ -13,6 +14,10 @@
1314
import static redis.clients.jedis.Protocol.CLUSTER_HASHSLOTS;
1415

1516
import java.util.*;
17+
import java.util.concurrent.CountDownLatch;
18+
import java.util.concurrent.ExecutorService;
19+
import java.util.concurrent.Executors;
20+
import java.util.concurrent.Future;
1621

1722
import org.hamcrest.MatcherAssert;
1823
import org.hamcrest.Matchers;
@@ -1171,4 +1176,142 @@ private static void assertThreadsCount() {
11711176
.count();
11721177
MatcherAssert.assertThat(count, Matchers.lessThanOrEqualTo(20));
11731178
}
1179+
1180+
@Test
1181+
public void sharedExecutorPipelineDoesNotShutdownSharedExecutor() {
1182+
ExecutorService executorService = Executors.newFixedThreadPool(3);
1183+
try ( RedisClusterClient cluster = RedisClusterClient.builder().nodes(nodes).clientConfig(DEFAULT_CLIENT_CONFIG).build()){
1184+
try (ClusterPipeline pipeline = cluster.pipelined(executorService)) {
1185+
// multiple keys at different slots, to ensure multi-node pipeline
1186+
pipeline.set("key1", "value1");
1187+
pipeline.set("key2", "value2");
1188+
pipeline.set("key3", "value3");
1189+
pipeline.sync();
1190+
}
1191+
} finally {
1192+
assertFalse(executorService.isShutdown());
1193+
executorService.shutdown();
1194+
}
1195+
}
1196+
@Test
1197+
public void sharedExecutorPipelineKeysAtSameNode() {
1198+
try (RedisClusterClient cluster = RedisClusterClient.builder().nodes(nodes).clientConfig(DEFAULT_CLIENT_CONFIG).build()) {
1199+
ExecutorService executorService = Executors.newFixedThreadPool(3);
1200+
// test simple key
1201+
cluster.set("foo", "bar");
1202+
1203+
try (ClusterPipeline pipeline = cluster.pipelined(executorService)) {
1204+
Response<String> foo = pipeline.get("foo");
1205+
pipeline.sync();
1206+
1207+
assertEquals("bar", foo.get());
1208+
}
1209+
1210+
// test multi key but at same node
1211+
int cnt = 3;
1212+
String prefix = "{foo}:";
1213+
for (int i = 0; i < cnt; i++) {
1214+
String key = prefix + i;
1215+
cluster.set(key, String.valueOf(i));
1216+
}
1217+
1218+
try (ClusterPipeline pipeline = cluster.pipelined(executorService)) {
1219+
List<Response<String>> results = new ArrayList<>();
1220+
for (int i = 0; i < cnt; i++) {
1221+
String key = prefix + i;
1222+
results.add(pipeline.get(key));
1223+
}
1224+
1225+
pipeline.sync();
1226+
int idx = 0;
1227+
for (Response<String> res : results) {
1228+
assertEquals(String.valueOf(idx), res.get());
1229+
idx++;
1230+
}
1231+
}
1232+
}
1233+
}
1234+
1235+
@Test
1236+
@Timeout(10)
1237+
public void sharedExecutorConcurrentPipelinesNoDeadlock() throws Exception {
1238+
// This test verifies that multiple pipelines using the same shared executor
1239+
// concurrently do not deadlock, even when the executor has fewer threads
1240+
// than concurrent pipelines
1241+
final int EXECUTOR_THREADS = 3;
1242+
final int CONCURRENT_PIPELINES = 10; // More pipelines than executor threads
1243+
final int OPERATIONS_PER_PIPELINE = 20;
1244+
1245+
ExecutorService sharedExecutor = Executors.newFixedThreadPool(EXECUTOR_THREADS);
1246+
ExecutorService testExecutor = Executors.newFixedThreadPool(CONCURRENT_PIPELINES);
1247+
1248+
try (RedisClusterClient cluster = RedisClusterClient.builder().nodes(nodes).clientConfig(DEFAULT_CLIENT_CONFIG).build()) {
1249+
1250+
CountDownLatch startLatch = new CountDownLatch(1);
1251+
CountDownLatch completionLatch = new CountDownLatch(CONCURRENT_PIPELINES);
1252+
List<Future<Integer>> futures = new ArrayList<>();
1253+
1254+
// Launch multiple pipelines concurrently
1255+
for (int pipelineId = 0; pipelineId < CONCURRENT_PIPELINES; pipelineId++) {
1256+
final int id = pipelineId;
1257+
Future<Integer> future = testExecutor.submit(() -> {
1258+
try {
1259+
// Wait for all threads to be ready before starting
1260+
startLatch.await();
1261+
1262+
int successCount = 0;
1263+
try (ClusterPipeline pipeline = cluster.pipelined(sharedExecutor)) {
1264+
List<Response<String>> responses = new ArrayList<>();
1265+
1266+
// Perform operations on different keys to ensure multi-node pipeline
1267+
for (int i = 0; i < OPERATIONS_PER_PIPELINE; i++) {
1268+
String key = "pipeline" + id + "_key" + i;
1269+
String value = "value" + i;
1270+
pipeline.set(key, value);
1271+
responses.add(pipeline.get(key));
1272+
}
1273+
1274+
// This sync() will use the shared executor
1275+
pipeline.sync();
1276+
1277+
// Verify all responses
1278+
for (int i = 0; i < OPERATIONS_PER_PIPELINE; i++) {
1279+
String expected = "value" + i;
1280+
String actual = responses.get(i).get();
1281+
if (expected.equals(actual)) {
1282+
successCount++;
1283+
}
1284+
}
1285+
}
1286+
return successCount;
1287+
} finally {
1288+
completionLatch.countDown();
1289+
}
1290+
});
1291+
futures.add(future);
1292+
}
1293+
1294+
// Release all threads to start concurrently
1295+
startLatch.countDown();
1296+
1297+
boolean completed = completionLatch.await(5, SECONDS);
1298+
assertTrue(completed, "All concurrent pipelines should complete without deadlock");
1299+
1300+
// Verify all operations succeeded
1301+
for (int i = 0; i < CONCURRENT_PIPELINES; i++) {
1302+
Integer successCount = futures.get(i).get();
1303+
assertEquals(OPERATIONS_PER_PIPELINE, successCount.intValue(),
1304+
"Pipeline " + i + " should have all operations succeed");
1305+
}
1306+
1307+
// Verify shared executor is still active
1308+
assertFalse(sharedExecutor.isShutdown(), "Shared executor should not be shut down");
1309+
1310+
} finally {
1311+
testExecutor.shutdown();
1312+
testExecutor.awaitTermination(5, SECONDS);
1313+
sharedExecutor.shutdown();
1314+
sharedExecutor.awaitTermination(5, SECONDS);
1315+
}
1316+
}
11741317
}

0 commit comments

Comments
 (0)