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 2015/04/08 08:23:19 UTC
[14/22] incubator-ignite git commit: ignite-621 fix
ignite-621 fix
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/65f34729
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/65f34729
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/65f34729
Branch: refs/heads/ignite-680
Commit: 65f347299c8699ec8b43dad8ec63a40758677dd2
Parents: 2444811
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Tue Apr 7 17:42:04 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Tue Apr 7 17:42:04 2015 +0300
----------------------------------------------------------------------
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 24 +++++-
.../communication/tcp/TcpCommunicationSpi.java | 14 ++--
.../discovery/tcp/TcpClientDiscoverySpi.java | 47 ++++++------
.../spi/discovery/tcp/TcpDiscoverySpi.java | 78 ++++++++++----------
.../discovery/tcp/TcpDiscoverySpiAdapter.java | 8 +-
.../spi/swapspace/file/FileSwapSpaceSpi.java | 4 +-
.../GridAbstractCommunicationSelfTest.java | 3 +-
...mmunicationSpiConcurrentConnectSelfTest.java | 3 +-
...cpCommunicationSpiMultithreadedSelfTest.java | 3 +-
...dTcpCommunicationSpiRecoveryAckSelfTest.java | 3 +-
...GridTcpCommunicationSpiRecoverySelfTest.java | 2 +-
11 files changed, 104 insertions(+), 85 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
index 65303db..f198210 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
@@ -54,10 +54,15 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
@LoggerResource
private IgniteLogger log;
- /** Ignite instance */
- @IgniteInstanceResource
+ /** Ignite instance. */
protected Ignite ignite;
+ /** Local node id. */
+ protected UUID nodeId;
+
+ /** Grid instance name. */
+ protected String gridName;
+
/** SPI name. */
private String name;
@@ -105,7 +110,7 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
/** {@inheritDoc} */
@Override public UUID getLocalNodeId() {
- return ignite.configuration().getNodeId();
+ return nodeId;
}
/** {@inheritDoc} */
@@ -189,6 +194,19 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
}
/**
+ * Inject ignite instance.
+ */
+ @IgniteInstanceResource
+ protected void injectResources(Ignite ignite){
+ this.ignite = ignite;
+
+ if (ignite != null) {
+ nodeId = ignite.configuration().getNodeId();
+ gridName = ignite.name();
+ }
+ }
+
+ /**
* Method to be called in the beginning of onContextDestroyed() method.
*/
protected void onContextDestroyed0() {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/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 613ff64..6cd439b 100644
--- 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
@@ -591,9 +591,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** Local port range. */
private int locPortRange = DFLT_PORT_RANGE;
- /** Grid name. */
- private String gridName;
-
/** Allocate direct buffer or heap buffer. */
private boolean directBuf = true;
@@ -742,10 +739,11 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
*/
@IgniteInstanceResource
protected void injectResources(Ignite ignite) {
+ super.injectResources(ignite);
+
if (ignite != null) {
setAddressResolver(ignite.configuration().getAddressResolver());
setLocalAddress(ignite.configuration().getLocalHost());
- gridName = ignite.name();
}
}
@@ -1189,7 +1187,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
/** {@inheritDoc} */
@Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException {
- nodeIdMsg = new NodeIdMessage(ignite.configuration().getNodeId());
+ nodeIdMsg = new NodeIdMessage(getLocalNodeId());
assertParameter(locPort > 1023, "locPort > 1023");
assertParameter(locPort <= 0xffff, "locPort < 0xffff");
@@ -1557,7 +1555,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (log.isTraceEnabled())
log.trace("Sending message to node [node=" + node + ", msg=" + msg + ']');
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (node.id().equals(locNodeId))
notifyListener(locNodeId, msg, NOOP);
@@ -1957,7 +1955,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
ch.write(ByteBuffer.wrap(U.IGNITE_HEADER));
if (recovery != null) {
- HandshakeMessage msg = new HandshakeMessage(ignite.configuration().getNodeId(),
+ HandshakeMessage msg = new HandshakeMessage(getLocalNodeId(),
recovery.incrementConnectCount(),
recovery.receivedCount());
@@ -2630,7 +2628,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
out.flush();
if (log.isDebugEnabled())
- log.debug("Sent local node ID [locNodeId=" + ignite.configuration().getNodeId() + ", rmtNodeId="
+ log.debug("Sent local node ID [locNodeId=" + getLocalNodeId() + ", rmtNodeId="
+ rmtNodeId + ']');
}
catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
index 48125bf..bf69efb 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
@@ -255,7 +255,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
}
locNode = new TcpDiscoveryNode(
- ignite.configuration().getNodeId(),
+ getLocalNodeId(),
addrs.get1(),
addrs.get2(),
0,
@@ -295,7 +295,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
leaveLatch = new CountDownLatch(1);
try {
- TcpDiscoveryNodeLeftMessage msg = new TcpDiscoveryNodeLeftMessage(ignite.configuration().getNodeId());
+ TcpDiscoveryNodeLeftMessage msg = new TcpDiscoveryNodeLeftMessage(getLocalNodeId());
msg.client(true);
@@ -344,7 +344,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
/** {@inheritDoc} */
@Nullable @Override public ClusterNode getNode(UUID nodeId) {
- if (ignite.configuration().getNodeId().equals(nodeId))
+ if (getLocalNodeId().equals(nodeId))
return locNode;
TcpDiscoveryNode node = rmtNodes.get(nodeId);
@@ -356,7 +356,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
@Override public boolean pingNode(UUID nodeId) {
assert nodeId != null;
- if (nodeId.equals(ignite.configuration().getNodeId()))
+ if (nodeId.equals(getLocalNodeId()))
return true;
TcpDiscoveryNode node = rmtNodes.get(nodeId);
@@ -427,7 +427,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
locNode.clientRouterNodeId(rmtNodeId);
TcpDiscoveryAbstractMessage msg = recon ?
- new TcpDiscoveryClientReconnectMessage(ignite.configuration().getNodeId(), rmtNodeId,
+ new TcpDiscoveryClientReconnectMessage(getLocalNodeId(), rmtNodeId,
lastMsgId) :
new TcpDiscoveryJoinRequestMessage(locNode, null);
@@ -547,7 +547,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
Socket sock = openSocket(addr);
- TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(ignite.configuration().getNodeId());
+ TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(getLocalNodeId());
req.client(true);
@@ -558,7 +558,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
UUID nodeId = res.creatorNodeId();
assert nodeId != null;
- assert !ignite.configuration().getNodeId().equals(nodeId);
+ assert !getLocalNodeId().equals(nodeId);
return F.t(sock, nodeId);
}
@@ -567,7 +567,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
* FOR TEST PURPOSE ONLY!
*/
void simulateNodeFailure() {
- U.warn(log, "Simulating client node failure: " + ignite.configuration().getNodeId());
+ U.warn(log, "Simulating client node failure: " + getLocalNodeId());
U.closeQuiet(sock);
@@ -591,7 +591,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
/**
*/
protected DisconnectHandler() {
- super(ignite.name(), "tcp-client-disco-disconnect-hnd", log);
+ super(gridName, "tcp-client-disco-disconnect-hnd", log);
}
/** {@inheritDoc} */
@@ -644,7 +644,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
/**
*/
protected HeartbeatSender() {
- super(ignite.name(), "tcp-client-disco-heartbeat-sender", log);
+ super(gridName, "tcp-client-disco-heartbeat-sender", log);
}
/** {@inheritDoc} */
@@ -662,8 +662,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
while (!isInterrupted()) {
U.sleep(hbFreq);
- TcpDiscoveryHeartbeatMessage msg =
- new TcpDiscoveryHeartbeatMessage(ignite.configuration().getNodeId());
+ TcpDiscoveryHeartbeatMessage msg = new TcpDiscoveryHeartbeatMessage(getLocalNodeId());
msg.client(true);
@@ -692,7 +691,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
* @param msgWrk Message worker.
*/
protected SocketReader(UUID nodeId, MessageWorker msgWrk) {
- super(ignite.name(), "tcp-client-disco-sock-reader", log);
+ super(gridName, "tcp-client-disco-sock-reader", log);
assert nodeId != null;
assert msgWrk != null;
@@ -760,7 +759,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
U.error(log, "Failed to read message [sock=" + sock0 + ", " +
- "locNodeId=" + ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ']', e);
+ "locNodeId=" + getLocalNodeId() + ", rmtNodeId=" + nodeId + ']', e);
IOException ioEx = X.cause(e, IOException.class);
@@ -775,14 +774,14 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
"[rmtNodeId=" + nodeId + ", err=" + clsNotFoundEx.getMessage() + ']');
else
LT.error(log, e, "Failed to read message [sock=" + sock0 + ", locNodeId=" +
- ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ']');
+ getLocalNodeId() + ", rmtNodeId=" + nodeId + ']');
}
}
}
catch (IOException e) {
if (log.isDebugEnabled())
U.error(log, "Connection failed [sock=" + sock0 + ", locNodeId=" +
- ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ']', e);
+ getLocalNodeId() + ", rmtNodeId=" + nodeId + ']', e);
}
finally {
U.closeQuiet(sock0);
@@ -877,7 +876,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
UUID newNodeId = node.id();
- if (ignite.configuration().getNodeId().equals(newNodeId)) {
+ if (getLocalNodeId().equals(newNodeId)) {
if (joinLatch.getCount() > 0) {
Collection<TcpDiscoveryNode> top = msg.topology();
@@ -933,7 +932,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
if (leaveLatch != null)
return;
- if (ignite.configuration().getNodeId().equals(msg.nodeId())) {
+ if (getLocalNodeId().equals(msg.nodeId())) {
if (joinLatch.getCount() > 0) {
long topVer = msg.topologyVersion();
@@ -986,7 +985,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
* @param msg Message.
*/
private void processNodeLeftMessage(TcpDiscoveryNodeLeftMessage msg) {
- if (ignite.configuration().getNodeId().equals(msg.creatorNodeId())) {
+ if (getLocalNodeId().equals(msg.creatorNodeId())) {
if (log.isDebugEnabled())
log.debug("Received node left message for local node: " + msg);
@@ -1031,7 +1030,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
if (leaveLatch != null)
return;
- if (!ignite.configuration().getNodeId().equals(msg.creatorNodeId())) {
+ if (!getLocalNodeId().equals(msg.creatorNodeId())) {
TcpDiscoveryNode node = rmtNodes.remove(msg.failedNodeId());
if (node == null) {
@@ -1063,12 +1062,12 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
if (leaveLatch != null)
return;
- if (ignite.configuration().getNodeId().equals(msg.creatorNodeId())) {
+ if (getLocalNodeId().equals(msg.creatorNodeId())) {
if (msg.senderNodeId() == null) {
Socket sock0 = sock;
if (sock0 != null) {
- msg.setMetrics(ignite.configuration().getNodeId(), metricsProvider.metrics());
+ msg.setMetrics(getLocalNodeId(), metricsProvider.metrics());
try {
writeToSocket(sock0, msg);
@@ -1117,7 +1116,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
if (leaveLatch != null)
return;
- if (ignite.configuration().getNodeId().equals(msg.creatorNodeId())) {
+ if (getLocalNodeId().equals(msg.creatorNodeId())) {
if (msg.success()) {
pending = true;
@@ -1158,7 +1157,7 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
assert nodeId != null;
assert metrics != null;
- TcpDiscoveryNode node = nodeId.equals(ignite.configuration().getNodeId()) ? locNode : rmtNodes.get(nodeId);
+ TcpDiscoveryNode node = nodeId.equals(getLocalNodeId()) ? locNode : rmtNodes.get(nodeId);
if (node != null && node.visible()) {
node.setMetrics(metrics);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index 1cee279..bad8837 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -626,7 +626,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
@Nullable @Override public ClusterNode getNode(UUID nodeId) {
assert nodeId != null;
- UUID locNodeId0 = ignite.configuration().getNodeId();
+ UUID locNodeId0 = getLocalNodeId();
if (locNodeId0 != null && locNodeId0.equals(nodeId))
// Return local node directly.
@@ -703,7 +703,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
}
locNode = new TcpDiscoveryNode(
- ignite.configuration().getNodeId(),
+ getLocalNodeId(),
addrs.get1(),
addrs.get2(),
tcpSrvr.port,
@@ -856,7 +856,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
if (hbFreq < 2000)
U.warn(log, "Heartbeat frequency is too high (at least 2000 ms recommended): " + hbFreq);
- registerMBean(ignite.name(), this, TcpDiscoverySpiMBean.class);
+ registerMBean(gridName, this, TcpDiscoverySpiMBean.class);
if (ipFinder instanceof TcpDiscoveryMulticastIpFinder) {
TcpDiscoveryMulticastIpFinder mcastIpFinder = ((TcpDiscoveryMulticastIpFinder)ipFinder);
@@ -926,7 +926,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
if (msgWorker != null && msgWorker.isAlive() && !disconnect) {
// Send node left message only if it is final stop.
- msgWorker.addMessage(new TcpDiscoveryNodeLeftMessage(ignite.configuration().getNodeId()));
+ msgWorker.addMessage(new TcpDiscoveryNodeLeftMessage(getLocalNodeId()));
synchronized (mux) {
long threshold = U.currentTimeMillis() + netTimeout;
@@ -1088,7 +1088,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
@Override public boolean pingNode(UUID nodeId) {
assert nodeId != null;
- if (nodeId == ignite.configuration().getNodeId())
+ if (nodeId == getLocalNodeId())
return true;
TcpDiscoveryNode node = ring.node(nodeId);
@@ -1116,7 +1116,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
private boolean pingNode(TcpDiscoveryNode node) {
assert node != null;
- if (node.id().equals(ignite.configuration().getNodeId()))
+ if (node.id().equals(getLocalNodeId()))
return true;
UUID clientNodeId = null;
@@ -1160,10 +1160,10 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
throws IgniteCheckedException {
assert addr != null;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (F.contains(locNodeAddrs, addr))
- return F.t(ignite.configuration().getNodeId(), false);
+ return F.t(getLocalNodeId(), false);
GridFutureAdapter<IgniteBiTuple<UUID, Boolean>> fut = new GridFutureAdapter<>();
@@ -1388,7 +1388,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
@SuppressWarnings({"BusyWait"})
private boolean sendJoinRequestMessage() throws IgniteSpiException {
TcpDiscoveryAbstractMessage joinReq = new TcpDiscoveryJoinRequestMessage(locNode,
- exchange.collect(ignite.configuration().getNodeId()));
+ exchange.collect(getLocalNodeId()));
// Time when it has been detected, that addresses from IP finder do not respond.
long noResStart = 0;
@@ -1550,7 +1550,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
boolean joinReqSent = false;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
for (int i = 0; i < reconCnt; i++) {
// Need to set to false on each new iteration,
@@ -1985,7 +1985,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* This method is intended for test purposes only.
*/
void simulateNodeFailure() {
- U.warn(log, "Simulating node failure: " + ignite.configuration().getNodeId());
+ U.warn(log, "Simulating node failure: " + getLocalNodeId());
U.interrupt(tcpSrvr);
U.join(tcpSrvr, log);
@@ -2034,7 +2034,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
}
if (next != null)
- msgWorker.addMessage(new TcpDiscoveryNodeFailedMessage(ignite.configuration().getNodeId(), next.id(),
+ msgWorker.addMessage(new TcpDiscoveryNodeFailedMessage(getLocalNodeId(), next.id(),
next.internalOrder()));
}
@@ -2085,7 +2085,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
b.append(">>>").append("Dumping discovery SPI debug info.").append(U.nl());
b.append(">>>").append(U.nl());
- b.append("Local node ID: ").append(ignite.configuration().getNodeId()).append(U.nl()).append(U.nl());
+ b.append("Local node ID: ").append(getLocalNodeId()).append(U.nl()).append(U.nl());
b.append("Local node: ").append(locNode).append(U.nl()).append(U.nl());
b.append("SPI state: ").append(spiState).append(U.nl()).append(U.nl());
@@ -2141,7 +2141,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
assert debugMode;
String msg0 = new SimpleDateFormat("[HH:mm:ss,SSS]").format(new Date(System.currentTimeMillis())) +
- '[' + Thread.currentThread().getName() + "][" + ignite.configuration().getNodeId() +
+ '[' + Thread.currentThread().getName() + "][" + getLocalNodeId() +
"-" + locNode.internalOrder() + "] " +
msg;
@@ -2205,7 +2205,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* Constructor.
*/
private HeartbeatsSender() {
- super(ignite.name(), "tcp-disco-hb-sender", log);
+ super(gridName, "tcp-disco-hb-sender", log);
setPriority(threadPri);
}
@@ -2227,9 +2227,9 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
return;
}
- TcpDiscoveryHeartbeatMessage msg = new TcpDiscoveryHeartbeatMessage(ignite.configuration().getNodeId());
+ TcpDiscoveryHeartbeatMessage msg = new TcpDiscoveryHeartbeatMessage(getLocalNodeId());
- msg.verify(ignite.configuration().getNodeId());
+ msg.verify(getLocalNodeId());
msgWorker.addMessage(msg);
@@ -2249,7 +2249,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* Constructor.
*/
private CheckStatusSender() {
- super(ignite.name(), "tcp-disco-status-check-sender", log);
+ super(gridName, "tcp-disco-status-check-sender", log);
setPriority(threadPri);
}
@@ -2313,7 +2313,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* Constructor.
*/
private IpFinderCleaner() {
- super(ignite.name(), "tcp-disco-ip-finder-cleaner", log);
+ super(gridName, "tcp-disco-ip-finder-cleaner", log);
setPriority(threadPri);
}
@@ -2579,7 +2579,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
ClientMessageWorker wrk = clientMsgWorkers.get(msg.creatorNodeId());
if (wrk != null) {
- msg.verify(ignite.configuration().getNodeId());
+ msg.verify(getLocalNodeId());
wrk.addMessage(msg);
}
@@ -2638,7 +2638,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
boolean searchNext = true;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
while (true) {
if (searchNext) {
@@ -3035,7 +3035,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
TcpDiscoveryNode node = msg.node();
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (!msg.client()) {
boolean rmtHostLoopback = node.socketAddresses().size() == 1 &&
@@ -3354,7 +3354,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* @param msg Client reconnect message.
*/
private void processClientReconnectMessage(TcpDiscoveryClientReconnectMessage msg) {
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
boolean isLocalNodeRouter = locNodeId.equals(msg.routerNodeId());
@@ -3436,7 +3436,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
return;
}
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (isLocalNodeCoordinator()) {
if (msg.verified()) {
@@ -3650,7 +3650,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
boolean locNodeCoord = isLocalNodeCoordinator();
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (locNodeCoord) {
if (msg.verified()) {
@@ -3759,7 +3759,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
private void processNodeLeftMessage(TcpDiscoveryNodeLeftMessage msg) {
assert msg != null;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
UUID leavingNodeId = msg.creatorNodeId();
@@ -4004,7 +4004,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
boolean locNodeCoord = isLocalNodeCoordinator();
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (locNodeCoord) {
if (msg.verified()) {
@@ -4093,7 +4093,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
private void processStatusCheckMessage(TcpDiscoveryStatusCheckMessage msg) {
assert msg != null;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (msg.failedNodeId() != null) {
if (locNodeId.equals(msg.failedNodeId())) {
@@ -4215,7 +4215,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
private void processHeartbeatMessage(TcpDiscoveryHeartbeatMessage msg) {
assert msg != null;
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
if (ring.node(msg.creatorNodeId()) == null) {
if (log.isDebugEnabled())
@@ -4346,9 +4346,9 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
assert msgId != null;
if (isLocalNodeCoordinator()) {
- if (!ignite.configuration().getNodeId().equals(msg.verifierNodeId()))
+ if (!getLocalNodeId().equals(msg.verifierNodeId()))
// Message is not verified or verified by former coordinator.
- msg.verify(ignite.configuration().getNodeId());
+ msg.verify(getLocalNodeId());
else
// Discard the message.
return;
@@ -4424,7 +4424,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* @throws IgniteSpiException In case of error.
*/
TcpServer() throws IgniteSpiException {
- super(ignite.name(), "tcp-disco-srvr", log);
+ super(gridName, "tcp-disco-srvr", log);
setPriority(threadPri);
@@ -4522,7 +4522,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* @param sock Socket to read data from.
*/
SocketReader(Socket sock) {
- super(ignite.name(), "tcp-disco-sock-reader", log);
+ super(gridName, "tcp-disco-sock-reader", log);
this.sock = sock;
@@ -4533,7 +4533,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
/** {@inheritDoc} */
@Override protected void body() throws InterruptedException {
- UUID locNodeId = ignite.configuration().getNodeId();
+ UUID locNodeId = getLocalNodeId();
try {
InputStream in;
@@ -4995,7 +4995,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
if (state == CONNECTING) {
if (noResAddrs.contains(rmtAddr) ||
- ignite.configuration().getNodeId().compareTo(msg.creatorNodeId()) < 0)
+ getLocalNodeId().compareTo(msg.creatorNodeId()) < 0)
// Remote node node has not responded to join request or loses UUID race.
res = RES_WAIT;
else
@@ -5053,7 +5053,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
* Constructor.
*/
StatisticsPrinter() {
- super(ignite.name(), "tcp-disco-stats-printer", log);
+ super(gridName, "tcp-disco-stats-printer", log);
assert statsPrintFreq > 0;
@@ -5130,7 +5130,7 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
if (log.isDebugEnabled())
log.debug("Redirecting message to client [sock=" + sock + ", locNodeId="
- + ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']');
+ + getLocalNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']');
try {
prepareNodeAddedMessage(msg, nodeId, null, null);
@@ -5144,10 +5144,10 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
catch (IgniteCheckedException | IOException e) {
if (log.isDebugEnabled())
U.error(log, "Client connection failed [sock=" + sock + ", locNodeId="
- + ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']', e);
+ + getLocalNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']', e);
onException("Client connection failed [sock=" + sock + ", locNodeId="
- + ignite.configuration().getNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']', e);
+ + getLocalNodeId() + ", rmtNodeId=" + nodeId + ", msg=" + msg + ']', e);
U.interrupt(clientMsgWorkers.remove(nodeId));
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
index 1d9559e..e949846 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
@@ -152,6 +152,8 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov
*/
@IgniteInstanceResource
protected void injectResources(Ignite ignite) {
+ super.injectResources(ignite);
+
// Inject resource.
if (ignite != null)
setLocalAddress(ignite.configuration().getLocalHost());
@@ -767,7 +769,7 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov
*
*/
SocketTimeoutWorker() {
- super(ignite.name(), "tcp-disco-sock-timeout-worker", log);
+ super(gridName, "tcp-disco-sock-timeout-worker", log);
setPriority(threadPri);
}
@@ -935,7 +937,7 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov
* @param name Thread name.
*/
protected MessageWorkerAdapter(String name) {
- super(ignite.name(), name, log);
+ super(gridName, name, log);
setPriority(threadPri);
}
@@ -943,7 +945,7 @@ abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements Discov
/** {@inheritDoc} */
@Override protected void body() throws InterruptedException {
if (log.isDebugEnabled())
- log.debug("Message worker started [locNodeId=" + ignite.configuration().getNodeId() + ']');
+ log.debug("Message worker started [locNodeId=" + getLocalNodeId() + ']');
while (!isInterrupted()) {
TcpDiscoveryAbstractMessage msg = queue.poll(2000, TimeUnit.MILLISECONDS);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/main/java/org/apache/ignite/spi/swapspace/file/FileSwapSpaceSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/swapspace/file/FileSwapSpaceSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/swapspace/file/FileSwapSpaceSpi.java
index 1d2670e..5df37d7 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/swapspace/file/FileSwapSpaceSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/swapspace/file/FileSwapSpaceSpi.java
@@ -257,7 +257,7 @@ public class FileSwapSpaceSpi extends IgniteSpiAdapter implements SwapSpaceSpi,
registerMBean(gridName, this, FileSwapSpaceSpiMBean.class);
- String path = baseDir + File.separator + gridName + File.separator + ignite.configuration().getNodeId();
+ String path = baseDir + File.separator + gridName + File.separator + getLocalNodeId();
try {
dir = U.resolveWorkDirectory(path, true);
@@ -1433,8 +1433,6 @@ public class FileSwapSpaceSpi extends IgniteSpiAdapter implements SwapSpaceSpi,
final Object mux = new Object();
- String gridName = ignite.name();
-
writer = new IgniteSpiThread(gridName, "Swap writer: " + name, log) {
@Override protected void body() throws InterruptedException {
while (!isInterrupted()) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
index 0492dd1..bfed977 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
@@ -24,6 +24,7 @@ import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.spi.*;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.*;
import org.apache.ignite.testframework.junits.spi.*;
@@ -280,7 +281,7 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
for (int i = 0; i < getSpiCount(); i++) {
CommunicationSpi<Message> spi = getSpi(i);
- GridTestUtils.setFieldValue(spi, "gridName", "grid-" + i);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "gridName", "grid-" + i);
IgniteTestResources rsrcs = new IgniteTestResources();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
index 2ccb89d..c038ee7 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiConcurrentConnectSelfTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.spi.*;
import org.apache.ignite.spi.communication.*;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.*;
@@ -317,7 +318,7 @@ public class GridTcpCommunicationSpiConcurrentConnectSelfTest<T extends Communic
for (int i = 0; i < SPI_CNT; i++) {
CommunicationSpi<Message> spi = createSpi();
- GridTestUtils.setFieldValue(spi, "gridName", "grid-" + i);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "gridName", "grid-" + i);
IgniteTestResources rsrcs = new IgniteTestResources();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
index 692de62..e7ae957 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiMultithreadedSelfTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.spi.*;
import org.apache.ignite.spi.communication.*;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.*;
@@ -439,7 +440,7 @@ public class GridTcpCommunicationSpiMultithreadedSelfTest extends GridSpiAbstrac
for (int i = 0; i < getSpiCount(); i++) {
CommunicationSpi<Message> spi = newCommunicationSpi();
- GridTestUtils.setFieldValue(spi, "gridName", "grid-" + i);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "gridName", "grid-" + i);
IgniteTestResources rsrcs = new IgniteTestResources();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
index f1d0847..c0f0b11 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoveryAckSelfTest.java
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
import org.apache.ignite.lang.*;
import org.apache.ignite.plugin.extensions.communication.*;
+import org.apache.ignite.spi.*;
import org.apache.ignite.spi.communication.*;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.*;
@@ -343,7 +344,7 @@ public class GridTcpCommunicationSpiRecoveryAckSelfTest<T extends CommunicationS
for (int i = 0; i < SPI_CNT; i++) {
TcpCommunicationSpi spi = getSpi(ackCnt, idleTimeout, queueLimit);
- GridTestUtils.setFieldValue(spi, "gridName", "grid-" + i);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "gridName", "grid-" + i);
IgniteTestResources rsrcs = new IgniteTestResources();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65f34729/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
index 1b941f1..7463388 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiRecoverySelfTest.java
@@ -632,7 +632,7 @@ public class GridTcpCommunicationSpiRecoverySelfTest<T extends CommunicationSpi>
for (int i = 0; i < SPI_CNT; i++) {
TcpCommunicationSpi spi = getSpi(i);
- GridTestUtils.setFieldValue(spi, "gridName", "grid-" + i);
+ GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "gridName", "grid-" + i);
IgniteTestResources rsrcs = new IgniteTestResources();