You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/07/28 12:43:48 UTC
[2/4] ignite git commit: Impleemented re-send logic.
Impleemented re-send logic.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/4346db20
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/4346db20
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/4346db20
Branch: refs/heads/ignite-3553
Commit: 4346db2037569fef988cb925ea97decf2716102d
Parents: c3af574
Author: vozerov-gridgain <vo...@gridgain.com>
Authored: Thu Jul 28 15:26:35 2016 +0300
Committer: vozerov-gridgain <vo...@gridgain.com>
Committed: Thu Jul 28 15:26:35 2016 +0300
----------------------------------------------------------------------
.../igfs/client/IgfsClientManager.java | 94 ++++++++++++++++----
.../igfs/client/IgfsClientOutOperation.java | 41 +++++++--
2 files changed, 113 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/4346db20/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientManager.java
index e8a19e7..478af45 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientManager.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.processors.igfs.IgfsImpl;
import org.apache.ignite.internal.processors.igfs.IgfsManager;
import org.apache.ignite.internal.util.StripedCompositeReadWriteLock;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
@@ -42,6 +43,9 @@ import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -189,40 +193,44 @@ public class IgfsClientManager extends IgfsManager {
}
/**
- * Execute IGFS closure asynchronously.
+ * Internal execution logic.
*
- * @param igfsCtx IGFS context.
- * @param clo Closure.
- * @param strategy Node selection strategy.
- * @return Future.
+ * @param op Output operation.
*/
- @SuppressWarnings("unchecked")
- public <T> IgniteInternalFuture<T> executeAsync(IgfsContext igfsCtx, IgfsClientAbstractCallable<T> clo,
- IgfsClientNodeSelectionStrategy strategy) {
+ private void executeAsync0(IgfsClientOutOperation op) {
while (true) {
rwLock.readLock().lock();
try {
+ if (stopping) {
+ op.future().onDone(new IgfsException("Failed to execute IGFS task because node is stopping."));
+
+ return;
+ }
+
// Get suitable node.
- ClusterNode node = selectNode(igfsCtx, strategy);
+ ClusterNode node = selectNode(op.igfsContext(), op.strategy());
+
+ if (node == null) {
+ op.future().onDone(new IgfsException("Failed to execute operation because there are no " +
+ "IGFS metadata nodes [igfs=" + igfsCtx.igfs().name() + ']'));
- if (node == null)
- throw new IgfsException("Failed to execute operation because there are no IGFS metadata nodes [igfs="
- + igfsCtx.igfs().name() + ']');
+ return;
+ }
+
+ op.nodeId(node.id());
// Add operation to pending set.
long msgId = msgIdGen.incrementAndGet();
- IgfsClientOutOperation op = new IgfsClientOutOperation(node.id(), clo, new GridFutureAdapter());
-
outOps.put(msgId, op);
// Send request.
try {
- ctx.io().send(node, GridTopic.TOPIC_IGFS_CLI, new IgfsClientRequest(msgId, clo),
+ ctx.io().send(node, GridTopic.TOPIC_IGFS_CLI, new IgfsClientRequest(msgId, op.target()),
GridIoPolicy.PUBLIC_POOL);
- return op.future();
+ return;
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
@@ -237,6 +245,26 @@ public class IgfsClientManager extends IgfsManager {
}
/**
+ * Execute IGFS closure asynchronously.
+ *
+ * @param igfsCtx IGFS context.
+ * @param clo Closure.
+ * @param strategy Node selection strategy.
+ * @return Future.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> IgniteInternalFuture<T> executeAsync(IgfsContext igfsCtx, IgfsClientAbstractCallable<T> clo,
+ IgfsClientNodeSelectionStrategy strategy) {
+ GridFutureAdapter fut = new GridFutureAdapter();
+
+ IgfsClientOutOperation op = new IgfsClientOutOperation(igfsCtx, clo, strategy, fut);
+
+ executeAsync0(op);
+
+ return fut;
+ }
+
+ /**
* Select the most appropriate node for the task.
*
* @param igfsCtx IGFS context.
@@ -285,7 +313,39 @@ public class IgfsClientManager extends IgfsManager {
* @param nodeId Node ID.
*/
private void onNodeLeft(UUID nodeId) {
- // TODO
+ Collection<IgfsClientOutOperation> retryOps = new LinkedList<>();
+
+ rwLock.writeLock().lock();
+
+ try {
+ if (!stopping) {
+ // Get the list of affected requests.
+ Iterator<Map.Entry<Long, IgfsClientOutOperation>> iter = outOps.entrySet().iterator();
+
+ while (iter.hasNext()) {
+ Map.Entry<Long, IgfsClientOutOperation> entry = iter.next();
+
+ long msgId = entry.getKey();
+ IgfsClientOutOperation op = entry.getValue();
+
+ if (F.eq(nodeId, op.nodeId())) {
+ iter.remove();
+
+ retryOps.add(op);
+ }
+ }
+ }
+ }
+ finally {
+ rwLock.writeLock().unlock();
+ }
+
+ // Re-send affected ops.
+ for (IgfsClientOutOperation retryOp : retryOps) {
+ retryOp.nodeId(null);
+
+ executeAsync0(retryOp);
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/4346db20/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientOutOperation.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientOutOperation.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientOutOperation.java
index 47cea16..6f3ff54 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientOutOperation.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/client/IgfsClientOutOperation.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.igfs.client;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.igfs.IgfsContext;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.internal.S;
@@ -27,25 +28,34 @@ import java.util.UUID;
* IGFS client closure outgoing operation descriptor.
*/
public class IgfsClientOutOperation {
- /** Target node ID. */
- private final UUID nodeId;
+ /** IGFS context. */
+ private final IgfsContext igfsCtx;
/** Target operation. */
private final IgfsClientAbstractCallable target;
+ /** Node selection strategy. */
+ private final IgfsClientNodeSelectionStrategy strategy;
+
/** Future completed when operation is ready. */
private final GridFutureAdapter fut;
+ /** Target node ID. */
+ private UUID nodeId;
+
/**
* Constructor.
*
- * @param nodeId Target node ID.
+ * @param igfsCtx IGFS context.
* @param target Target operation.
+ * @param strategy Node selection strategy.
* @param fut Future completed when operation is ready.
*/
- public IgfsClientOutOperation(UUID nodeId, IgfsClientAbstractCallable target, GridFutureAdapter fut) {
- this.nodeId = nodeId;
+ public IgfsClientOutOperation(IgfsContext igfsCtx, IgfsClientAbstractCallable target,
+ IgfsClientNodeSelectionStrategy strategy, GridFutureAdapter fut) {
+ this.igfsCtx = igfsCtx;
this.target = target;
+ this.strategy = strategy;
this.fut = fut;
}
@@ -57,6 +67,20 @@ public class IgfsClientOutOperation {
}
/**
+ * @param nodeId Node ID.
+ */
+ public void nodeId(UUID nodeId) {
+ this.nodeId = nodeId;
+ }
+
+ /**
+ * @return IGFS context.
+ */
+ public IgfsContext igfsContext() {
+ return igfsCtx;
+ }
+
+ /**
* @return Target operation.
*/
public IgfsClientAbstractCallable target() {
@@ -64,6 +88,13 @@ public class IgfsClientOutOperation {
}
/**
+ * @return Node selection strategy.
+ */
+ public IgfsClientNodeSelectionStrategy strategy() {
+ return strategy;
+ }
+
+ /**
* @return Future completed when operation is ready.
*/
public GridFutureAdapter future() {