diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java index 106d77c21be..fe373c8d22a 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java @@ -44,6 +44,7 @@ import org.apache.celeborn.client.read.MetricsCallback; import org.apache.celeborn.common.CelebornConf; import org.apache.celeborn.common.exception.CelebornIOException; +import org.apache.celeborn.common.exception.CelebornRuntimeException; import org.apache.celeborn.common.identity.UserIdentifier; import org.apache.celeborn.common.metrics.source.Role; import org.apache.celeborn.common.network.TransportContext; @@ -83,6 +84,8 @@ public class ShuffleClientImpl extends ShuffleClient { private final int registerShuffleMaxRetries; private final long registerShuffleRetryWaitMs; + private final int rpcMaxRetries; + private final long rpcRetryWait; private final int maxReviveTimes; private final boolean testRetryRevive; private final int pushBufferMaxSize; @@ -181,6 +184,8 @@ public ShuffleClientImpl(String appUniqueId, CelebornConf conf, UserIdentifier u this.userIdentifier = userIdentifier; registerShuffleMaxRetries = conf.clientRegisterShuffleMaxRetry(); registerShuffleRetryWaitMs = conf.clientRegisterShuffleRetryWaitMs(); + rpcMaxRetries = conf.clientRpcMaxRetries(); + rpcRetryWait = conf.clientRpcRetryWait(); maxReviveTimes = conf.clientPushMaxReviveTimes(); testRetryRevive = conf.testRetryRevive(); pushBufferMaxSize = conf.clientPushBufferMaxSize(); @@ -537,6 +542,8 @@ private ConcurrentHashMap registerShuffle( lifecycleManagerRef.askSync( RegisterShuffle$.MODULE$.apply(shuffleId, numMappers, numPartitions), conf.clientRpcRegisterShuffleAskTimeout(), + rpcMaxRetries, + rpcRetryWait, ClassTag$.MODULE$.apply(PbRegisterShuffleResponse.class))); } @@ -1711,6 +1718,8 @@ private void mapEndInternal( MapperEndResponse response = lifecycleManagerRef.askSync( new MapperEnd(shuffleId, mapId, attemptId, numMappers, partitionId), + rpcMaxRetries, + rpcRetryWait, ClassTag$.MODULE$.apply(MapperEndResponse.class)); if (response.status() != StatusCode.SUCCESS) { throw new CelebornIOException("MapperEnd failed! StatusCode: " + response.status()); @@ -1745,69 +1754,65 @@ public boolean cleanupShuffle(int shuffleId) { protected Tuple3 loadFileGroupInternal( int shuffleId, boolean isSegmentGranularityVisible) { - { - long getReducerFileGroupStartTime = System.nanoTime(); - String exceptionMsg = null; - Exception exception = null; - try { - if (lifecycleManagerRef == null) { - exceptionMsg = "Driver endpoint is null!"; - logger.warn(exceptionMsg); - } else { - GetReducerFileGroup getReducerFileGroup = - new GetReducerFileGroup(shuffleId, isSegmentGranularityVisible); - - GetReducerFileGroupResponse response = - lifecycleManagerRef.askSync( - getReducerFileGroup, - conf.clientRpcGetReducerFileGroupAskTimeout(), - ClassTag$.MODULE$.apply(GetReducerFileGroupResponse.class)); + long getReducerFileGroupStartTime = System.nanoTime(); + String exceptionMsg = null; + Exception exception = null; + if (lifecycleManagerRef == null) { + exceptionMsg = "Driver endpoint is null!"; + logger.warn(exceptionMsg); + return Tuple3.apply(null, exceptionMsg, exception); + } + try { + GetReducerFileGroup getReducerFileGroup = + new GetReducerFileGroup(shuffleId, isSegmentGranularityVisible); - switch (response.status()) { - case SUCCESS: - logger.info( - "Shuffle {} request reducer file group success using {} ms, result partition size {}.", - shuffleId, - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - getReducerFileGroupStartTime), - response.fileGroup().size()); - return Tuple3.apply( - new ReduceFileGroups( - response.fileGroup(), response.attempts(), response.partitionIds()), - null, - null); - case SHUFFLE_NOT_REGISTERED: - logger.warn( - "Request {} return {} for {}.", - getReducerFileGroup, - response.status(), - shuffleId); - // return empty result - return Tuple3.apply( - new ReduceFileGroups( - response.fileGroup(), response.attempts(), response.partitionIds()), - null, - null); - case STAGE_END_TIME_OUT: - case SHUFFLE_DATA_LOST: - exceptionMsg = - String.format( - "Request %s return %s for %s.", - getReducerFileGroup, response.status(), shuffleId); - logger.warn(exceptionMsg); - break; - default: // fall out - } - } - } catch (Exception e) { - if (e instanceof InterruptedException) { - Thread.currentThread().interrupt(); - } - logger.error("Exception raised while call GetReducerFileGroup for {}.", shuffleId, e); - exceptionMsg = e.getMessage(); - exception = e; + GetReducerFileGroupResponse response = + lifecycleManagerRef.askSync( + getReducerFileGroup, + conf.clientRpcGetReducerFileGroupAskTimeout(), + rpcMaxRetries, + rpcRetryWait, + ClassTag$.MODULE$.apply(GetReducerFileGroupResponse.class)); + switch (response.status()) { + case SUCCESS: + logger.info( + "Shuffle {} request reducer file group success using {} ms, result partition size {}.", + shuffleId, + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - getReducerFileGroupStartTime), + response.fileGroup().size()); + return Tuple3.apply( + new ReduceFileGroups( + response.fileGroup(), response.attempts(), response.partitionIds()), + null, + null); + case SHUFFLE_NOT_REGISTERED: + logger.warn( + "Request {} return {} for {}.", getReducerFileGroup, response.status(), shuffleId); + // return empty result + return Tuple3.apply( + new ReduceFileGroups( + response.fileGroup(), response.attempts(), response.partitionIds()), + null, + null); + case STAGE_END_TIME_OUT: + case SHUFFLE_DATA_LOST: + exceptionMsg = + String.format( + "Request %s return %s for %s.", + getReducerFileGroup, response.status(), shuffleId); + logger.warn(exceptionMsg); + break; + default: // fall out } - return Tuple3.apply(null, exceptionMsg, exception); + } catch (Exception e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + logger.error("Exception raised while call GetReducerFileGroup for {}.", shuffleId, e); + exceptionMsg = e.getMessage(); + exception = e; } + return Tuple3.apply(null, exceptionMsg, exception); } @Override @@ -1939,8 +1944,17 @@ public void shutdown() { @Override public void setupLifecycleManagerRef(String host, int port) { logger.info("setupLifecycleManagerRef: host = {}, port = {}", host, port); - lifecycleManagerRef = - rpcEnv.setupEndpointRef(new RpcAddress(host, port), RpcNameConstants.LIFECYCLE_MANAGER_EP); + try { + lifecycleManagerRef = + rpcEnv.setupEndpointRef( + new RpcAddress(host, port), + RpcNameConstants.LIFECYCLE_MANAGER_EP, + rpcMaxRetries, + rpcRetryWait); + } catch (Exception e) { + throw new CelebornRuntimeException("setupLifecycleManagerRef failed!", e); + } + initDataClientFactoryIfNeeded(); } diff --git a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java index ba0a7c39b9c..886aff3dc8e 100644 --- a/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java +++ b/client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java @@ -258,6 +258,12 @@ private CelebornConf setupEnv( RegisterShuffleResponse$.MODULE$.apply( statusCode, new PartitionLocation[] {primaryLocation})); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + t -> + RegisterShuffleResponse$.MODULE$.apply( + statusCode, new PartitionLocation[] {primaryLocation})); + shuffleClient.setupLifecycleManagerRef(endpointRef); ChannelFuture mockedFuture = @@ -420,6 +426,14 @@ public void testUpdateReducerFileGroupInterrupted() throws InterruptedException StatusCode.SUCCESS, locations, new int[0], Collections.emptySet()); }); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + t -> { + Thread.sleep(60 * 1000); + return GetReducerFileGroupResponse$.MODULE$.apply( + StatusCode.SUCCESS, locations, new int[0], Collections.emptySet()); + }); + shuffleClient = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); shuffleClient.setupLifecycleManagerRef(endpointRef); @@ -459,6 +473,13 @@ public void testUpdateReducerFileGroupNonFetchFailureExceptions() { StatusCode.SHUFFLE_NOT_REGISTERED, locations, new int[0], Collections.emptySet()); }); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + t -> { + return GetReducerFileGroupResponse$.MODULE$.apply( + StatusCode.SHUFFLE_NOT_REGISTERED, locations, new int[0], Collections.emptySet()); + }); + shuffleClient = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); shuffleClient.setupLifecycleManagerRef(endpointRef); @@ -476,6 +497,13 @@ public void testUpdateReducerFileGroupNonFetchFailureExceptions() { StatusCode.STAGE_END_TIME_OUT, locations, new int[0], Collections.emptySet()); }); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + t -> { + return GetReducerFileGroupResponse$.MODULE$.apply( + StatusCode.STAGE_END_TIME_OUT, locations, new int[0], Collections.emptySet()); + }); + shuffleClient = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); shuffleClient.setupLifecycleManagerRef(endpointRef); @@ -493,6 +521,13 @@ public void testUpdateReducerFileGroupNonFetchFailureExceptions() { StatusCode.SHUFFLE_DATA_LOST, locations, new int[0], Collections.emptySet()); }); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + t -> { + return GetReducerFileGroupResponse$.MODULE$.apply( + StatusCode.SHUFFLE_DATA_LOST, locations, new int[0], Collections.emptySet()); + }); + shuffleClient = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); shuffleClient.setupLifecycleManagerRef(endpointRef); @@ -515,6 +550,12 @@ public void testUpdateReducerFileGroupTimeout() throws InterruptedException { throw new RpcTimeoutException( "Rpc timeout", new TimeoutException("ask sync timeout")); }); + when(endpointRef.askSync(any(), any(), any(Integer.class), any(Long.class), any())) + .thenAnswer( + invocation -> { + throw new RpcTimeoutException( + "Rpc timeout", new TimeoutException("ask sync timeout")); + }); shuffleClient = new ShuffleClientImpl(TEST_APPLICATION_ID, conf, new UserIdentifier("mock", "mock")); diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala index 2359c57a240..ad9b3538169 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -520,6 +520,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se new RpcTimeout(get(RPC_LOOKUP_TIMEOUT).milli, RPC_LOOKUP_TIMEOUT.key) def rpcAskTimeout: RpcTimeout = new RpcTimeout(get(RPC_ASK_TIMEOUT).milli, RPC_ASK_TIMEOUT.key) + def rpcRetryWait: Long = get(RPC_RETRY_WAIT) def rpcInMemoryBoundedInboxCapacity(): Int = { get(RPC_INBOX_CAPACITY) } @@ -1017,6 +1018,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se def clientRpcCacheExpireTime: Long = get(CLIENT_RPC_CACHE_EXPIRE_TIME) def clientRpcSharedThreads: Int = get(CLIENT_RPC_SHARED_THREADS) def clientRpcMaxRetries: Int = get(CLIENT_RPC_MAX_RETIRES) + def clientRpcRetryWait: Long = get(CLIENT_RPC_RETRY_WAIT) def pushDataTimeoutMs: Long = get(CLIENT_PUSH_DATA_TIMEOUT) def clientPushLimitStrategy: String = get(CLIENT_PUSH_LIMIT_STRATEGY) def clientPushSlowStartInitialSleepTime: Long = get(CLIENT_PUSH_SLOW_START_INITIAL_SLEEP_TIME) @@ -1887,6 +1889,14 @@ object CelebornConf extends Logging { .timeConf(TimeUnit.MILLISECONDS) .createWithDefaultString("60s") + val RPC_RETRY_WAIT: ConfigEntry[Long] = + buildConf("celeborn.rpc.retryWait") + .categories("network") + .version("0.6.0") + .doc("Time to wait before next retry on RpcTimeoutException.") + .timeConf(TimeUnit.MILLISECONDS) + .createWithDefaultString("1s") + val RPC_DISPATCHER_THREADS: ConfigEntry[Int] = buildConf("celeborn.rpc.dispatcher.threads") .withAlternative("celeborn.rpc.dispatcher.numThreads") @@ -4938,6 +4948,14 @@ object CelebornConf extends Logging { .timeConf(TimeUnit.MILLISECONDS) .createWithDefaultString("3s") + val CLIENT_RPC_RETRY_WAIT: ConfigEntry[Long] = + buildConf("celeborn.client.rpc.retryWait") + .categories("client") + .version("0.6.0") + .doc("Client-specified time to wait before next retry on RpcTimeoutException.") + .timeConf(TimeUnit.MILLISECONDS) + .createWithDefaultString("1s") + val CLIENT_RESERVE_SLOTS_MAX_RETRIES: ConfigEntry[Int] = buildConf("celeborn.client.reserveSlots.maxRetries") .withAlternative("celeborn.slots.reserve.maxRetries") @@ -5087,7 +5105,7 @@ object CelebornConf extends Logging { buildConf("celeborn.client.rpc.maxRetries") .categories("client") .version("0.3.2") - .doc("Max RPC retry times in LifecycleManager.") + .doc("Max RPC retry times in client.") .intConf .createWithDefault(3) diff --git a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala index edd7005e2e9..8c861cf57ab 100644 --- a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala +++ b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala @@ -17,6 +17,9 @@ package org.apache.celeborn.common.rpc +import java.util.Random +import java.util.concurrent.TimeUnit + import scala.concurrent.Future import scala.reflect.ClassTag @@ -30,6 +33,7 @@ abstract class RpcEndpointRef(conf: CelebornConf) extends Serializable with Logging { private[this] val defaultAskTimeout = conf.rpcAskTimeout + private[this] val defaultRetryWait = conf.rpcRetryWait /** * return the address for the [[RpcEndpointRef]] @@ -88,4 +92,67 @@ abstract class RpcEndpointRef(conf: CelebornConf) val future = ask[T](message, timeout) timeout.awaitResult(future, address) } + + /** + * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]] and get its result within a + * default timeout, retry if timeout, throw an exception if this still fails. + * + * Note: this is a blocking action which may cost a lot of time, so don't call it in a message + * loop of [[RpcEndpoint]]. + * + * @param message the message to send + * @param retryCount the number of retries for the timeout + * @param retryWait the waiting time for a retry + * @tparam T type of the reply message + * @return the reply message from the corresponding [[RpcEndpoint]] + */ + def askSync[T: ClassTag](message: Any, retryCount: Int, retryWait: Long = defaultRetryWait): T = + askSync(message, defaultAskTimeout, retryCount, retryWait) + + /** + * Send a message to the corresponding [[RpcEndpoint.receiveAndReply]] and get its result within a + * specified timeout, retry if timeout, throw an exception if this still fails. + * + * Note: this is a blocking action which may cost a lot of time, so don't call it in a message + * loop of [[RpcEndpoint]]. + * + * @param message the message to send + * @param timeout the timeout duration + * @param retryCount the number of retries for the timeout + * @param retryWait the waiting time for a retry + * @tparam T type of the reply message + * @return the reply message from the corresponding [[RpcEndpoint]] + */ + def askSync[T: ClassTag]( + message: Any, + timeout: RpcTimeout, + retryCount: Int, + retryWait: Long): T = { + var numRetries = retryCount + while (numRetries > 0) { + numRetries -= 1 + try { + val future = ask[T](message, timeout) + return timeout.awaitResult(future, address) + } catch { + case e: RpcTimeoutException => + if (numRetries > 0) { + val random = new Random + val retryWaitMs = random.nextInt(retryWait.toInt) + try { + TimeUnit.MILLISECONDS.sleep(retryWaitMs) + } catch { + case _: InterruptedException => + throw e + } + } else { + throw e + } + case e: Exception => + throw e + } + } + // should never be here + null.asInstanceOf[T] + } } diff --git a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEnv.scala b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEnv.scala index 89973a936be..19f522a0afe 100644 --- a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEnv.scala +++ b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEnv.scala @@ -18,6 +18,8 @@ package org.apache.celeborn.common.rpc import java.io.File +import java.util.Random +import java.util.concurrent.TimeUnit import scala.concurrent.Future @@ -109,6 +111,7 @@ object RpcEnv { abstract class RpcEnv(config: RpcEnvConfig) { private[celeborn] val defaultLookupTimeout = config.conf.rpcLookupTimeout + private[celeborn] val defaultRetryWait = config.conf.rpcRetryWait /** * Return RpcEndpointRef of the registered [[RpcEndpoint]]. Will be used to implement @@ -147,6 +150,42 @@ abstract class RpcEnv(config: RpcEnvConfig) { setupEndpointRefByAddr(RpcEndpointAddress(address, endpointName)) } + /** + * Retrieve the [[RpcEndpointRef]] represented by `address` and `endpointName` with timeout retry. + * This is a blocking action. + */ + def setupEndpointRef( + address: RpcAddress, + endpointName: String, + retryCount: Int, + retryWait: Long = defaultRetryWait): RpcEndpointRef = { + var numRetries = retryCount + while (numRetries > 0) { + numRetries -= 1 + try { + return setupEndpointRefByAddr(RpcEndpointAddress(address, endpointName)) + } catch { + case e: RpcTimeoutException => + if (numRetries > 0) { + val random = new Random + val retryWaitMs = random.nextInt(retryWait.toInt) + try { + TimeUnit.MILLISECONDS.sleep(retryWaitMs) + } catch { + case _: InterruptedException => + throw e + } + } else { + throw e + } + case e: Exception => + throw e + } + } + // should never be here + null + } + /** * Stop [[RpcEndpoint]] specified by `endpoint`. */ diff --git a/docs/configuration/client.md b/docs/configuration/client.md index fd28441e34f..c576e0c4d27 100644 --- a/docs/configuration/client.md +++ b/docs/configuration/client.md @@ -80,10 +80,11 @@ license: | | celeborn.client.rpc.cache.size | 256 | false | The max cache items count for rpc cache. | 0.3.0 | celeborn.rpc.cache.size | | celeborn.client.rpc.commitFiles.askTimeout | <value of celeborn.rpc.askTimeout> | false | Timeout for CommitHandler commit files. | 0.4.1 | | | celeborn.client.rpc.getReducerFileGroup.askTimeout | <value of celeborn.rpc.askTimeout> | false | Timeout for ask operations during getting reducer file group information. During this process, there are `celeborn.client.requestCommitFiles.maxRetries` times for retry opportunities for committing files and 1 times for releasing slots request. User can customize this value according to your setting. | 0.2.0 | | -| celeborn.client.rpc.maxRetries | 3 | false | Max RPC retry times in LifecycleManager. | 0.3.2 | | +| celeborn.client.rpc.maxRetries | 3 | false | Max RPC retry times in client. | 0.3.2 | | | celeborn.client.rpc.registerShuffle.askTimeout | <value of celeborn.rpc.askTimeout> | false | Timeout for ask operations during register shuffle. During this process, there are two times for retry opportunities for requesting slots, one request for establishing a connection with Worker and `celeborn.client.reserveSlots.maxRetries` times for retry opportunities for reserving slots. User can customize this value according to your setting. | 0.3.0 | celeborn.rpc.registerShuffle.askTimeout | | celeborn.client.rpc.requestPartition.askTimeout | <value of celeborn.rpc.askTimeout> | false | Timeout for ask operations during requesting change partition location, such as reviving or splitting partition. During this process, there are `celeborn.client.reserveSlots.maxRetries` times for retry opportunities for reserving slots. User can customize this value according to your setting. | 0.2.0 | | | celeborn.client.rpc.reserveSlots.askTimeout | <value of celeborn.rpc.askTimeout> | false | Timeout for LifecycleManager request reserve slots. | 0.3.0 | | +| celeborn.client.rpc.retryWait | 1s | false | Client-specified time to wait before next retry on RpcTimeoutException. | 0.6.0 | | | celeborn.client.rpc.shared.threads | 16 | false | Number of shared rpc threads in LifecycleManager. | 0.3.2 | | | celeborn.client.shuffle.batchHandleChangePartition.interval | 100ms | false | Interval for LifecycleManager to schedule handling change partition requests in batch. | 0.3.0 | celeborn.shuffle.batchHandleChangePartition.interval | | celeborn.client.shuffle.batchHandleChangePartition.partitionBuckets | 256 | false | Max number of change partition requests which can be concurrently processed. | 0.5.0 | | diff --git a/docs/configuration/network.md b/docs/configuration/network.md index c958451ba05..50b56c17637 100644 --- a/docs/configuration/network.md +++ b/docs/configuration/network.md @@ -56,6 +56,7 @@ license: | | celeborn.rpc.inbox.capacity | 0 | false | Specifies size of the in memory bounded capacity. | 0.5.0 | | | celeborn.rpc.io.threads | <undefined> | false | Netty IO thread number of NettyRpcEnv to handle RPC request. The default threads number is the number of runtime available processors. | 0.2.0 | | | celeborn.rpc.lookupTimeout | 30s | false | Timeout for RPC lookup operations. | 0.2.0 | | +| celeborn.rpc.retryWait | 1s | false | Time to wait before next retry on RpcTimeoutException. | 0.6.0 | | | celeborn.rpc.slow.interval | <undefined> | false | min interval (ms) for RPC framework to log slow RPC | 0.6.0 | | | celeborn.rpc.slow.threshold | 1s | false | threshold for RPC framework to log slow RPC | 0.6.0 | | | celeborn.shuffle.io.maxChunksBeingTransferred | <undefined> | false | The max number of chunks allowed to be transferred at the same time on shuffle service. Note that new incoming connections will be closed when the max number is hit. The client will retry according to the shuffle retry configs (see `celeborn..io.maxRetries` and `celeborn..io.retryWait`), if those limits are reached the task will fail with fetch failure. | 0.2.0 | |