You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2019/02/28 10:05:20 UTC
[ignite] branch master updated: IGNITE-11262 Compression on
Discovery data bag - Fixes #6125.
This is an automated email from the ASF dual-hosted git repository.
irakov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e237383 IGNITE-11262 Compression on Discovery data bag - Fixes #6125.
e237383 is described below
commit e237383a83f50530accfdc8b2e5e628db35f62b3
Author: Vladislav Pyatkov <vl...@gmail.com>
AuthorDate: Thu Feb 28 13:04:52 2019 +0300
IGNITE-11262 Compression on Discovery data bag - Fixes #6125.
Signed-off-by: Ivan Rakov <ir...@apache.org>
---
.../org/apache/ignite/internal/IgniteFeatures.java | 5 ++-
.../apache/ignite/internal/util/IgniteUtils.java | 27 +++++++++++++
.../ignite/spi/discovery/tcp/ClientImpl.java | 5 +++
.../ignite/spi/discovery/tcp/ServerImpl.java | 25 +++++++++++-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 27 +++++++++++--
.../tcp/internal/DiscoveryDataPacket.java | 47 +++++++++++++++++-----
.../tcp/messages/TcpDiscoveryAbstractMessage.java | 3 ++
.../messages/TcpDiscoveryHandshakeResponse.java | 14 +++++++
8 files changed, 137 insertions(+), 16 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
index 011c419..9c23222 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
@@ -37,7 +37,10 @@ public enum IgniteFeatures {
CACHE_METRICS_V2(1),
/** Distributed metastorage. */
- DISTRIBUTED_METASTORAGE(2);
+ DISTRIBUTED_METASTORAGE(2),
+
+ /** Data paket compression. */
+ DATA_PACKET_COMPRESSION(3);
/**
* Unique feature identifier.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 2686840..ffbb76d 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -9991,6 +9991,33 @@ public abstract class IgniteUtils {
}
/**
+ * @param zipBytes Zipped bytes.
+ * @return Raw bytes.
+ * @throws IgniteCheckedException If unzip resulted in error.
+ */
+ public static byte[] unzip(byte[] zipBytes) throws IgniteCheckedException {
+ assert zipBytes != null;
+
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ZipInputStream in = new ZipInputStream(new ByteArrayInputStream(zipBytes))) {
+
+ in.getNextEntry();
+
+ byte[] tmp = new byte[4 << 10];
+
+ int size;
+
+ while ((size = in.read(tmp)) != -1)
+ baos.write(tmp, 0, size);
+
+ return baos.toByteArray();
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ /**
* Unmarshals object from the input stream using given class loader.
* This method should not close given input stream.
* <p/>
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 0ec59b1..196b9ad 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
@@ -746,6 +746,11 @@ class ClientImpl extends TcpDiscoveryImpl {
discoveryData = spi.collectExchangeData(new DiscoveryDataPacket(getLocalNodeId()));
msg = new TcpDiscoveryJoinRequestMessage(node, discoveryData);
+
+ // During marshalling, SPI didn't know whether all nodes support compression as we didn't join yet.
+ // The only way to know is passing flag directly with handshake response.
+ if (!res.isDiscoveryDataPacketCompression())
+ ((TcpDiscoveryJoinRequestMessage)msg).gridDiscoveryData().unzipData(log);
}
else
msg = new TcpDiscoveryClientReconnectMessage(getLocalNodeId(), rmtNodeId, lastMsgId);
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 214641d..50af845 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -187,7 +187,7 @@ import static org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryStatusChe
/**
*
*/
-@SuppressWarnings({"MissortedModifiers", "IfMayBeConditional"})
+@SuppressWarnings({"IfMayBeConditional"})
class ServerImpl extends TcpDiscoveryImpl {
/** */
private static final int ENSURED_MSG_HIST_SIZE = getInteger(IGNITE_DISCOVERY_CLIENT_RECONNECT_HISTORY_SIZE, 512);
@@ -1321,6 +1321,11 @@ class ServerImpl extends TcpDiscoveryImpl {
if (msg instanceof TcpDiscoveryJoinRequestMessage) {
boolean ignore = false;
+ // During marshalling, SPI didn't know whether all nodes support compression as we didn't join yet.
+ // The only way to know is passing flag directly with handshake response.
+ if (!res.isDiscoveryDataPacketCompression())
+ ((TcpDiscoveryJoinRequestMessage)msg).gridDiscoveryData().unzipData(log);
+
synchronized (mux) {
for (TcpDiscoveryNode failedNode : failedNodes.keySet()) {
if (failedNode.id().equals(res.creatorNodeId())) {
@@ -3993,7 +3998,12 @@ class ServerImpl extends TcpDiscoveryImpl {
err = spi.getSpiContext().validateNode(node);
if (err == null)
- err = spi.getSpiContext().validateNode(node, msg.gridDiscoveryData().unmarshalJoiningNodeData(spi.marshaller(), U.resolveClassLoader(spi.ignite().configuration()), false, log));
+ err = spi.getSpiContext().validateNode(node, msg.gridDiscoveryData().unmarshalJoiningNodeData(
+ spi.marshaller(),
+ U.resolveClassLoader(spi.ignite().configuration()),
+ false,
+ isDiscoDataPaketCompressed(node),
+ log));
if (err != null) {
final IgniteNodeValidationResult err0 = err;
@@ -4308,6 +4318,15 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
+ * @param node Joined node.
+ * @return <code>true</code> if packet is compressed.
+ */
+ private boolean isDiscoDataPaketCompressed(TcpDiscoveryNode node) {
+ return IgniteFeatures.nodeSupports(node, IgniteFeatures.DATA_PACKET_COMPRESSION) &&
+ allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION);
+ }
+
+ /**
* @param node Node.
* @param name Attribute name.
* @param dflt Default value.
@@ -6332,6 +6351,8 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoveryHandshakeResponse res =
new TcpDiscoveryHandshakeResponse(locNodeId, locNode.internalOrder());
+ res.setDiscoveryDataPacketCompression(allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION));
+
if (req.client())
res.clientAck(true);
else if (req.changeTopology()) {
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 b5f08a4..fcb88a6 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
@@ -2030,9 +2030,20 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements IgniteDiscovery
//marshall collected bag into packet, return packet
if (dataPacket.joiningNodeId().equals(locNode.id()))
- dataPacket.marshalJoiningNodeData(dataBag, marshaller(), log);
+ dataPacket.marshalJoiningNodeData(
+ dataBag,
+ marshaller(),
+ allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION),
+ ignite.configuration().getNetworkCompressionLevel(),
+ log);
else
- dataPacket.marshalGridNodeData(dataBag, locNode.id(), marshaller(), log);
+ dataPacket.marshalGridNodeData(
+ dataBag,
+ locNode.id(),
+ marshaller(),
+ allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION),
+ ignite.configuration().getNetworkCompressionLevel(),
+ log);
return dataPacket;
}
@@ -2051,9 +2062,17 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements IgniteDiscovery
DiscoveryDataBag dataBag;
if (dataPacket.joiningNodeId().equals(locNode.id()))
- dataBag = dataPacket.unmarshalGridData(marshaller(), clsLdr, locNode.clientRouterNodeId() != null, log);
+ dataBag = dataPacket.unmarshalGridData(marshaller(),
+ clsLdr,
+ locNode.clientRouterNodeId() != null,
+ allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION),
+ log);
else
- dataBag = dataPacket.unmarshalJoiningNodeData(marshaller(), clsLdr, locNode.clientRouterNodeId() != null, log);
+ dataBag = dataPacket.unmarshalJoiningNodeData(marshaller(),
+ clsLdr,
+ locNode.clientRouterNodeId() != null,
+ allNodesSupport(IgniteFeatures.DATA_PACKET_COMPRESSION),
+ log);
exchange.onExchange(dataBag);
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/DiscoveryDataPacket.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/DiscoveryDataPacket.java
index f342503..2664422 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/DiscoveryDataPacket.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/DiscoveryDataPacket.java
@@ -80,15 +80,16 @@ public class DiscoveryDataPacket implements Serializable {
* @param marsh Marsh.
* @param log Logger.
*/
- public void marshalGridNodeData(DiscoveryDataBag bag, UUID nodeId, Marshaller marsh, IgniteLogger log) {
- marshalData(bag.commonData(), commonData, marsh, log);
+ public void marshalGridNodeData(DiscoveryDataBag bag, UUID nodeId, Marshaller marsh, boolean isCompressionEnabled,
+ int compressionLevel, IgniteLogger log) {
+ marshalData(bag.commonData(), commonData, marsh, isCompressionEnabled, compressionLevel, log);
Map<Integer, Serializable> locNodeSpecificData = bag.localNodeSpecificData();
if (locNodeSpecificData != null) {
Map<Integer, byte[]> marshLocNodeSpecificData = U.newHashMap(locNodeSpecificData.size());
- marshalData(locNodeSpecificData, marshLocNodeSpecificData, marsh, log);
+ marshalData(locNodeSpecificData, marshLocNodeSpecificData, marsh, isCompressionEnabled, compressionLevel, log);
filterDuplicatedData(marshLocNodeSpecificData);
@@ -102,8 +103,9 @@ public class DiscoveryDataPacket implements Serializable {
* @param marsh Marsh.
* @param log Logger.
*/
- public void marshalJoiningNodeData(DiscoveryDataBag bag, Marshaller marsh, IgniteLogger log) {
- marshalData(bag.joiningNodeData(), joiningNodeData, marsh, log);
+ public void marshalJoiningNodeData(DiscoveryDataBag bag, Marshaller marsh, boolean isCompressionEnabled,
+ int compressionLevel, IgniteLogger log) {
+ marshalData(bag.joiningNodeData(), joiningNodeData, marsh, isCompressionEnabled, compressionLevel, log);
}
/**
@@ -116,12 +118,14 @@ public class DiscoveryDataPacket implements Serializable {
Marshaller marsh,
ClassLoader clsLdr,
boolean clientNode,
+ boolean isCompressionEnabled,
IgniteLogger log
) {
DiscoveryDataBag dataBag = new DiscoveryDataBag(joiningNodeId, joiningNodeClient);
if (commonData != null && !commonData.isEmpty()) {
- Map<Integer, Serializable> unmarshCommonData = unmarshalData(commonData, marsh, clsLdr, clientNode, log);
+ Map<Integer, Serializable> unmarshCommonData = unmarshalData(commonData, marsh, clsLdr, clientNode,
+ isCompressionEnabled, log);
dataBag.commonData(unmarshCommonData);
}
@@ -135,7 +139,8 @@ public class DiscoveryDataPacket implements Serializable {
if (nodeBinData == null || nodeBinData.isEmpty())
continue;
- Map<Integer, Serializable> unmarshData = unmarshalData(nodeBinData, marsh, clsLdr, clientNode, log);
+ Map<Integer, Serializable> unmarshData = unmarshalData(nodeBinData, marsh, clsLdr, clientNode,
+ isCompressionEnabled, log);
unmarshNodeSpecData.put(nodeBinEntry.getKey(), unmarshData);
}
@@ -156,6 +161,7 @@ public class DiscoveryDataPacket implements Serializable {
Marshaller marsh,
ClassLoader clsLdr,
boolean clientNode,
+ boolean isCompressionEnabled,
IgniteLogger log
) {
DiscoveryDataBag dataBag = new DiscoveryDataBag(joiningNodeId, joiningNodeClient);
@@ -166,6 +172,7 @@ public class DiscoveryDataPacket implements Serializable {
marsh,
clsLdr,
clientNode,
+ isCompressionEnabled,
log);
dataBag.joiningNodeData(unmarshalledJoiningNodeData);
@@ -272,13 +279,16 @@ public class DiscoveryDataPacket implements Serializable {
Marshaller marsh,
ClassLoader clsLdr,
boolean clientNode,
+ boolean isCompressionEnabled,
IgniteLogger log
) {
Map<Integer, Serializable> res = U.newHashMap(src.size());
for (Map.Entry<Integer, byte[]> binEntry : src.entrySet()) {
try {
- Serializable compData = marsh.unmarshal(binEntry.getValue(), clsLdr);
+ Serializable compData = isCompressionEnabled ?
+ U.unmarshalZip(marsh, binEntry.getValue(), clsLdr) :
+ U.unmarshal(marsh, binEntry.getValue(), clsLdr);
res.put(binEntry.getKey(), compData);
}
catch (IgniteCheckedException e) {
@@ -307,6 +317,8 @@ public class DiscoveryDataPacket implements Serializable {
Map<Integer, Serializable> src,
Map<Integer, byte[]> target,
Marshaller marsh,
+ boolean isCompressionEnabled,
+ int compressionLevel,
IgniteLogger log
) {
//may happen if nothing was collected from components,
@@ -316,7 +328,9 @@ public class DiscoveryDataPacket implements Serializable {
for (Map.Entry<Integer, Serializable> entry : src.entrySet()) {
try {
- target.put(entry.getKey(), marsh.marshal(entry.getValue()));
+ target.put(entry.getKey(), isCompressionEnabled ?
+ U.zip(U.marshal(marsh, entry.getValue()), compressionLevel) :
+ U.marshal(marsh, entry.getValue()));
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to marshal discovery data " +
@@ -326,6 +340,21 @@ public class DiscoveryDataPacket implements Serializable {
}
/**
+ * @param log Logger.
+ */
+ public void unzipData(IgniteLogger log) {
+ for (Map.Entry<Integer, byte[]> entry : joiningNodeData.entrySet()) {
+ try {
+ entry.setValue(U.unzip(entry.getValue()));
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to unzip discovery data " +
+ "[comp=" + entry.getKey() + ']', e);
+ }
+ }
+ }
+
+ /**
* TODO https://issues.apache.org/jira/browse/IGNITE-4435
*/
private void filterDuplicatedData(Map<Integer, byte[]> discoData) {
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
index 93d8bcf..5bb5e0a 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
@@ -54,6 +54,9 @@ public abstract class TcpDiscoveryAbstractMessage implements Serializable {
/** */
protected static final int FORCE_FAIL_FLAG_POS = 8;
+ /** */
+ protected static final int COMPRESS_DATA_PACKET = 9;
+
/** Sender of the message (transient). */
private transient UUID sndNodeId;
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryHandshakeResponse.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryHandshakeResponse.java
index 75df5c2..034b5a3 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryHandshakeResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryHandshakeResponse.java
@@ -43,6 +43,20 @@ public class TcpDiscoveryHandshakeResponse extends TcpDiscoveryAbstractMessage {
}
/**
+ * @return DiscoveryDataPacketCompression flag.
+ */
+ public boolean isDiscoveryDataPacketCompression() {
+ return getFlag(COMPRESS_DATA_PACKET);
+ }
+
+ /**
+ * @param discoveryDataPacketCompression DiscoveryDataPacketCompression flag.
+ */
+ public void setDiscoveryDataPacketCompression(boolean discoveryDataPacketCompression) {
+ setFlag(COMPRESS_DATA_PACKET, discoveryDataPacketCompression);
+ }
+
+ /**
* Gets previous node alive flag.<br>
* {@code True} means node has connectivity to it's previous node in a ring.
*