You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2018/03/19 07:16:29 UTC
[08/12] ignite git commit: GG-13537 dummy ConnectionKey was replaced
with regular one (dummy flag was added)
GG-13537 dummy ConnectionKey was replaced with regular one (dummy flag was added)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/dd55a928
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/dd55a928
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/dd55a928
Branch: refs/heads/ignite-zk
Commit: dd55a92847e10b811330e537f6387ca5c28b56ab
Parents: 2185bbd
Author: Sergey Chugunov <se...@gmail.com>
Authored: Wed Mar 7 12:50:17 2018 +0300
Committer: Sergey Chugunov <se...@gmail.com>
Committed: Wed Mar 7 12:50:28 2018 +0300
----------------------------------------------------------------------
.../communication/tcp/TcpCommunicationSpi.java | 9 +++---
.../tcp/internal/ConnectionKey.java | 29 ++++++++++++++++++--
.../TcpCommunicationConnectionCheckFuture.java | 17 ++++++------
3 files changed, 39 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd55a928/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index ac3ba1c..a0e942c 100755
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -151,7 +151,6 @@ import org.jsr166.ConcurrentLinkedDeque8;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
-import static org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture.CONN_CHECK_DUMMY_KEY;
import static org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture.SES_FUT_META;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.ALREADY_CONNECTED;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NEED_WAIT;
@@ -414,10 +413,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
@Override public void onDisconnected(GridNioSession ses, @Nullable Exception e) {
ConnectionKey connId = ses.meta(CONN_IDX_META);
- if (connId == CONN_CHECK_DUMMY_KEY)
- return;
-
if (connId != null) {
+ if (connId.dummy())
+ return;
+
UUID id = connId.nodeId();
GridCommunicationClient[] nodeClients = clients.get(id);
@@ -748,7 +747,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
recovery.lastAcknowledged(rcvCnt);
}
}
- else if (connKey == CONN_CHECK_DUMMY_KEY) {
+ else if (connKey.dummy()) {
assert msg instanceof NodeIdMessage : msg;
TcpCommunicationNodeConnectionCheckFuture fut = ses.meta(SES_FUT_META);
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd55a928/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
index 6716446..f03dd62 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/ConnectionKey.java
@@ -19,6 +19,7 @@ package org.apache.ignite.spi.communication.tcp.internal;
import java.util.UUID;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.NotNull;
/**
*
@@ -33,15 +34,32 @@ public class ConnectionKey {
/** */
private final long connCnt;
+ /** */
+ private final boolean dummy;
+
+ /**
+ * Creates ConnectionKey with false value of dummy flag.
+ *
+ * @param nodeId Node ID. Should be not null.
+ * @param idx Connection index.
+ * @param connCnt Connection counter (set only for incoming connections).
+ */
+ public ConnectionKey(@NotNull UUID nodeId, int idx, long connCnt) {
+ this(nodeId, idx, connCnt, false);
+ }
+
/**
- * @param nodeId Node ID.
+ * @param nodeId Node ID. Should be not null.
* @param idx Connection index.
* @param connCnt Connection counter (set only for incoming connections).
+ * @param dummy Indicates that session with this ConnectionKey is temporary
+ * (for now dummy sessions are used only for Communication Failure Resolving process).
*/
- public ConnectionKey(UUID nodeId, int idx, long connCnt) {
+ public ConnectionKey(@NotNull UUID nodeId, int idx, long connCnt, boolean dummy) {
this.nodeId = nodeId;
this.idx = idx;
this.connCnt = connCnt;
+ this.dummy = dummy;
}
/**
@@ -65,6 +83,13 @@ public class ConnectionKey {
return idx;
}
+ /**
+ * @return {@code True} if this ConnectionKey is dummy and serves temporary session.
+ */
+ public boolean dummy() {
+ return dummy;
+ }
+
/** {@inheritDoc} */
@Override public boolean equals(Object o) {
if (this == o)
http://git-wip-us.apache.org/repos/asf/ignite/blob/dd55a928/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpCommunicationConnectionCheckFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpCommunicationConnectionCheckFuture.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpCommunicationConnectionCheckFuture.java
index a2bfbda..5c7ef85 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpCommunicationConnectionCheckFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpCommunicationConnectionCheckFuture.java
@@ -55,9 +55,6 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
public static final int SES_FUT_META = GridNioSessionMetaKey.nextUniqueKey();
/** */
- public static final ConnectionKey CONN_CHECK_DUMMY_KEY = new ConnectionKey(null, -1, -1);
-
- /** */
private static final AtomicIntegerFieldUpdater<SingleAddressConnectFuture> connFutDoneUpdater =
AtomicIntegerFieldUpdater.newUpdater(SingleAddressConnectFuture.class, "done");
@@ -145,14 +142,14 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
if (addrs.size() == 1) {
SingleAddressConnectFuture fut = new SingleAddressConnectFuture(i);
- fut.init(addrs.iterator().next());
+ fut.init(addrs.iterator().next(), node.id());
futs[i] = fut;
}
else {
MultipleAddressesConnectFuture fut = new MultipleAddressesConnectFuture(i);
- fut.init(addrs);
+ fut.init(addrs, node.id());
futs[i] = fut;
}
@@ -295,8 +292,9 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
/**
* @param addr Node address.
+ * @param rmtNodeId Id of node to open connection check session with.
*/
- public void init(InetSocketAddress addr) {
+ public void init(InetSocketAddress addr, UUID rmtNodeId) {
boolean connect;
try {
@@ -319,7 +317,7 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
sesMeta = new GridLeanMap<>(3);
// Set dummy key to identify connection-check outgoing connection.
- sesMeta.put(TcpCommunicationSpi.CONN_IDX_META, CONN_CHECK_DUMMY_KEY);
+ sesMeta.put(TcpCommunicationSpi.CONN_IDX_META, new ConnectionKey(rmtNodeId, -1, -1, true));
sesMeta.put(SES_FUT_META, this);
nioSrvr.createSession(ch, sesMeta, true, new IgniteInClosure<IgniteInternalFuture<GridNioSession>>() {
@@ -424,8 +422,9 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
/**
* @param addrs Node addresses.
+ * @param rmtNodeId Id of node to open connection check session with.
*/
- void init(Collection<InetSocketAddress> addrs) {
+ void init(Collection<InetSocketAddress> addrs, UUID rmtNodeId) {
SingleAddressConnectFuture[] futs = new SingleAddressConnectFuture[addrs.size()];
for (int i = 0; i < addrs.size(); i++) {
@@ -443,7 +442,7 @@ public class TcpCommunicationConnectionCheckFuture extends GridFutureAdapter<Bit
int idx = 0;
for (InetSocketAddress addr : addrs) {
- futs[idx++].init(addr);
+ futs[idx++].init(addr, rmtNodeId);
if (resCnt == Integer.MAX_VALUE)
return;