You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ma...@apache.org on 2018/11/09 19:42:48 UTC
git commit: updated refs/heads/trunk to d412219
Repository: giraph
Updated Branches:
refs/heads/trunk f69f77d20 -> d41221966
JIRA-1211
closes #93
Project: http://git-wip-us.apache.org/repos/asf/giraph/repo
Commit: http://git-wip-us.apache.org/repos/asf/giraph/commit/d4122196
Tree: http://git-wip-us.apache.org/repos/asf/giraph/tree/d4122196
Diff: http://git-wip-us.apache.org/repos/asf/giraph/diff/d4122196
Branch: refs/heads/trunk
Commit: d41221966cece5ed8c029a7e941b5d621486ddb1
Parents: f69f77d
Author: Maja Kabiljo <ma...@fb.com>
Authored: Fri Nov 9 11:42:35 2018 -0800
Committer: Maja Kabiljo <ma...@fb.com>
Committed: Fri Nov 9 11:42:35 2018 -0800
----------------------------------------------------------------------
.../apache/giraph/comm/netty/NettyClient.java | 23 ++++++++++++++++----
.../org/apache/giraph/conf/GiraphConstants.java | 9 ++++++++
2 files changed, 28 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/giraph/blob/d4122196/giraph-core/src/main/java/org/apache/giraph/comm/netty/NettyClient.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/comm/netty/NettyClient.java b/giraph-core/src/main/java/org/apache/giraph/comm/netty/NettyClient.java
index 7e9fa87..74011b9 100644
--- a/giraph-core/src/main/java/org/apache/giraph/comm/netty/NettyClient.java
+++ b/giraph-core/src/main/java/org/apache/giraph/comm/netty/NettyClient.java
@@ -94,6 +94,7 @@ import static org.apache.giraph.conf.GiraphConstants.NETTY_CLIENT_EXECUTION_AFTE
import static org.apache.giraph.conf.GiraphConstants.NETTY_CLIENT_EXECUTION_THREADS;
import static org.apache.giraph.conf.GiraphConstants.NETTY_CLIENT_USE_EXECUTION_HANDLER;
import static org.apache.giraph.conf.GiraphConstants.NETTY_MAX_CONNECTION_FAILURES;
+import static org.apache.giraph.conf.GiraphConstants.RESEND_TIMED_OUT_REQUESTS;
import static org.apache.giraph.conf.GiraphConstants.WAIT_TIME_BETWEEN_CONNECTION_RETRIES_MS;
import static org.apache.giraph.conf.GiraphConstants.WAITING_REQUEST_MSECS;
@@ -188,6 +189,11 @@ public class NettyClient {
private final long waitTimeBetweenConnectionRetriesMs;
/** Maximum number of milliseconds for a request */
private final int maxRequestMilliseconds;
+ /**
+ * Whether to resend request which timed out or fail the job if timeout
+ * happens
+ */
+ private final boolean resendTimedOutRequests;
/** Waiting interval for checking outstanding requests msecs */
private final int waitingRequestMsecs;
/** Timed logger for printing request debugging */
@@ -278,6 +284,7 @@ public class NettyClient {
NETWORK_REQUESTS_RESENT_FOR_CONNECTION_FAILURE_NAME));
maxRequestMilliseconds = MAX_REQUEST_MILLISECONDS.get(conf);
+ resendTimedOutRequests = RESEND_TIMED_OUT_REQUESTS.get(conf);
maxConnectionFailures = NETTY_MAX_CONNECTION_FAILURES.get(conf);
waitTimeBetweenConnectionRetriesMs =
WAIT_TIME_BETWEEN_CONNECTION_RETRIES_MS.get(conf);
@@ -997,7 +1004,7 @@ public class NettyClient {
// If the request is taking too long, re-establish and resend
return requestInfo.getElapsedMsecs() > maxRequestMilliseconds;
}
- }, networkRequestsResentForTimeout);
+ }, networkRequestsResentForTimeout, resendTimedOutRequests);
resendRequestsWhenNeeded(new Predicate<RequestInfo>() {
@Override
public boolean apply(RequestInfo requestInfo) {
@@ -1006,7 +1013,7 @@ public class NettyClient {
return writeFuture != null && (!writeFuture.channel().isActive() ||
(writeFuture.isDone() && !writeFuture.isSuccess()));
}
- }, networkRequestsResentForConnectionFailure);
+ }, networkRequestsResentForConnectionFailure, true);
}
/**
@@ -1014,10 +1021,13 @@ public class NettyClient {
* @param shouldResendRequestPredicate Predicate to use to check whether
* request should be resent
* @param counter Counter to increment for every resent network request
+ * @param resendProblematicRequest Whether to resend problematic request or
+ * fail the job if such request is found
*/
private void resendRequestsWhenNeeded(
Predicate<RequestInfo> shouldResendRequestPredicate,
- GiraphHadoopCounter counter) {
+ GiraphHadoopCounter counter,
+ boolean resendProblematicRequest) {
// Check if there are open requests which have been sent a long time ago,
// and if so, resend them.
List<ClientRequestId> addedRequestIds = Lists.newArrayList();
@@ -1028,6 +1038,11 @@ public class NettyClient {
RequestInfo requestInfo = entry.getValue();
// If request should be resent
if (shouldResendRequestPredicate.apply(requestInfo)) {
+ if (!resendProblematicRequest) {
+ throw new IllegalStateException("Problem with request id " +
+ entry.getKey() + " for " + requestInfo.getDestinationAddress() +
+ ", failing the job");
+ }
ChannelFuture writeFuture = requestInfo.getWriteFuture();
String logMessage;
if (writeFuture == null) {
@@ -1135,7 +1150,7 @@ public class NettyClient {
return requestInfo.getDestinationAddress().equals(
channel.remoteAddress());
}
- }, networkRequestsResentForChannelFailure);
+ }, networkRequestsResentForChannelFailure, true);
}
/**
http://git-wip-us.apache.org/repos/asf/giraph/blob/d4122196/giraph-core/src/main/java/org/apache/giraph/conf/GiraphConstants.java
----------------------------------------------------------------------
diff --git a/giraph-core/src/main/java/org/apache/giraph/conf/GiraphConstants.java b/giraph-core/src/main/java/org/apache/giraph/conf/GiraphConstants.java
index 4c02fff..e9a5f32 100644
--- a/giraph-core/src/main/java/org/apache/giraph/conf/GiraphConstants.java
+++ b/giraph-core/src/main/java/org/apache/giraph/conf/GiraphConstants.java
@@ -690,6 +690,15 @@ public interface GiraphConstants {
new IntConfOption("giraph.maxRequestMilliseconds", MINUTES.toMillis(10),
"Milliseconds for a request to complete (or else resend)");
+ /**
+ * Whether to resend request which timed out or fail the job if timeout
+ * happens
+ */
+ BooleanConfOption RESEND_TIMED_OUT_REQUESTS =
+ new BooleanConfOption("giraph.resendTimedOutRequests", true,
+ "Whether to resend request which timed out or fail the job if " +
+ "timeout happens");
+
/** Netty max connection failures */
IntConfOption NETTY_MAX_CONNECTION_FAILURES =
new IntConfOption("giraph.nettyMaxConnectionFailures", 1000,