You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mb...@apache.org on 2026/08/19 18:00:16 UTC
(asterixdb) 06/12: [NO ISSUE][HYR][NET] Use executor service for async handshakes
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
commit 7e665ff1c9ffe4f4a16d8d3f965670ca13d7a0a4
Author: Michael Blow <mi...@couchbase.com>
AuthorDate: Sun Feb 18 22:31:52 2024 -0500
[NO ISSUE][HYR][NET] Use executor service for async handshakes
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
- asynchronous socket handshakes were submitted to the
ForkJoinPool.commonPool() default executor of
CompletableFuture.supplyAsync(). Handshakes block, so on a
cores-1-wide common pool they can starve, and compete with any other
common pool user in the JVM. Thread the owning service's executor
through to TCPEndpoint and IPCConnectionManager instead
- interface changes: the constructors of MuxDemux, TCPEndpoint,
IPCSystem, ResultSet, ResultDirectory, ClientNetworkManager,
NetworkManager, ResultNetworkManager, HyracksConnection and the two
hdfs Schedulers now take the executor to use, as their last argument
- HyracksConnection(ccHost, ccPort) has no executor to inherit, so it
creates one; it is daemon-threaded and shut down by close() so the
convenience constructor does not leak a pool per connection
- backport of 18168, which landed on master only. master is downstream
of morpheus, so the fix never reached trinity/morpheus/totoro and the
starvation kept hanging asterix-gerrit-ssl-compression there
- the sole conflict was in AsterixHyracksIntegrationUtil, where an
unrelated toArray cleanup on master met this change's added argument;
trinity's form of that sibling edit is kept
(cherry picked from commit b487f67a7034feb4c0da7bb77ee6148281946ad7)
Ext-ref: MB-73310
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: Ibe2fdce979a49d962517494dc77a5616a19f6abf
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21545
Reviewed-by: Ali Alsuliman <al...@gmail.com>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
---
.../asterix/app/cc/CcApplicationContext.java | 3 +-
.../apache/asterix/app/nc/NCAppRuntimeContext.java | 2 +-
.../apache/asterix/app/result/ResultReader.java | 2 +-
.../asterix/hyracks/bootstrap/CCApplication.java | 3 +-
.../api/common/AsterixHyracksIntegrationUtil.java | 2 +-
.../external/library/ExternalLibraryManager.java | 3 +-
.../apache/asterix/external/util/HDFSUtils.java | 2 +-
.../hyracks/client/net/ClientNetworkManager.java | 5 ++-
.../hyracks/client/result/ResultDirectory.java | 7 +--
.../apache/hyracks/client/result/ResultSet.java | 8 ++--
.../control/cc/ClusterControllerService.java | 5 ++-
.../hyracks/control/nc/NodeControllerService.java | 6 +--
.../control/nc/net/MessagingNetworkManager.java | 2 +-
.../hyracks/control/nc/net/NetworkManager.java | 6 ++-
.../control/nc/net/ResultNetworkManager.java | 23 +++++-----
.../tests/integration/AbstractIntegrationTest.java | 3 +-
.../AbstractMultiNCIntegrationTest.java | 4 +-
.../apache/hyracks/hdfs/scheduler/Scheduler.java | 7 ++-
.../apache/hyracks/hdfs2/scheduler/Scheduler.java | 7 ++-
.../apache/hyracks/ipc/impl/HyracksConnection.java | 52 ++++++++++++++++++----
.../hyracks/ipc/impl/IPCConnectionManager.java | 13 ++++--
.../org/apache/hyracks/ipc/impl/IPCSystem.java | 5 ++-
.../java/org/apache/hyracks/ipc/tests/IPCTest.java | 9 ++--
.../hyracks/net/protocols/muxdemux/MuxDemux.java | 24 +++++-----
.../hyracks/net/protocols/tcp/TCPEndpoint.java | 8 +++-
.../protocols/muxdemux/MuxDemuxReconnectTest.java | 18 +++++++-
.../java/org/apache/hyracks/net/tests/NetTest.java | 17 ++++++-
27 files changed, 166 insertions(+), 80 deletions(-)
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
index 880880e851..ade440ba14 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/CcApplicationContext.java
@@ -209,7 +209,8 @@ public class CcApplicationContext implements ICcApplicationContext {
NetworkUtil.closeQuietly(rs);
NetworkUtil.closeQuietly(hc);
- hcc = hc = new HyracksConnection(hcc.getHost(), hcc.getPort());
+ hcc = hc = new HyracksConnection(hcc.getHost(), hcc.getPort(),
+ ccServiceCtx.getControllerService().getExecutor());
} catch (Exception e) {
throw HyracksDataException.create(e);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 61656b7540..41608333aa 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -535,7 +535,7 @@ public class NCAppRuntimeContext implements INcApplicationContext {
ClusterControllerInfo ccInfo = ncSrv.getNodeParameters(primaryCcId).getClusterControllerInfo();
NetworkUtil.closeQuietly(hc);
hcc = hc = new HyracksConnection(ccInfo.getClientNetAddress(), ccInfo.getClientNetPort(),
- ncSrv.getNetworkSecurityManager().getSocketChannelFactory());
+ ncSrv.getNetworkSecurityManager().getSocketChannelFactory(), getThreadExecutor());
} catch (Exception e) {
throw HyracksDataException.create(e);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
index 56dfd5e0c2..42dd404762 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/result/ResultReader.java
@@ -65,6 +65,6 @@ public class ResultReader {
public static ResultSet createResultSet(IHyracksClientConnection hcc, IControllerService srv,
CompilerProperties compilerProperties) throws Exception {
return new ResultSet(hcc, srv.getNetworkSecurityManager().getSocketChannelFactory(),
- compilerProperties.getFrameSize(), ResultReader.NUM_READERS);
+ compilerProperties.getFrameSize(), ResultReader.NUM_READERS, srv.getExecutor());
}
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
index 31ec44cb58..184a6da0ad 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -151,7 +151,8 @@ public class CCApplication extends BaseCCApplication {
String strIP = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetAddress();
int port = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetPort();
hcc = new HyracksConnection(strIP, port,
- ccServiceCtx.getControllerService().getNetworkSecurityManager().getSocketChannelFactory());
+ ccServiceCtx.getControllerService().getNetworkSecurityManager().getSocketChannelFactory(),
+ ccServiceCtx.getControllerService().getExecutor());
MetadataBuiltinFunctions.init();
ReplicationProperties repProp =
new ReplicationProperties(PropertiesAccessor.getInstance(ccServiceCtx.getAppConfig()));
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
index 1e096ac089..f99d663128 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
@@ -182,7 +182,7 @@ public class AsterixHyracksIntegrationUtil {
// Wait until cluster becomes active
((ICcApplicationContext) cc.getApplicationContext()).getClusterStateManager().waitForState(ClusterState.ACTIVE);
hcc = new HyracksConnection(cc.getConfig().getClientListenAddress(), cc.getConfig().getClientListenPort(),
- cc.getNetworkSecurityManager().getSocketChannelFactory());
+ cc.getNetworkSecurityManager().getSocketChannelFactory(), cc.getExecutor());
this.ncs = nodeControllers.toArray(new NodeControllerService[nodeControllers.size()]);
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
index 050d9f0071..cff8232d74 100755
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/library/ExternalLibraryManager.java
@@ -163,7 +163,8 @@ public class ExternalLibraryManager implements ILibraryManager, ILifeCycleCompon
public void initialize(boolean resetStorageData) throws HyracksDataException {
try {
pythonIPC = new IPCSystem(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0),
- PlainSocketChannelFactory.INSTANCE, router, new ExternalFunctionResultRouter.NoOpNoSerJustDe());
+ PlainSocketChannelFactory.INSTANCE, router, new ExternalFunctionResultRouter.NoOpNoSerJustDe(),
+ ncs.getExecutor());
pythonIPC.start();
Path baseDirPath = baseDir.getFile().toPath();
if (Files.isDirectory(baseDirPath)) {
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index f4edafcb5f..25926c91e5 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -78,7 +78,7 @@ public class HDFSUtils {
try {
scheduler = new Scheduler(ccContext.getClusterControllerInfo().getClientNetAddress(),
ccContext.getClusterControllerInfo().getClientNetPort(),
- networkSecurityManager.getSocketChannelFactory());
+ networkSecurityManager.getSocketChannelFactory(), serviceCtx.getControllerService().getExecutor());
} catch (HyracksException e) {
throw new RuntimeDataException(ErrorCode.UTIL_HDFS_UTILS_CANNOT_OBTAIN_HDFS_SCHEDULER);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
index 5fe6ecbee6..94e83c5ee7 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/net/ClientNetworkManager.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.client.net;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.exceptions.NetException;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -36,12 +37,12 @@ public class ClientNetworkManager implements IChannelConnectionFactory {
private final MuxDemux md;
- public ClientNetworkManager(int nThreads, ISocketChannelFactory socketChannelFactory) {
+ public ClientNetworkManager(int nThreads, ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
/* This is a connect only socket and does not listen to any incoming connections, so pass null to
* localAddress and listener.
*/
md = new MuxDemux(null, null, nThreads, MAX_CONNECTION_ATTEMPTS, FullFrameChannelInterfaceFactory.INSTANCE,
- socketChannelFactory);
+ socketChannelFactory, executor);
}
public void start() throws IOException {
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
index 5dd5fff3be..7fe9016096 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultDirectory.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.client.result;
import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -40,11 +41,11 @@ public class ResultDirectory implements IResultDirectory, Closeable {
private final IPCSystem ipc;
private final IResultDirectory remoteResultDirectory;
- public ResultDirectory(String resultHost, int resultPort, ISocketChannelFactory socketChannelFactory)
- throws IOException, IPCException {
+ public ResultDirectory(String resultHost, int resultPort, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws IOException, IPCException {
RPCInterface rpci = new RPCInterface();
ipc = new IPCSystem(new InetSocketAddress(0), socketChannelFactory, rpci,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ new JavaSerializationBasedPayloadSerializerDeserializer(), executor);
ipc.start();
IIPCHandle ddsIpchandle = ipc.getReconnectingHandle(new InetSocketAddress(resultHost, resultPort));
this.remoteResultDirectory = new ResultDirectoryRemoteProxy(ddsIpchandle, rpci);
diff --git a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
index 8a88045fc3..474e0cc03a 100644
--- a/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
+++ b/hyracks-fullstack/hyracks/hyracks-client/src/main/java/org/apache/hyracks/client/result/ResultSet.java
@@ -20,6 +20,7 @@ package org.apache.hyracks.client.result;
import java.io.Closeable;
import java.io.IOException;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.comm.NetworkAddress;
@@ -43,11 +44,12 @@ public class ResultSet implements IResultSet, Closeable {
private final IHyracksCommonContext resultClientCtx;
public ResultSet(IHyracksClientConnection hcc, ISocketChannelFactory socketChannelFactory, int frameSize,
- int nReaders) throws Exception {
+ int nReaders, ExecutorService executor) throws Exception {
NetworkAddress ddsAddress = hcc.getResultDirectoryAddress();
- resultDirectory = new ResultDirectory(ddsAddress.getAddress(), ddsAddress.getPort(), socketChannelFactory);
+ resultDirectory =
+ new ResultDirectory(ddsAddress.getAddress(), ddsAddress.getPort(), socketChannelFactory, executor);
- netManager = new ClientNetworkManager(nReaders, socketChannelFactory);
+ netManager = new ClientNetworkManager(nReaders, socketChannelFactory, executor);
netManager.start();
resultClientCtx = new ResultClientContext(frameSize);
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
index a386ab8941..1dd54ad130 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-cc/src/main/java/org/apache/hyracks/control/cc/ClusterControllerService.java
@@ -221,12 +221,13 @@ public class ClusterControllerService implements IControllerService {
IIPCI ccIPCI = new ClusterControllerIPCI(this);
clusterIPC = new IPCSystem(
new InetSocketAddress(ccConfig.getClusterListenAddress(), ccConfig.getClusterListenPort()),
- networkSecurityManager.getSocketChannelFactory(), ccIPCI, new CCNCFunctions.SerializerDeserializer());
+ networkSecurityManager.getSocketChannelFactory(), ccIPCI, new CCNCFunctions.SerializerDeserializer(),
+ executor);
IIPCI ciIPCI = new ClientInterfaceIPCI(this, jobIdFactory);
clientIPC =
new IPCSystem(new InetSocketAddress(ccConfig.getClientListenAddress(), ccConfig.getClientListenPort()),
networkSecurityManager.getSocketChannelFactory(), ciIPCI,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ new JavaSerializationBasedPayloadSerializerDeserializer(), executor);
clusterIPC.start();
clientIPC.start();
startWebServers();
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index f3df6a8ee1..90d4255b1a 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -269,7 +269,7 @@ public class NodeControllerService implements IControllerService {
resultNetworkManager = new ResultNetworkManager(ncConfig.getResultListenAddress(),
ncConfig.getResultListenPort(), resultPartitionManager, ncConfig.getNetThreadCount(),
ncConfig.getNetBufferCount(), ncConfig.getResultPublicAddress(), ncConfig.getResultPublicPort(),
- FullFrameChannelInterfaceFactory.INSTANCE, networkSecurityManager.getSocketChannelFactory());
+ FullFrameChannelInterfaceFactory.INSTANCE, networkSecurityManager.getSocketChannelFactory(), executor);
if (ncConfig.getMessagingListenAddress() != null && serviceCtx.getMessagingChannelInterfaceFactory() != null) {
messagingNetManager = new MessagingNetworkManager(this, ncConfig.getMessagingListenAddress(),
ncConfig.getMessagingListenPort(), ncConfig.getNetThreadCount(),
@@ -283,13 +283,13 @@ public class NodeControllerService implements IControllerService {
LOGGER.log(Level.INFO, "Starting NodeControllerService");
ipc = new IPCSystem(new InetSocketAddress(ncConfig.getClusterListenAddress(), ncConfig.getClusterListenPort()),
networkSecurityManager.getSocketChannelFactory(), new NodeControllerIPCI(this),
- new CCNCFunctions.SerializerDeserializer());
+ new CCNCFunctions.SerializerDeserializer(), executor);
ipc.start();
partitionManager = new PartitionManager(this);
netManager = new NetworkManager(ncConfig.getDataListenAddress(), ncConfig.getDataListenPort(), partitionManager,
ncConfig.getNetThreadCount(), ncConfig.getNetBufferCount(), ncConfig.getDataPublicAddress(),
ncConfig.getDataPublicPort(), FullFrameChannelInterfaceFactory.INSTANCE,
- networkSecurityManager.getSocketChannelFactory());
+ networkSecurityManager.getSocketChannelFactory(), executor);
netManager.start();
startApplication();
init();
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
index 44f9f02e25..48bd3cb100 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/MessagingNetworkManager.java
@@ -58,7 +58,7 @@ public class MessagingNetworkManager {
ISocketChannelFactory socketChannelFactory) {
this.ncs = ncs;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, ncs.getExecutor());
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
index 468a96981d..35bbd27a80 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.comm.ICloseableBufferAcceptor;
@@ -63,11 +64,12 @@ public class NetworkManager implements IChannelConnectionFactory {
public NetworkManager(String inetAddress, int inetPort, PartitionManager partitionManager, int nThreads,
int nBuffers, String publicInetAddress, int publicInetPort,
- IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory) {
+ IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) {
this.partitionManager = partitionManager;
this.nBuffers = nBuffers;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, executor);
// Just save these values for the moment; may be reset in start()
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
index fe8f2afe07..c338a897e4 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/ResultNetworkManager.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.comm.ICloseableBufferAcceptor;
@@ -61,24 +62,22 @@ public class ResultNetworkManager implements IChannelConnectionFactory {
private NetworkAddress publicNetworkAddress;
/**
- * @param inetAddress
- * - Internet address to bind the listen port to
- * @param inetPort
- * - Port to bind on inetAddress
- * @param publicInetAddress
- * - Internet address to report to consumers;
- * useful when behind NAT. null = same as inetAddress
- * @param publicInetPort
- * - Port to report to consumers; useful when
- * behind NAT. Ignored if publicInetAddress is null. 0 = same as inetPort
+ * @param inetAddress - Internet address to bind the listen port to
+ * @param inetPort - Port to bind on inetAddress
+ * @param publicInetAddress - Internet address to report to consumers;
+ * useful when behind NAT. null = same as inetAddress
+ * @param publicInetPort - Port to report to consumers; useful when
+ * behind NAT. Ignored if publicInetAddress is null. 0 = same as inetPort
+ * @param executor - Executor on which asynchronous socket handshakes are performed
*/
public ResultNetworkManager(String inetAddress, int inetPort, IResultPartitionManager partitionManager,
int nThreads, int nBuffers, String publicInetAddress, int publicInetPort,
- IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory) {
+ IChannelInterfaceFactory channelInterfaceFactory, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) {
this.partitionManager = partitionManager;
this.nBuffers = nBuffers;
md = new MuxDemux(new InetSocketAddress(inetAddress, inetPort), new ChannelOpenListener(), nThreads,
- MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory);
+ MAX_CONNECTION_ATTEMPTS, channelInterfaceFactory, socketChannelFactory, executor);
// Just save these values for the moment; may be reset in start()
publicNetworkAddress = new NetworkAddress(publicInetAddress, publicInetPort);
}
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
index 6f59ed8c17..7c1d66b080 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractIntegrationTest.java
@@ -153,7 +153,8 @@ public abstract class AbstractIntegrationTest {
IFrameTupleAccessor frameTupleAccessor = new ResultFrameTupleAccessor();
- IResultSet resultSet = new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
+ IResultSet resultSet =
+ new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders, cc.getExecutor());
IResultSetReader reader = resultSet.createReader(jobId, resultSetId);
List<String> resultRecords = new ArrayList<>();
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
index 7a75a0fcca..58f1c88240 100644
--- a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/AbstractMultiNCIntegrationTest.java
@@ -163,8 +163,8 @@ public abstract class AbstractMultiNCIntegrationTest {
IFrameTupleAccessor frameTupleAccessor = new ResultFrameTupleAccessor();
if (!spec.getResultSetIds().isEmpty()) {
- IResultSet resultSet =
- new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders);
+ IResultSet resultSet = new ResultSet(hcc, PlainSocketChannelFactory.INSTANCE, spec.getFrameSize(), nReaders,
+ cc.getExecutor());
IResultSetReader reader = resultSet.createReader(jobId, spec.getResultSetIds().get(0));
ObjectMapper om = new ObjectMapper();
diff --git a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
index f5bf07b6b0..bba2ec5c27 100644
--- a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
+++ b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs/scheduler/Scheduler.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.Random;
+import java.util.concurrent.ExecutorService;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapred.InputSplit;
@@ -88,12 +89,14 @@ public class Scheduler {
* @param ipAddress IP address
* @param port Port
* @param channelFactory Channel Factory
+ * @param executor Executor on which asynchronous socket handshakes are performed
* @throws HyracksException
*/
- public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory) throws HyracksException {
+ public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory, ExecutorService executor)
+ throws HyracksException {
try {
- IHyracksClientConnection hcc = new HyracksConnection(ipAddress, port, channelFactory);
+ IHyracksClientConnection hcc = new HyracksConnection(ipAddress, port, channelFactory, executor);
this.ncNameToNcInfos = hcc.getNodeControllerInfos();
ClusterTopology topology = hcc.getClusterTopology();
this.ncCollectionBuilder = topology == null ? new IPProximityNcCollectionBuilder()
diff --git a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
index a26a5f7261..1386e0ed3e 100644
--- a/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
+++ b/hyracks-fullstack/hyracks/hyracks-hdfs/src/main/java/org/apache/hyracks/hdfs2/scheduler/Scheduler.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.hdfs2.scheduler;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hyracks.api.client.NodeControllerInfo;
@@ -43,10 +44,12 @@ public class Scheduler {
* The constructor of the scheduler
*
* @param ncNameToNcInfos
+ * @param executor Executor on which asynchronous socket handshakes are performed
* @throws HyracksException
*/
- public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory) throws HyracksException {
- scheduler = new org.apache.hyracks.hdfs.scheduler.Scheduler(ipAddress, port, channelFactory);
+ public Scheduler(String ipAddress, int port, ISocketChannelFactory channelFactory, ExecutorService executor)
+ throws HyracksException {
+ scheduler = new org.apache.hyracks.hdfs.scheduler.Scheduler(ipAddress, port, channelFactory, executor);
}
/**
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
index 1b6d8af81a..88b0dcb591 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/HyracksConnection.java
@@ -62,6 +62,7 @@ import org.apache.hyracks.ipc.api.RPCInterface;
import org.apache.hyracks.ipc.sockets.PlainSocketChannelFactory;
import org.apache.hyracks.util.ExitUtil;
import org.apache.hyracks.util.InterruptibleAction;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -95,23 +96,34 @@ public final class HyracksConnection implements Closeable, IHyracksClientConnect
private final BlockingQueue<UnInterruptibleRequest<?>> uninterruptibles = new ArrayBlockingQueue<>(1);
+ // non-null only when the handshake executor was created by this instance, and so must be shut down by close()
+ private final ExecutorService ownedExecutor;
+
/**
* Constructor to create a connection to the Hyracks Cluster Controller.
*
- * @param ccHost
- * Host name (or IP Address) where the Cluster Controller can be
- * reached.
- * @param ccPort
- * Port to reach the Hyracks Cluster Controller at the specified
- * host name.
+ * @param ccHost Host name (or IP Address) where the Cluster Controller can be
+ * reached.
+ * @param ccPort Port to reach the Hyracks Cluster Controller at the specified
+ * host name.
+ * @param socketChannelFactory The socket channel factory
+ * @param executor The executor on which asynchronous socket handshakes are performed; owned by the
+ * caller and not shut down by {@link #close()}
* @throws Exception
*/
- public HyracksConnection(String ccHost, int ccPort, ISocketChannelFactory socketChannelFactory) throws Exception {
+ public HyracksConnection(String ccHost, int ccPort, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws Exception {
+ this(ccHost, ccPort, socketChannelFactory, executor, false);
+ }
+
+ private HyracksConnection(String ccHost, int ccPort, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor, boolean ownsExecutor) throws Exception {
this.ccHost = ccHost;
this.ccPort = ccPort;
+ this.ownedExecutor = ownsExecutor ? executor : null;
RPCInterface rpci = new RPCInterface();
ipc = new IPCSystem(new InetSocketAddress(0), socketChannelFactory, rpci,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ new JavaSerializationBasedPayloadSerializerDeserializer(), executor);
ipc.start();
hci = new HyracksClientInterfaceRemoteProxy(ipc.getReconnectingHandle(new InetSocketAddress(ccHost, ccPort)),
rpci);
@@ -120,13 +132,35 @@ public final class HyracksConnection implements Closeable, IHyracksClientConnect
uninterruptibleExecutor.execute(new UninterrubtileHandlerWatcher());
}
+ /**
+ * Constructor to create a connection to the Hyracks Cluster Controller, using an internally created executor for
+ * asynchronous socket handshakes. The internal executor is shut down by {@link #close()}; prefer one of the
+ * constructors taking an {@link ExecutorService} where a suitable executor is already available.
+ */
public HyracksConnection(String ccHost, int ccPort) throws Exception {
- this(ccHost, ccPort, PlainSocketChannelFactory.INSTANCE);
+ this(ccHost, ccPort, PlainSocketChannelFactory.INSTANCE, newHandshakeExecutor(), true);
+ }
+
+ public HyracksConnection(String ccHost, int ccPort, ExecutorService executor) throws Exception {
+ this(ccHost, ccPort, PlainSocketChannelFactory.INSTANCE, executor);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "owned handshake executor")
+ private static ExecutorService newHandshakeExecutor() {
+ return Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "HyracksConnection Handshake thread");
+ t.setDaemon(true);
+ return t;
+ });
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "shut down owned executor")
public void close() {
ipc.stop();
+ if (ownedExecutor != null) {
+ ownedExecutor.shutdownNow();
+ }
}
@Override
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
index cdb91c3c71..c25cf02edb 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCConnectionManager.java
@@ -37,6 +37,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.hyracks.api.network.ISocketChannel;
@@ -77,11 +78,13 @@ public class IPCConnectionManager {
private volatile boolean stopped;
private final ISocketChannelFactory socketChannelFactory;
+ private final ExecutorService executor;
- IPCConnectionManager(IPCSystem system, InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory)
- throws IOException {
+ IPCConnectionManager(IPCSystem system, InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory,
+ ExecutorService executor) throws IOException {
this.system = system;
this.socketChannelFactory = socketChannelFactory;
+ this.executor = executor;
this.serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().setReuseAddress(true);
serverSocketChannel.configureBlocking(false);
@@ -514,8 +517,10 @@ public class IPCConnectionManager {
}
private void asyncHandshake(ISocketChannel socketChannel, IPCHandle handle, SelectionKey channelKey) {
- CompletableFuture.supplyAsync(socketChannel::handshake).exceptionally(ex -> false).thenAccept(
- handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, socketChannel, handle, channelKey));
+ CompletableFuture.supplyAsync(socketChannel::handshake, IPCConnectionManager.this.executor)
+ .exceptionally(ex -> false)
+ .thenAccept(handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, socketChannel, handle,
+ channelKey));
}
private void handleHandshakeCompletion(Boolean handshakeSuccess, ISocketChannel socketChannel, IPCHandle handle,
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
index 4a19a3348e..837d239f7c 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/main/java/org/apache/hyracks/ipc/impl/IPCSystem.java
@@ -20,6 +20,7 @@ package org.apache.hyracks.ipc.impl;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -42,8 +43,8 @@ public class IPCSystem {
private final IPCPerformanceCounters perfCounters;
public IPCSystem(InetSocketAddress socketAddress, ISocketChannelFactory socketChannelFactory, IIPCI ipci,
- IPayloadSerializerDeserializer serde) throws IOException {
- cMgr = new IPCConnectionManager(this, socketAddress, socketChannelFactory);
+ IPayloadSerializerDeserializer serde, ExecutorService executor) throws IOException {
+ cMgr = new IPCConnectionManager(this, socketAddress, socketChannelFactory, executor);
this.ipci = ipci;
this.serde = serde;
midFactory = new AtomicLong();
diff --git a/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java b/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
index 0d8d69b8e4..a2cecd728c 100644
--- a/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-ipc/src/test/java/org/apache/hyracks/ipc/tests/IPCTest.java
@@ -20,7 +20,7 @@ package org.apache.hyracks.ipc.tests;
import java.io.IOException;
import java.net.InetSocketAddress;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.hyracks.ipc.api.IIPCHandle;
@@ -59,7 +59,7 @@ public class IPCTest {
}
private IPCSystem createServerIPCSystem() throws IOException {
- final Executor executor = Executors.newCachedThreadPool();
+ final ExecutorService executor = Executors.newCachedThreadPool();
IIPCI ipci = new IIPCI() {
@Override
public void deliverIncomingMessage(IIPCHandle handle, long mid, long rmid, Object payload) {
@@ -82,11 +82,12 @@ public class IPCTest {
}
};
return new IPCSystem(new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, ipci,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ new JavaSerializationBasedPayloadSerializerDeserializer(), executor);
}
private IPCSystem createClientIPCSystem(RPCInterface rpci) throws IOException {
+ final ExecutorService executor = Executors.newCachedThreadPool();
return new IPCSystem(new InetSocketAddress("127.0.0.1", 0), PlainSocketChannelFactory.INSTANCE, rpci,
- new JavaSerializationBasedPayloadSerializerDeserializer());
+ new JavaSerializationBasedPayloadSerializerDeserializer(), executor);
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
index 1bbb333559..bffe3d6805 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.exceptions.NetException;
@@ -65,22 +66,17 @@ public class MuxDemux {
/**
* Constructor.
*
- * @param localAddress
- * - TCP/IP socket address to listen on. Null for non-listening unidirectional sockets
- * @param listener
- * - Callback interface to report channel events. Null for non-listening unidirectional sockets
- * @param nThreads
- * - Number of threads to use for data transfer
- * @param maxConnectionAttempts
- * - Maximum number of connection attempts
- * @param channelInterfaceFactory
- * - The channel interface factory
- * @param socketChannelFactory
- * - The socket channel factory
+ * @param localAddress - TCP/IP socket address to listen on. Null for non-listening unidirectional sockets
+ * @param listener - Callback interface to report channel events. Null for non-listening unidirectional sockets
+ * @param nThreads - Number of threads to use for data transfer
+ * @param maxConnectionAttempts - Maximum number of connection attempts
+ * @param channelInterfaceFactory - The channel interface factory
+ * @param socketChannelFactory - The socket channel factory
+ * @param executor - Executor on which asynchronous socket handshakes are performed
*/
public MuxDemux(InetSocketAddress localAddress, IChannelOpenListener listener, int nThreads,
int maxConnectionAttempts, IChannelInterfaceFactory channelInterfaceFactory,
- ISocketChannelFactory socketChannelFactory) {
+ ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
this.localAddress = localAddress;
this.channelOpenListener = listener;
this.maxConnectionAttempts = maxConnectionAttempts;
@@ -163,7 +159,7 @@ public class MuxDemux {
}
}
}
- }, nThreads, socketChannelFactory);
+ }, nThreads, socketChannelFactory, executor);
perfCounters = new MuxDemuxPerformanceCounters();
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
index d6ffe23e81..cdffe3e489 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/tcp/TCPEndpoint.java
@@ -34,6 +34,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
import org.apache.hyracks.api.network.ISocketChannel;
import org.apache.hyracks.api.network.ISocketChannelFactory;
@@ -58,12 +59,14 @@ public class TCPEndpoint {
private int nextThread;
private final ISocketChannelFactory socketChannelFactory;
+ private final ExecutorService executor;
public TCPEndpoint(ITCPConnectionListener connectionListener, int nThreads,
- ISocketChannelFactory socketChannelFactory) {
+ ISocketChannelFactory socketChannelFactory, ExecutorService executor) {
this.connectionListener = connectionListener;
this.nThreads = nThreads;
this.socketChannelFactory = socketChannelFactory;
+ this.executor = executor;
}
public void start(InetSocketAddress localAddress) throws IOException {
@@ -318,7 +321,8 @@ public class TCPEndpoint {
}
private void asyncHandshake(PendingHandshakeConnection connection) {
- CompletableFuture.supplyAsync(connection.socketChannel::handshake).exceptionally(ex -> false)
+ CompletableFuture.supplyAsync(connection.socketChannel::handshake, TCPEndpoint.this.executor)
+ .exceptionally(ex -> false)
.thenAccept(handshakeSuccess -> handleHandshakeCompletion(handshakeSuccess, connection));
}
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemuxReconnectTest.java b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemuxReconnectTest.java
index b4ed7f7c89..641c27b53f 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemuxReconnectTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemuxReconnectTest.java
@@ -22,13 +22,27 @@ import java.io.IOException;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import org.apache.hyracks.ipc.sockets.PlainSocketChannelFactory;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
public class MuxDemuxReconnectTest {
+ private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "MuxDemuxReconnectTest Handshake thread");
+ t.setDaemon(true);
+ return t;
+ });
+
+ @After
+ public void shutdownExecutor() {
+ executor.shutdownNow();
+ }
+
@Test
public void testReconnectReplacesFailedCachedConnection() throws Exception {
MuxDemux server = createMuxDemux();
@@ -47,10 +61,10 @@ public class MuxDemuxReconnectTest {
Assert.assertSame(secondConnection, getOutgoingConnection(client, serverAddress));
}
- private static MuxDemux createMuxDemux() {
+ private MuxDemux createMuxDemux() {
return new MuxDemux(new InetSocketAddress("127.0.0.1", 0), channel -> {
// This test only exercises connection caching and reconnect behavior.
- }, 1, 5, FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE);
+ }, 1, 5, FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE, executor);
}
@SuppressWarnings("unchecked")
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
index ec1d21cc36..e30f99bab7 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/test/java/org/apache/hyracks/net/tests/NetTest.java
@@ -22,6 +22,8 @@ import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -33,10 +35,23 @@ import org.apache.hyracks.net.protocols.muxdemux.FullFrameChannelInterfaceFactor
import org.apache.hyracks.net.protocols.muxdemux.IChannelOpenListener;
import org.apache.hyracks.net.protocols.muxdemux.MultiplexedConnection;
import org.apache.hyracks.net.protocols.muxdemux.MuxDemux;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
public class NetTest {
+
+ private final ExecutorService executor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "NetTest Handshake thread");
+ t.setDaemon(true);
+ return t;
+ });
+
+ @After
+ public void shutdownExecutor() {
+ executor.shutdownNow();
+ }
+
@Test
public void test() throws Exception {
AtomicBoolean failFlag = new AtomicBoolean();
@@ -163,7 +178,7 @@ public class NetTest {
}
};
return new MuxDemux(new InetSocketAddress("127.0.0.1", 0), md1OpenListener, 1, 5,
- FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE);
+ FullFrameChannelInterfaceFactory.INSTANCE, PlainSocketChannelFactory.INSTANCE, executor);
}
private class ChannelIO {