You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by nt...@apache.org on 2016/04/07 16:46:37 UTC
[1/7] ignite git commit: IGNITE-2951 - Stability fixes for cluster
with many clients
Repository: ignite
Updated Branches:
refs/heads/ignite-2630 668895563 -> c180c127b
IGNITE-2951 - Stability fixes for cluster with many clients
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/5e266153
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/5e266153
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/5e266153
Branch: refs/heads/ignite-2630
Commit: 5e266153707021a8866e91dfa3f958066f80fc99
Parents: da47901
Author: Valentin Kulichenko <va...@gmail.com>
Authored: Wed Apr 6 18:10:45 2016 -0700
Committer: Valentin Kulichenko <va...@gmail.com>
Committed: Wed Apr 6 23:25:03 2016 -0700
----------------------------------------------------------------------
.../GridDhtPartitionsExchangeFuture.java | 2 +-
.../continuous/GridContinuousProcessor.java | 13 +++++-
.../ignite/spi/discovery/tcp/ClientImpl.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 44 +++++++++-----------
.../spi/discovery/tcp/TcpDiscoverySpi.java | 23 ++++------
.../IgniteClientReconnectAbstractTest.java | 7 ++--
.../tcp/TcpClientDiscoverySpiSelfTest.java | 8 ++--
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 39 +++++++++--------
.../TcpDiscoverySpiFailureTimeoutSelfTest.java | 23 ++++------
.../spi/discovery/tcp/TestTcpDiscoverySpi.java | 5 ++-
10 files changed, 80 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index bbfc71a..82e9bda 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -931,7 +931,7 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
locMap = new GridDhtPartitionMap(locMap.nodeId(), locMap.updateSequence(), locMap.map());
m.addLocalPartitionMap(cacheCtx.cacheId(), locMap);
-
+
m.partitionUpdateCounters(cacheCtx.cacheId(), cacheCtx.topology().updateCounters());
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index abafe85..d7838f3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -393,7 +393,18 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** {@inheritDoc} */
@Override @Nullable public Serializable collectDiscoveryData(UUID nodeId) {
if (!nodeId.equals(ctx.localNodeId()) || !locInfos.isEmpty()) {
- DiscoveryData data = new DiscoveryData(ctx.localNodeId(), clientInfos);
+ Map<UUID, Map<UUID, LocalRoutineInfo>> clientInfos0 = U.newHashMap(clientInfos.size());
+
+ for (Map.Entry<UUID, Map<UUID, LocalRoutineInfo>> e : clientInfos.entrySet()) {
+ Map<UUID, LocalRoutineInfo> copy = U.newHashMap(e.getValue().size());
+
+ for (Map.Entry<UUID, LocalRoutineInfo> e0 : e.getValue().entrySet())
+ copy.put(e0.getKey(), e0.getValue());
+
+ clientInfos0.put(e.getKey(), copy);
+ }
+
+ DiscoveryData data = new DiscoveryData(ctx.localNodeId(), clientInfos0);
// Collect listeners information (will be sent to joining node during discovery process).
for (Map.Entry<UUID, LocalRoutineInfo> e : locInfos.entrySet()) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
----------------------------------------------------------------------
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 950c680..31d614f 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
@@ -1062,7 +1062,7 @@ class ClientImpl extends TcpDiscoveryImpl {
try {
if (ack) {
synchronized (mux) {
- assert unackedMsg == null : unackedMsg;
+ assert unackedMsg == null : "Unacked=" + unackedMsg + ", received=" + msg;
unackedMsg = msg;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
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 88e34e8..27a31c4 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
@@ -18,9 +18,11 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectStreamException;
+import java.io.OutputStream;
import java.io.Serializable;
import java.net.ConnectException;
import java.net.InetAddress;
@@ -74,7 +76,6 @@ import org.apache.ignite.internal.util.GridBoundedLinkedHashSet;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
import org.apache.ignite.internal.util.lang.GridTuple;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.C1;
@@ -2134,6 +2135,9 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Socket. */
private Socket sock;
+ /** Output stream. */
+ private OutputStream out;
+
/** Last time status message has been sent. */
private long lastTimeStatusMsgSent;
@@ -2470,10 +2474,12 @@ class ServerImpl extends TcpDiscoveryImpl {
sock = spi.openSocket(addr, timeoutHelper);
+ out = new BufferedOutputStream(sock.getOutputStream(), sock.getSendBufferSize());
+
openSock = true;
// Handshake.
- writeToSocket(sock, new TcpDiscoveryHandshakeRequest(locNodeId),
+ spi.writeToSocket(sock, out, new TcpDiscoveryHandshakeRequest(locNodeId),
timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
TcpDiscoveryHandshakeResponse res = spi.readMessage(sock, null,
@@ -2627,7 +2633,7 @@ class ServerImpl extends TcpDiscoveryImpl {
timeoutHelper = new IgniteSpiOperationTimeoutHelper(spi);
try {
- writeToSocket(sock, pendingMsg, timeoutHelper.nextTimeoutChunk(
+ spi.writeToSocket(sock, out, pendingMsg, timeoutHelper.nextTimeoutChunk(
spi.getSocketTimeout()));
}
finally {
@@ -2679,7 +2685,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
}
- writeToSocket(sock, msg, timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
+ spi.writeToSocket(sock, out, msg, timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
spi.stats.onMessageSent(msg, U.currentTimeMillis() - tstamp);
@@ -3999,7 +4005,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (leftNode.equals(next) && sock != null) {
try {
- writeToSocket(sock, msg, spi.failureDetectionTimeoutEnabled() ?
+ spi.writeToSocket(sock, out, msg, spi.failureDetectionTimeoutEnabled() ?
spi.failureDetectionTimeout() : spi.getSocketTimeout());
if (log.isDebugEnabled())
@@ -5617,6 +5623,9 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Socket. */
private final Socket sock;
+ /** Output stream. */
+ private final OutputStream out;
+
/** Current client metrics. */
private volatile ClusterMetrics metrics;
@@ -5630,11 +5639,13 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param sock Socket.
* @param clientNodeId Node ID.
*/
- protected ClientMessageWorker(Socket sock, UUID clientNodeId) {
+ protected ClientMessageWorker(Socket sock, UUID clientNodeId) throws IOException {
super("tcp-disco-client-message-worker", 2000);
this.sock = sock;
this.clientNodeId = clientNodeId;
+
+ out = new BufferedOutputStream(sock.getOutputStream(), sock.getSendBufferSize());
}
/**
@@ -5681,7 +5692,7 @@ class ServerImpl extends TcpDiscoveryImpl {
log.debug("Sending message ack to client [sock=" + sock + ", locNodeId="
+ getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
- writeToSocket(sock, msg, spi.failureDetectionTimeoutEnabled() ?
+ spi.writeToSocket(sock, out, msg, spi.failureDetectionTimeoutEnabled() ?
spi.failureDetectionTimeout() : spi.getSocketTimeout());
}
}
@@ -5692,7 +5703,7 @@ class ServerImpl extends TcpDiscoveryImpl {
assert topologyInitialized(msg) : msg;
- writeToSocket(sock, msg, spi.failureDetectionTimeoutEnabled() ?
+ spi.writeToSocket(sock, out, msg, spi.failureDetectionTimeoutEnabled() ?
spi.failureDetectionTimeout() : spi.getSocketTimeout());
}
}
@@ -5799,9 +5810,6 @@ class ServerImpl extends TcpDiscoveryImpl {
* Base class for message workers.
*/
protected abstract class MessageWorkerAdapter extends IgniteSpiThread {
- /** Pre-allocated output stream (100K). */
- private final GridByteArrayOutputStream bout = new GridByteArrayOutputStream(100 * 1024);
-
/** Message queue. */
private final BlockingDeque<TcpDiscoveryAbstractMessage> queue = new LinkedBlockingDeque<>();
@@ -5883,20 +5891,6 @@ class ServerImpl extends TcpDiscoveryImpl {
protected void noMessageLoop() {
// No-op.
}
-
- /**
- * @param sock Socket.
- * @param msg Message.
- * @param timeout Socket timeout.
- * @throws IOException If IO failed.
- * @throws IgniteCheckedException If marshalling failed.
- */
- protected final void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout)
- throws IOException, IgniteCheckedException {
- bout.reset();
-
- spi.writeToSocket(sock, msg, bout, timeout);
- }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/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 df152f8..d981609 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
@@ -17,6 +17,7 @@
package org.apache.ignite.spi.discovery.tcp;
+import java.io.BufferedOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
@@ -51,7 +52,6 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.AddressResolver;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
-import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
@@ -1346,45 +1346,38 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
*/
protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
IgniteCheckedException {
- writeToSocket(sock, msg, new GridByteArrayOutputStream(8 * 1024), timeout); // 8K.
+ writeToSocket(sock, new BufferedOutputStream(sock.getOutputStream(), sock.getSendBufferSize()), msg, timeout);
}
/**
* Writes message to the socket.
*
* @param sock Socket.
+ * @param out Stream to write to.
* @param msg Message.
- * @param bout Byte array output stream.
* @param timeout Timeout.
* @throws IOException If IO failed or write timed out.
* @throws IgniteCheckedException If marshalling failed.
*/
@SuppressWarnings("ThrowFromFinallyBlock")
protected void writeToSocket(Socket sock,
+ OutputStream out,
TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout,
long timeout) throws IOException, IgniteCheckedException {
assert sock != null;
assert msg != null;
- assert bout != null;
-
- // Marshall message first to perform only write after.
- marsh.marshal(msg, bout);
+ assert out != null;
SocketTimeoutObject obj = new SocketTimeoutObject(sock, U.currentTimeMillis() + timeout);
addTimeoutObject(obj);
- IOException err = null;
+ IgniteCheckedException err = null;
try {
- OutputStream out = sock.getOutputStream();
-
- bout.writeTo(out);
-
- out.flush();
+ marsh.marshal(msg, out);
}
- catch (IOException e) {
+ catch (IgniteCheckedException e) {
err = e;
}
finally {
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
index 6869d1c..4d49366 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal;
import java.io.IOException;
+import java.io.OutputStream;
import java.net.Socket;
import java.util.Collection;
import java.util.Collections;
@@ -384,7 +385,7 @@ public abstract class IgniteClientReconnectAbstractTest extends GridCommonAbstra
volatile CountDownLatch writeLatch;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout)
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg, long timeout)
throws IOException, IgniteCheckedException {
if (msg instanceof TcpDiscoveryJoinRequestMessage) {
CountDownLatch writeLatch0 = writeLatch;
@@ -396,7 +397,7 @@ public abstract class IgniteClientReconnectAbstractTest extends GridCommonAbstra
}
}
- super.writeToSocket(sock, msg, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -464,4 +465,4 @@ public abstract class IgniteClientReconnectAbstractTest extends GridCommonAbstra
log.error(s);
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 7debb41..e01094c 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
+import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.ArrayList;
@@ -44,7 +45,6 @@ import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
-import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.typedef.CIX2;
@@ -2158,8 +2158,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
waitFor(writeLock);
boolean fail = false;
@@ -2184,7 +2184,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
sock.close();
}
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
if (afterWrite != null)
afterWrite.apply(msg, sock);
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
index 7635f0b..7efaca0 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.spi.discovery.tcp;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
@@ -54,7 +55,6 @@ import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage;
import org.apache.ignite.internal.processors.port.GridPortRecord;
-import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
@@ -1852,9 +1852,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
private volatile boolean failed;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock,
+ @Override protected void writeToSocket(Socket sock, OutputStream out,
TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout,
long timeout) throws IOException, IgniteCheckedException {
boolean add = msgIds.add(msg.id());
@@ -1864,7 +1863,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
failed = true;
}
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -1877,8 +1876,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override protected void writeToSocket(Socket sock,
+ OutputStream out,
TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout,
long timeout) throws IOException, IgniteCheckedException {
if (stopBeforeSndAck) {
if (msg instanceof TcpDiscoveryCustomEventMessage) {
@@ -1908,7 +1907,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
}
}
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -1940,8 +1939,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
/** {@inheritDoc} */
@Override protected void writeToSocket(Socket sock,
+ OutputStream out,
TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout,
long timeout) throws IOException, IgniteCheckedException {
if (stop)
return;
@@ -1986,7 +1985,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
return;
}
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -2001,8 +2000,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
private boolean stop;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
if (msg instanceof TcpDiscoveryCustomEventMessage && latch != null) {
log.info("Stop node on custom event: " + msg);
@@ -2014,7 +2013,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
if (stop)
return;
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -2035,8 +2034,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
private boolean debug;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
if (msg instanceof TcpDiscoveryNodeAddedMessage) {
if (nodeAdded1 != null) {
nodeAdded1.countDown();
@@ -2063,7 +2062,7 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
if (debug && msg instanceof TcpDiscoveryCustomEventMessage)
log.info("--- Send custom event: " + msg);
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -2075,13 +2074,13 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
private volatile boolean stop;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
if (stop)
throw new RuntimeException("Failing ring message worker explicitly");
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
}
@@ -2093,12 +2092,12 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
private volatile boolean stop;
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg,
+ long timeout) throws IOException, IgniteCheckedException {
if (stop)
throw new RuntimeException("Failing ring message worker explicitly");
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
if (msg instanceof TcpDiscoveryNodeAddedMessage)
stop = true;
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
index 4cf9bd0..4ef984f 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
@@ -18,12 +18,12 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
+import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.internal.util.io.GridByteArrayOutputStream;
import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.spi.IgniteSpiOperationTimeoutException;
import org.apache.ignite.spi.IgniteSpiOperationTimeoutHelper;
@@ -348,10 +348,14 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends AbstractDiscoverySelf
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout)
+ @Override protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg, long timeout)
throws IOException, IgniteCheckedException {
if (!(msg instanceof TcpDiscoveryPingRequest)) {
- super.writeToSocket(sock, msg, timeout);
+ if (countConnCheckMsg && msg instanceof TcpDiscoveryConnectionCheckMessage)
+ connCheckStatusMsgCntSent++;
+
+ super.writeToSocket(sock, out, msg, timeout);
+
return;
}
@@ -370,16 +374,7 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends AbstractDiscoverySelf
}
}
else
- super.writeToSocket(sock, msg, timeout);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
- GridByteArrayOutputStream bout, long timeout) throws IOException, IgniteCheckedException {
- if (countConnCheckMsg && msg instanceof TcpDiscoveryConnectionCheckMessage)
- connCheckStatusMsgCntSent++;
-
- super.writeToSocket(sock, msg, bout, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
/** {@inheritDoc} */
@@ -405,4 +400,4 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends AbstractDiscoverySelf
countConnCheckMsg = false;
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/5e266153/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
index dbc54bc..721192f 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java
@@ -18,6 +18,7 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
+import java.io.OutputStream;
import java.net.Socket;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
@@ -31,12 +32,12 @@ public class TestTcpDiscoverySpi extends TcpDiscoverySpi {
public boolean ignorePingResponse;
/** {@inheritDoc} */
- protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
+ protected void writeToSocket(Socket sock, OutputStream out, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
IgniteCheckedException {
if (msg instanceof TcpDiscoveryPingResponse && ignorePingResponse)
return;
else
- super.writeToSocket(sock, msg, timeout);
+ super.writeToSocket(sock, out, msg, timeout);
}
/** {@inheritDoc} */
[5/7] ignite git commit: IGNITE-2822 Fix for error in commit #da47901.
Posted by nt...@apache.org.
IGNITE-2822 Fix for error in commit #da47901.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/196954f2
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/196954f2
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/196954f2
Branch: refs/heads/ignite-2630
Commit: 196954f29c5ed96130f2e8706ff85797a7904b03
Parents: d09ef6c
Author: sboikov <sb...@gridgain.com>
Authored: Thu Apr 7 12:13:29 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Apr 7 12:13:29 2016 +0300
----------------------------------------------------------------------
.../query/continuous/CacheContinuousQueryHandler.java | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/196954f2/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
index 3576424..16513b0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java
@@ -645,9 +645,11 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
// Initial query entry or evicted entry. These events should be fired immediately.
- if (e.updateCounter() == -1L)
- return F.<CacheEntryEvent<? extends K, ? extends V>>asList(
- new CacheContinuousQueryEvent<K, V>(cache, cctx, e));
+ if (e.updateCounter() == -1L) {
+ return !e.isFiltered() ? F.<CacheEntryEvent<? extends K, ? extends V>>asList(
+ new CacheContinuousQueryEvent<K, V>(cache, cctx, e)) :
+ Collections.<CacheEntryEvent<? extends K, ? extends V>>emptyList();
+ }
PartitionRecovery rec = getOrCreatePartitionRecovery(ctx, e.partition());
@@ -683,9 +685,8 @@ public class CacheContinuousQueryHandler<K, V> implements GridContinuousHandler
}
}
}
- else if (initUpdCntrs != null) {
+ else if (initUpdCntrs != null)
partCntr = initUpdCntrs.get(partId);
- }
}
rec = new PartitionRecovery(ctx.log(getClass()), initTopVer0, partCntr);
[7/7] ignite git commit: Merge branch 'master' into ignite-2630
Posted by nt...@apache.org.
Merge branch 'master' into ignite-2630
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c180c127
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c180c127
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c180c127
Branch: refs/heads/ignite-2630
Commit: c180c127b045a9c7ac5776cd75e63e2d9d922cf1
Parents: 6688955 f970c11
Author: nikolay_tikhonov <nt...@gridgain.com>
Authored: Thu Apr 7 17:43:03 2016 +0300
Committer: nikolay_tikhonov <nt...@gridgain.com>
Committed: Thu Apr 7 17:43:03 2016 +0300
----------------------------------------------------------------------
.../discovery/GridDiscoveryManager.java | 2 +-
.../affinity/GridAffinityAssignment.java | 15 +
.../affinity/GridAffinityAssignmentCache.java | 72 +++-
.../cache/CacheAffinitySharedManager.java | 10 -
.../cache/GridCacheAffinityManager.java | 11 -
.../processors/cache/GridCacheMapEntry.java | 7 +-
.../GridCachePartitionExchangeManager.java | 16 -
.../GridDhtPartitionsExchangeFuture.java | 2 +-
.../continuous/CacheContinuousQueryHandler.java | 11 +-
.../transactions/IgniteTxLocalAdapter.java | 2 +-
.../continuous/GridContinuousProcessor.java | 13 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 46 +--
.../spi/discovery/tcp/TcpDiscoverySpi.java | 23 +-
.../affinity/AffinityHistoryCleanupTest.java | 414 +++++++++++++++++++
.../IgniteClientReconnectAbstractTest.java | 7 +-
.../BinaryObjectOffHeapUnswapTemporaryTest.java | 362 ++++++++++++++++
.../GridCacheBinaryObjectsAbstractSelfTest.java | 181 +++++---
...ffinityAssignmentNodeJoinValidationTest.java | 134 ++++++
.../tcp/TcpClientDiscoverySpiSelfTest.java | 8 +-
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 39 +-
.../TcpDiscoverySpiFailureTimeoutSelfTest.java | 23 +-
.../spi/discovery/tcp/TestTcpDiscoverySpi.java | 5 +-
.../IgniteBinaryObjectsTestSuite.java | 6 +-
.../ignite/testsuites/IgniteCacheTestSuite.java | 2 +
.../testsuites/IgniteCacheTestSuite5.java | 2 +
26 files changed, 1213 insertions(+), 202 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c180c127/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
[3/7] ignite git commit: Merge remote-tracking branch 'origin/master'
Posted by nt...@apache.org.
Merge remote-tracking branch 'origin/master'
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c173973e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c173973e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c173973e
Branch: refs/heads/ignite-2630
Commit: c173973ed2e8c5d99b01dc6c9f5a9969693f4616
Parents: 4d43553 5e26615
Author: sboikov <sb...@gridgain.com>
Authored: Thu Apr 7 10:03:36 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Apr 7 10:03:36 2016 +0300
----------------------------------------------------------------------
.../GridDhtPartitionsExchangeFuture.java | 2 +-
.../continuous/GridContinuousProcessor.java | 13 +++++-
.../ignite/spi/discovery/tcp/ClientImpl.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 44 +++++++++-----------
.../spi/discovery/tcp/TcpDiscoverySpi.java | 23 ++++------
.../IgniteClientReconnectAbstractTest.java | 7 ++--
.../tcp/TcpClientDiscoverySpiSelfTest.java | 8 ++--
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 39 +++++++++--------
.../TcpDiscoverySpiFailureTimeoutSelfTest.java | 23 ++++------
.../spi/discovery/tcp/TestTcpDiscoverySpi.java | 5 ++-
10 files changed, 80 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
[2/7] ignite git commit: gridgain-7.5.11 Affinity history cleanup
ignoring client discovery event. (cherry picked from commit
e7ab8eef504cdcf8987941e8b7a552ada451aec8)
Posted by nt...@apache.org.
gridgain-7.5.11 Affinity history cleanup ignoring client discovery event.
(cherry picked from commit e7ab8eef504cdcf8987941e8b7a552ada451aec8)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/4d435537
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/4d435537
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/4d435537
Branch: refs/heads/ignite-2630
Commit: 4d4355375497134e881188a7fcad831eb8a9013e
Parents: da47901
Author: sboikov <sb...@gridgain.com>
Authored: Thu Apr 7 10:03:09 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Apr 7 10:03:09 2016 +0300
----------------------------------------------------------------------
.../affinity/GridAffinityAssignment.java | 15 +
.../affinity/GridAffinityAssignmentCache.java | 72 +++-
.../cache/CacheAffinitySharedManager.java | 10 -
.../cache/GridCacheAffinityManager.java | 11 -
.../GridCachePartitionExchangeManager.java | 16 -
.../affinity/AffinityHistoryCleanupTest.java | 414 +++++++++++++++++++
.../ignite/testsuites/IgniteCacheTestSuite.java | 2 +
7 files changed, 486 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
index 92908cc..568e4e8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignment.java
@@ -57,6 +57,9 @@ public class GridAffinityAssignment implements Serializable {
/** */
private transient List<List<ClusterNode>> idealAssignment;
+ /** */
+ private final boolean clientEvtChange;
+
/**
* Constructs cached affinity calculations item.
*
@@ -66,6 +69,7 @@ public class GridAffinityAssignment implements Serializable {
this.topVer = topVer;
primary = new HashMap<>();
backup = new HashMap<>();
+ clientEvtChange = false;
}
/**
@@ -86,6 +90,7 @@ public class GridAffinityAssignment implements Serializable {
primary = new HashMap<>();
backup = new HashMap<>();
+ clientEvtChange = false;
initPrimaryBackupMaps();
}
@@ -101,6 +106,16 @@ public class GridAffinityAssignment implements Serializable {
idealAssignment = aff.idealAssignment;
primary = aff.primary;
backup = aff.backup;
+
+ clientEvtChange = true;
+ }
+
+ /**
+ * @return {@code True} if related discovery event did not not cause affinity assignment change and
+ * this assignment is just reference to the previous one.
+ */
+ public boolean clientEventChange() {
+ return clientEvtChange;
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index 0cacf68..a81b34d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -27,7 +27,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -48,12 +48,17 @@ import org.apache.ignite.lang.IgnitePredicate;
import org.jetbrains.annotations.Nullable;
import org.jsr166.ConcurrentHashMap8;
+import static org.apache.ignite.IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE;
+import static org.apache.ignite.IgniteSystemProperties.getInteger;
import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
/**
* Affinity cached function.
*/
public class GridAffinityAssignmentCache {
+ /** Cleanup history size. */
+ private final int MAX_HIST_SIZE = getInteger(IGNITE_AFFINITY_HISTORY_SIZE, 500);
+
/** Cache name. */
private final String cacheName;
@@ -96,6 +101,12 @@ public class GridAffinityAssignmentCache {
/** Node stop flag. */
private volatile IgniteCheckedException stopErr;
+ /** History size ignoring client events changes. */
+ private final AtomicInteger histSize = new AtomicInteger();
+
+ /** Full history size. */
+ private final AtomicInteger fullHistSize = new AtomicInteger();
+
/**
* Constructs affinity cached calculations.
*
@@ -171,6 +182,8 @@ public class GridAffinityAssignmentCache {
entry.getValue().onDone(topVer);
}
}
+
+ onHistoryAdded(assignment);
}
/**
@@ -299,6 +312,8 @@ public class GridAffinityAssignmentCache {
entry.getValue().onDone(topVer);
}
}
+
+ onHistoryAdded(assignmentCpy);
}
/**
@@ -309,22 +324,6 @@ public class GridAffinityAssignmentCache {
}
/**
- * Clean up outdated cache items.
- *
- * @param topVer Actual topology version, older versions will be removed.
- */
- public void cleanUpCache(AffinityTopologyVersion topVer) {
- if (log.isDebugEnabled())
- log.debug("Cleaning up cache for version [locNodeId=" + ctx.localNodeId() +
- ", topVer=" + topVer + ']');
-
- for (Iterator<AffinityTopologyVersion> it = affCache.keySet().iterator(); it.hasNext(); ) {
- if (it.next().compareTo(topVer) < 0)
- it.remove();
- }
- }
-
- /**
* @param topVer Topology version.
* @return Affinity assignment.
*/
@@ -530,6 +529,45 @@ public class GridAffinityAssignmentCache {
}
/**
+ * @param aff Added affinity assignment.
+ */
+ private void onHistoryAdded(GridAffinityAssignment aff) {
+ int fullSize = fullHistSize.incrementAndGet();
+
+ int size;
+
+ if (aff.clientEventChange())
+ size = histSize.get();
+ else
+ size = histSize.incrementAndGet();
+
+ int rmvCnt = size - MAX_HIST_SIZE;
+
+ if (rmvCnt <= 0) {
+ if (fullSize > MAX_HIST_SIZE * 2)
+ rmvCnt = MAX_HIST_SIZE;
+ }
+
+ if (rmvCnt > 0) {
+ Iterator<GridAffinityAssignment> it = affCache.values().iterator();
+
+ while (it.hasNext() && rmvCnt > 0) {
+ GridAffinityAssignment aff0 = it.next();
+
+ it.remove();
+
+ rmvCnt--;
+
+ if (!aff0.clientEventChange())
+ histSize.decrementAndGet();
+
+ fullHistSize.decrementAndGet();
+ }
+ }
+ }
+
+
+ /**
* Affinity ready future. Will remove itself from ready futures map.
*/
private class AffinityReadyFuture extends GridFutureAdapter<AffinityTopologyVersion> {
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index eb85288..1aedf4e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@ -925,16 +925,6 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
}
/**
- * @param topVer Actual topology version.
- */
- public void cleanUpCache(AffinityTopologyVersion topVer) {
- for (CacheHolder cache : caches.values()) {
- if (cache.client())
- cache.affinity().cleanUpCache(topVer);
- }
- }
-
- /**
* @param fut Exchange future.
* @throws IgniteCheckedException If failed.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index e264043..f1767e0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -155,17 +155,6 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
}
/**
- * Clean up outdated cache items.
- *
- * @param topVer Actual topology version, older versions will be removed.
- */
- public void cleanUpCache(AffinityTopologyVersion topVer) {
- assert !cctx.isLocal();
-
- aff.cleanUpCache(topVer);
- }
-
- /**
* @param topVer Topology version.
* @return Affinity assignments.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 2b7b621..78a8d6c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -89,9 +89,7 @@ import org.jsr166.ConcurrentHashMap8;
import org.jsr166.ConcurrentLinkedDeque8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static org.apache.ignite.IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_PRELOAD_RESEND_TIMEOUT;
-import static org.apache.ignite.IgniteSystemProperties.getInteger;
import static org.apache.ignite.IgniteSystemProperties.getLong;
import static org.apache.ignite.events.EventType.EVT_CACHE_REBALANCE_STARTED;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -109,9 +107,6 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
/** Exchange history size. */
private static final int EXCHANGE_HISTORY_SIZE = 1000;
- /** Cleanup history size. */
- public static final int EXCH_FUT_CLEANUP_HISTORY_SIZE = getInteger(IGNITE_AFFINITY_HISTORY_SIZE, 500);
-
/** Atomic reference for pending timeout object. */
private AtomicReference<ResendTimeoutObject> pendingResend = new AtomicReference<>();
@@ -962,17 +957,6 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
skipped++;
- if (skipped == EXCH_FUT_CLEANUP_HISTORY_SIZE) {
- if (err == null) {
- for (GridCacheContext cacheCtx : cctx.cacheContexts()) {
- if (!cacheCtx.isLocal())
- cacheCtx.affinity().cleanUpCache(fut.topologyVersion());
- }
-
- cctx.affinity().cleanUpCache(fut.topologyVersion());
- }
- }
-
if (skipped > 10)
fut.cleanUp();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/test/java/org/apache/ignite/cache/affinity/AffinityHistoryCleanupTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/affinity/AffinityHistoryCleanupTest.java b/modules/core/src/test/java/org/apache/ignite/cache/affinity/AffinityHistoryCleanupTest.java
new file mode 100644
index 0000000..87d1e5b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/cache/affinity/AffinityHistoryCleanupTest.java
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.affinity;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.affinity.fair.FairAffinityFunction;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class AffinityHistoryCleanupTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean client;
+
+ /** */
+ private boolean lateAffAssignment;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ CacheConfiguration[] ccfgs = new CacheConfiguration[4];
+
+ for (int i = 0; i < ccfgs.length; i++) {
+ CacheConfiguration ccfg = new CacheConfiguration();
+
+ ccfg.setName("static-cache-" + i);
+ ccfg.setAffinity(i % 2 == 0 ? new RendezvousAffinityFunction() : new FairAffinityFunction());
+
+ ccfgs[i] = ccfg;
+ }
+
+ cfg.setCacheConfiguration(ccfgs);
+
+ cfg.setClientMode(client);
+
+ cfg.setLateAffinityAssignment(lateAffAssignment);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAffinityHistoryCleanup() throws Exception {
+ String histProp = System.getProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE);
+
+ try {
+ System.setProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, "5");
+
+ Ignite ignite = startGrid(0);
+
+ checkHistory(ignite, F.asList(topVer(1, 0)), 1);
+
+ for (int i = 0; i < 3; i++) {
+ startGrid(1);
+
+ stopGrid(1);
+ }
+
+ checkHistory(ignite, F.asList(
+ topVer(3, 0),
+ topVer(4, 0),
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0)),
+ 5);
+
+ client = true;
+
+ startGrid(1);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(3, 0),
+ topVer(4, 0),
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0),
+ topVer(9, 0)),
+ 5);
+
+ startGrid(1);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(3, 0),
+ topVer(4, 0),
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0)),
+ 5);
+
+ startGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(3, 0),
+ topVer(4, 0),
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0),
+ topVer(12, 0)),
+ 5);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0),
+ topVer(12, 0),
+ topVer(13, 0)),
+ 0);
+
+ client = false;
+
+ startGrid(1);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0),
+ topVer(12, 0),
+ topVer(13, 0),
+ topVer(14, 0),
+ topVer(15, 0)),
+ 2);
+
+ startGrid(1);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0),
+ topVer(12, 0),
+ topVer(13, 0),
+ topVer(14, 0),
+ topVer(15, 0),
+ topVer(16, 0),
+ topVer(17, 0)),
+ 4);
+
+ startGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(13, 0),
+ topVer(14, 0),
+ topVer(15, 0),
+ topVer(16, 0),
+ topVer(17, 0),
+ topVer(18, 0)),
+ 5);
+
+ stopGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(14, 0),
+ topVer(15, 0),
+ topVer(16, 0),
+ topVer(17, 0),
+ topVer(18, 0),
+ topVer(19, 0)),
+ 6);
+
+ startGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(16, 0),
+ topVer(17, 0),
+ topVer(18, 0),
+ topVer(19, 0),
+ topVer(20, 0)),
+ 5);
+
+ client = true;
+
+ startGrid(2);
+
+ stopGrid(2);
+
+ checkHistory(ignite, F.asList(
+ topVer(16, 0),
+ topVer(17, 0),
+ topVer(18, 0),
+ topVer(19, 0),
+ topVer(20, 0),
+ topVer(21, 0),
+ topVer(22, 0)),
+ 5);
+ }
+ finally {
+ if (histProp != null)
+ System.setProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, histProp);
+ else
+ System.clearProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAffinityHistoryCleanupLateAffinityAssignment() throws Exception {
+ lateAffAssignment = true;
+
+ String histProp = System.getProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE);
+
+ try {
+ System.setProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, "5");
+
+ Ignite ignite = startGrid(0);
+
+ checkHistory(ignite, F.asList(topVer(1, 0)), 1);
+
+ startGrid(1);
+
+ checkHistory(ignite, F.asList(
+ topVer(1, 0),
+ topVer(2, 0),
+ topVer(2, 1)),
+ 3);
+
+ startGrid(2);
+
+ checkHistory(ignite, F.asList(
+ topVer(1, 0),
+ topVer(2, 0),
+ topVer(2, 1),
+ topVer(3, 0),
+ topVer(3, 1)),
+ 5);
+
+ startGrid(3);
+
+ checkHistory(ignite, F.asList(
+ topVer(2, 1),
+ topVer(3, 0),
+ topVer(3, 1),
+ topVer(4, 0),
+ topVer(4, 1)),
+ 5);
+
+ client = true;
+
+ startGrid(4);
+
+ stopGrid(4);
+
+ checkHistory(ignite, F.asList(
+ topVer(2, 1),
+ topVer(3, 0),
+ topVer(3, 1),
+ topVer(4, 0),
+ topVer(4, 1),
+ topVer(5, 0),
+ topVer(6, 0)),
+ 5);
+
+ startGrid(4);
+
+ stopGrid(4);
+
+ checkHistory(ignite, F.asList(
+ topVer(2, 1),
+ topVer(3, 0),
+ topVer(3, 1),
+ topVer(4, 0),
+ topVer(4, 1),
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0)),
+ 5);
+
+ startGrid(4);
+
+ stopGrid(4);
+
+ checkHistory(ignite, F.asList(
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0)),
+ 0);
+
+ client = false;
+
+ startGrid(4);
+
+ checkHistory(ignite, F.asList(
+ topVer(5, 0),
+ topVer(6, 0),
+ topVer(7, 0),
+ topVer(8, 0),
+ topVer(9, 0),
+ topVer(10, 0),
+ topVer(11, 0),
+ topVer(11, 1)),
+ 2);
+ }
+ finally {
+ if (histProp != null)
+ System.setProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE, histProp);
+ else
+ System.clearProperty(IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE);
+ }
+ }
+
+ /**
+ * @param ignite Node.
+ * @param expHist Expected history.
+ * @param expSize Expected 'non client events' history size.
+ * @throws Exception If failed.
+ */
+ private void checkHistory(Ignite ignite, List<AffinityTopologyVersion> expHist, int expSize) throws Exception {
+ awaitPartitionMapExchange();
+
+ GridCacheProcessor proc = ((IgniteKernal)ignite).context().cache();
+
+ int cnt = 0;
+
+ for (GridCacheContext cctx : proc.context().cacheContexts()) {
+ GridAffinityAssignmentCache aff = GridTestUtils.getFieldValue(cctx.affinity(), "aff");
+
+ AtomicInteger histSize = GridTestUtils.getFieldValue(aff, "histSize");
+
+ assertEquals(expSize, histSize.get());
+
+ Map<AffinityTopologyVersion, Object> cache = GridTestUtils.getFieldValue(aff, "affCache");
+
+ assertEquals("Unexpected history: " + cache.keySet(), expHist.size(), cache.size());
+
+ for (AffinityTopologyVersion topVer : expHist)
+ assertTrue("No history [ver=" + topVer + ", hist=" + cache.keySet() + ']', cache.containsKey(topVer));
+
+ cnt++;
+ }
+
+ assert cnt > 4;
+ }
+
+ /**
+ * @param major Major version.
+ * @param minor Minor version.
+ * @return Version.
+ */
+ private static AffinityTopologyVersion topVer(int major, int minor) {
+ return new AffinityTopologyVersion(major, minor);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/4d435537/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
index 75674fa..0fdf817 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
@@ -23,6 +23,7 @@ import org.apache.ignite.GridCacheAffinityBackupsSelfTest;
import org.apache.ignite.IgniteCacheAffinitySelfTest;
import org.apache.ignite.cache.IgniteWarmupClosureSelfTest;
import org.apache.ignite.cache.affinity.AffinityClientNodeSelfTest;
+import org.apache.ignite.cache.affinity.AffinityHistoryCleanupTest;
import org.apache.ignite.cache.affinity.fair.FairAffinityDynamicCacheSelfTest;
import org.apache.ignite.cache.affinity.fair.FairAffinityFunctionNodesSelfTest;
import org.apache.ignite.cache.affinity.fair.FairAffinityFunctionSelfTest;
@@ -210,6 +211,7 @@ public class IgniteCacheTestSuite extends TestSuite {
suite.addTestSuite(GridCacheAffinityBackupsSelfTest.class);
suite.addTestSuite(IgniteCacheAffinitySelfTest.class);
suite.addTestSuite(AffinityClientNodeSelfTest.class);
+ suite.addTestSuite(AffinityHistoryCleanupTest.class);
// Swap tests.
suite.addTestSuite(GridCacheSwapPreloadSelfTest.class);
[6/7] ignite git commit: ignite-2835: Fixed BinaryObjectOffHeapImpl
leakage to public code (cherry picked from e85a71)
Posted by nt...@apache.org.
ignite-2835: Fixed BinaryObjectOffHeapImpl leakage to public code (cherry picked from e85a71)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f970c11b
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f970c11b
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f970c11b
Branch: refs/heads/ignite-2630
Commit: f970c11b3be14d3888e4843aca0d8b05ccac9986
Parents: 196954f
Author: ashutak <as...@gridgain.com>
Authored: Thu Apr 7 13:49:34 2016 +0300
Committer: ashutak <as...@gridgain.com>
Committed: Thu Apr 7 13:49:34 2016 +0300
----------------------------------------------------------------------
.../processors/cache/GridCacheMapEntry.java | 7 +-
.../transactions/IgniteTxLocalAdapter.java | 2 +-
.../BinaryObjectOffHeapUnswapTemporaryTest.java | 362 +++++++++++++++++++
.../GridCacheBinaryObjectsAbstractSelfTest.java | 181 +++++++---
.../IgniteBinaryObjectsTestSuite.java | 6 +-
5 files changed, 503 insertions(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/f970c11b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index cc1a8d3..a448307 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -34,6 +34,7 @@ import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheAtomicWriteOrderMode;
import org.apache.ignite.cache.CacheMemoryMode;
import org.apache.ignite.cache.eviction.EvictableEntry;
+import org.apache.ignite.internal.binary.BinaryObjectOffheapImpl;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfo;
import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -904,9 +905,12 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- if (ret != null)
+ if (ret != null) {
+ assert tmp || !(ret instanceof BinaryObjectOffheapImpl);
+
// If return value is consistent, then done.
return retVer ? new T2<>(ret, resVer) : ret;
+ }
boolean loadedFromStore = false;
@@ -980,6 +984,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
assert ret == null || !retVer;
+ assert tmp || !(ret instanceof BinaryObjectOffheapImpl);
return ret;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f970c11b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index b1e150b..ef69ad9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1889,7 +1889,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
/*unmarshal*/true,
/*metrics*/true,
/*events*/!skipVals,
- /*temporary*/true,
+ /*temporary*/false,
CU.subjectId(IgniteTxLocalAdapter.this, cctx),
transformClo,
resolveTaskName(),
http://git-wip-us.apache.org/repos/asf/ignite/blob/f970c11b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/BinaryObjectOffHeapUnswapTemporaryTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/BinaryObjectOffHeapUnswapTemporaryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/BinaryObjectOffHeapUnswapTemporaryTest.java
new file mode 100644
index 0000000..f26191f
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/BinaryObjectOffHeapUnswapTemporaryTest.java
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.CacheMemoryMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
+import org.apache.ignite.internal.binary.BinaryObjectOffheapImpl;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED;
+import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_VALUES;
+
+/**
+ *
+ */
+public class BinaryObjectOffHeapUnswapTemporaryTest extends GridCommonAbstractTest {
+ /** */
+ private static final int CNT = 20;
+
+ /** */
+ @SuppressWarnings("serial")
+ private static final CacheEntryProcessor PROC = new CacheEntryProcessor() {
+ @Override public Object process(MutableEntry entry, Object... arguments) throws EntryProcessorException {
+ return entry.getValue();
+ }
+ };
+
+ /** */
+ private CacheAtomicityMode atomicityMode;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration c = super.getConfiguration(gridName);
+
+ c.setMarshaller(new BinaryMarshaller());
+
+ return c;
+ }
+
+ /**
+ * @param atomicityMode Atomicity mode.
+ * @param memoryMode Memory mode.
+ * @return Cache configuration.
+ */
+ private CacheConfiguration<Object, Object> cacheConfiguration(CacheAtomicityMode atomicityMode,
+ CacheMemoryMode memoryMode) {
+ this.atomicityMode = atomicityMode;
+
+ CacheConfiguration cfg = new CacheConfiguration();
+
+ cfg.setCacheMode(CacheMode.PARTITIONED);
+ cfg.setAtomicityMode(atomicityMode);
+ cfg.setMemoryMode(memoryMode);
+ cfg.setBackups(1);
+ cfg.setSwapEnabled(true);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ startGridsMultiThreaded(3);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOffheapTiered() throws Exception {
+ ignite(0).getOrCreateCache(cacheConfiguration(TRANSACTIONAL, OFFHEAP_TIERED));
+
+ try {
+ doTest();
+ }
+ finally {
+ ignite(0).destroyCache(null);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTxOffheapValues() throws Exception {
+ ignite(0).getOrCreateCache(cacheConfiguration(TRANSACTIONAL, OFFHEAP_VALUES));
+
+ try {
+ doTest();
+ }
+ finally {
+ ignite(0).destroyCache(null);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapTiered() throws Exception {
+ ignite(0).getOrCreateCache(cacheConfiguration(ATOMIC, OFFHEAP_TIERED));
+
+ try {
+ doTest();
+ }
+ finally {
+ ignite(0).destroyCache(null);
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicOffheapValues() throws Exception {
+ ignite(0).getOrCreateCache(cacheConfiguration(ATOMIC, OFFHEAP_VALUES));
+
+ try {
+ doTest();
+ }
+ finally {
+ ignite(0).destroyCache(null);
+ }
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("serial")
+ private void doTest() {
+ final IgniteCache<Integer, BinaryObject> cache = jcache(0).withKeepBinary();
+
+ for (int key = 0; key < CNT; key++)
+ jcache(0).put(key, new TestObject(key));
+
+ for (int key = CNT; key < 2 * CNT; key++) {
+ BinaryObjectBuilder builder = ignite(0).binary().builder("SomeType");
+ builder.setField("field1", key);
+ builder.setField("field2", "name_" + key);
+
+ cache.put(key, builder.build());
+ }
+
+ Set<Integer> keys = new LinkedHashSet<>();
+
+ for (int i = 0; i < 2 * CNT; i++)
+ keys.add(i);
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.get(key) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.getEntry(key).getValue() instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.getAndPut(key, cache.get(key)) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.getAndReplace(key, cache.get(key)) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.getAndPutIfAbsent(key, cache.get(key)) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.localPeek(key) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.getAndRemove(key) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ check(new IgniteInClosure<Integer>() {
+ @Override public void apply(Integer key) {
+ assertFalse(cache.invoke(key, PROC) instanceof BinaryObjectOffheapImpl);
+ }
+ });
+
+ // GetAll.
+ Map<Integer, BinaryObject> res = cache.getAll(keys);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ if (atomicityMode == TRANSACTIONAL) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = ignite(0).transactions().txStart(concurrency, isolation)) {
+ res = cache.getAll(keys);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ tx.commit();
+ }
+ }
+ }
+ }
+
+ // GetAllOutTx.
+ res = cache.getAllOutTx(keys);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ if (atomicityMode == TRANSACTIONAL) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = ignite(0).transactions().txStart(concurrency, isolation)) {
+ res = cache.getAllOutTx(keys);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ tx.commit();
+ }
+ }
+ }
+ }
+
+ // InvokeAll.
+ res = cache.invokeAll(keys, PROC);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ if (atomicityMode == TRANSACTIONAL) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = ignite(0).transactions().txStart(concurrency, isolation)) {
+ res = cache.invokeAll(keys, PROC);
+
+ for (BinaryObject val : res.values())
+ assertFalse(val instanceof BinaryObjectOffheapImpl);
+
+ tx.commit();
+ }
+ }
+ }
+ }
+
+ // GetEntries.
+ Collection<CacheEntry<Integer, BinaryObject>> entries = cache.getEntries(keys);
+
+ for (CacheEntry<Integer, BinaryObject> e : entries)
+ assertFalse(e.getValue() instanceof BinaryObjectOffheapImpl);
+
+ if (atomicityMode == TRANSACTIONAL) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = ignite(0).transactions().txStart(concurrency, isolation)) {
+ entries = cache.getEntries(keys);
+
+ for (CacheEntry<Integer, BinaryObject> e : entries)
+ assertFalse(e.getValue() instanceof BinaryObjectOffheapImpl);
+
+ tx.commit();
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ private void check(IgniteInClosure<Integer> checkOp) {
+ for (int key = 0; key < 2 * CNT; key++) {
+ checkOp.apply(key);
+
+ if (atomicityMode == TRANSACTIONAL) {
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ try (Transaction tx = ignite(0).transactions().txStart(concurrency, isolation)) {
+ checkOp.apply(key);
+
+ tx.commit();
+ }
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ *
+ */
+ @SuppressWarnings("PackageVisibleField")
+ private static class TestObject {
+ /** */
+ String field;
+
+ /** */
+ int field2;
+
+ /**
+ * @param key Key.
+ */
+ TestObject(int key) {
+ field = "str" + key;
+ field2 = key;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f970c11b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
index 36a9450..f776146 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/binary/GridCacheBinaryObjectsAbstractSelfTest.java
@@ -29,9 +29,15 @@ import javax.cache.processor.EntryProcessor;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteBinary;
+import org.apache.ignite.IgniteCache;
import org.apache.ignite.binary.BinaryNameMapper;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.binary.BinaryObjectException;
+import org.apache.ignite.binary.BinaryReader;
+import org.apache.ignite.binary.BinaryWriter;
+import org.apache.ignite.binary.Binarylizable;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
@@ -41,7 +47,9 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.binary.BinaryContext;
+import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.binary.BinaryObjectImpl;
+import org.apache.ignite.internal.binary.BinaryObjectOffheapImpl;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
@@ -49,13 +57,6 @@ import org.apache.ignite.internal.util.typedef.P2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
-import org.apache.ignite.internal.binary.BinaryMarshaller;
-import org.apache.ignite.binary.BinaryObjectBuilder;
-import org.apache.ignite.binary.BinaryObjectException;
-import org.apache.ignite.binary.Binarylizable;
-import org.apache.ignite.binary.BinaryObject;
-import org.apache.ignite.binary.BinaryReader;
-import org.apache.ignite.binary.BinaryWriter;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -421,17 +422,33 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
/**
* @throws Exception If failed.
*/
- public void testGetTx() throws Exception {
+ public void testGetTx1() throws Exception {
+ checkGetTx(PESSIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetTx2() throws Exception {
+ checkGetTx(PESSIMISTIC, READ_COMMITTED);
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ */
+ private void checkGetTx(TransactionConcurrency concurrency, TransactionIsolation isolation) {
if (atomicityMode() != TRANSACTIONAL)
return;
IgniteCache<Integer, TestObject> c = jcache(0);
+ IgniteCache<Integer, BinaryObject> kbCache = keepBinaryCache();
for (int i = 0; i < ENTRY_CNT; i++)
c.put(i, new TestObject(i));
for (int i = 0; i < ENTRY_CNT; i++) {
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
TestObject obj = c.get(i);
assertEquals(i, obj.val);
@@ -441,32 +458,75 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
}
for (int i = 0; i < ENTRY_CNT; i++) {
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
- TestObject obj = c.get(i);
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
+ BinaryObject val = kbCache.get(i);
- assertEquals(i, obj.val);
+ assertFalse("Key=" + i, val instanceof BinaryObjectOffheapImpl);
+
+ assertEquals(i, (int)val.field("val"));
+
+ kbCache.put(i, val);
tx.commit();
}
}
+ }
- IgniteCache<Integer, BinaryObject> kpc = keepBinaryCache();
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetTxAsync1() throws Exception {
+ checkGetAsyncTx(PESSIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetTxAsync2() throws Exception {
+ checkGetAsyncTx(PESSIMISTIC, READ_COMMITTED);
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ */
+ private void checkGetAsyncTx(TransactionConcurrency concurrency, TransactionIsolation isolation) {
+ if (atomicityMode() != TRANSACTIONAL)
+ return;
+
+ IgniteCache<Integer, TestObject> c = jcache(0);
+ IgniteCache<Integer, TestObject> cAsync = c.withAsync();
+ IgniteCache<Integer, BinaryObject> kbCache = keepBinaryCache();
+ IgniteCache<Integer, BinaryObject> kbCacheAsync = kbCache.withAsync();
+
+ for (int i = 0; i < ENTRY_CNT; i++)
+ c.put(i, new TestObject(i));
for (int i = 0; i < ENTRY_CNT; i++) {
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
- BinaryObject po = kpc.get(i);
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
+ cAsync.get(i);
- assertEquals(i, (int)po.field("val"));
+ TestObject obj = (TestObject)cAsync.future().get();
+
+ assertEquals(i, obj.val);
tx.commit();
}
}
for (int i = 0; i < ENTRY_CNT; i++) {
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
- BinaryObject po = kpc.get(i);
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
+ kbCacheAsync.get(i);
- assertEquals(i, (int)po.field("val"));
+ BinaryObject val = (BinaryObject)kbCacheAsync.future().get();
+
+ assertFalse("Key=" + i, val instanceof BinaryObjectOffheapImpl);
+
+ assertEquals(i, (int)val.field("val"));
+
+ kbCacheAsync.put(i, val);
+
+ kbCacheAsync.future().get();
tx.commit();
}
@@ -606,11 +666,27 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
/**
* @throws Exception If failed.
*/
- public void testGetAllTx() throws Exception {
+ public void testGetAllTx1() throws Exception {
+ checkGetAllTx(PESSIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetAllTx2() throws Exception {
+ checkGetAllTx(PESSIMISTIC, READ_COMMITTED);
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ */
+ private void checkGetAllTx(TransactionConcurrency concurrency, TransactionIsolation isolation) {
if (atomicityMode() != TRANSACTIONAL)
return;
IgniteCache<Integer, TestObject> c = jcache(0);
+ IgniteCache<Integer, BinaryObject> kpc = keepBinaryCache();
for (int i = 0; i < ENTRY_CNT; i++)
c.put(i, new TestObject(i));
@@ -621,18 +697,7 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
for (int j = 0; j < 10; j++)
keys.add(i++);
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
- Map<Integer, TestObject> objs = c.getAll(keys);
-
- assertEquals(10, objs.size());
-
- for (Map.Entry<Integer, TestObject> e : objs.entrySet())
- assertEquals(e.getKey().intValue(), e.getValue().val);
-
- tx.commit();
- }
-
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
Map<Integer, TestObject> objs = c.getAll(keys);
assertEquals(10, objs.size());
@@ -644,32 +709,26 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
}
}
- IgniteCache<Integer, BinaryObject> kpc = keepBinaryCache();
-
for (int i = 0; i < ENTRY_CNT; ) {
Set<Integer> keys = new HashSet<>();
for (int j = 0; j < 10; j++)
keys.add(i++);
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
Map<Integer, BinaryObject> objs = kpc.getAll(keys);
assertEquals(10, objs.size());
- for (Map.Entry<Integer, BinaryObject> e : objs.entrySet())
- assertEquals(new Integer(e.getKey().intValue()), e.getValue().field("val"));
-
- tx.commit();
- }
+ for (Map.Entry<Integer, BinaryObject> e : objs.entrySet()) {
+ BinaryObject val = e.getValue();
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
- Map<Integer, BinaryObject> objs = kpc.getAll(keys);
+ assertEquals(new Integer(e.getKey().intValue()), val.field("val"));
- assertEquals(10, objs.size());
+ kpc.put(e.getKey(), val);
- for (Map.Entry<Integer, BinaryObject> e : objs.entrySet())
- assertEquals(new Integer(e.getKey().intValue()), e.getValue().field("val"));
+ assertFalse("Key=" + i, val instanceof BinaryObjectOffheapImpl);
+ }
tx.commit();
}
@@ -679,7 +738,22 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
/**
* @throws Exception If failed.
*/
- public void testGetAllAsyncTx() throws Exception {
+ public void testGetAllAsyncTx1() throws Exception {
+ checkGetAllAsyncTx(PESSIMISTIC, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testGetAllAsyncTx2() throws Exception {
+ checkGetAllAsyncTx(PESSIMISTIC, READ_COMMITTED);
+ }
+
+ /**
+ * @param concurrency Concurrency.
+ * @param isolation Isolation.
+ */
+ private void checkGetAllAsyncTx(TransactionConcurrency concurrency, TransactionIsolation isolation) {
if (atomicityMode() != TRANSACTIONAL)
return;
@@ -695,7 +769,7 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
for (int j = 0; j < 10; j++)
keys.add(i++);
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
cacheAsync.getAll(keys);
Map<Integer, TestObject> objs = cacheAsync.<Map<Integer, TestObject>>future().get();
@@ -719,15 +793,20 @@ public abstract class GridCacheBinaryObjectsAbstractSelfTest extends GridCommonA
IgniteCache<Integer, BinaryObject> asyncCache = cache.withAsync();
- try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ try (Transaction tx = grid(0).transactions().txStart(concurrency, isolation)) {
asyncCache.getAll(keys);
Map<Integer, BinaryObject> objs = asyncCache.<Map<Integer, BinaryObject>>future().get();
assertEquals(10, objs.size());
- for (Map.Entry<Integer, BinaryObject> e : objs.entrySet())
- assertEquals(new Integer(e.getKey().intValue()), e.getValue().field("val"));
+ for (Map.Entry<Integer, BinaryObject> e : objs.entrySet()) {
+ BinaryObject val = e.getValue();
+
+ assertEquals(new Integer(e.getKey().intValue()), val.field("val"));
+
+ assertFalse("Key=" + e.getKey(), val instanceof BinaryObjectOffheapImpl);
+ }
tx.commit();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/f970c11b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
index 043ea06..cedf9a7 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBinaryObjectsTestSuite.java
@@ -44,11 +44,12 @@ import org.apache.ignite.internal.binary.noncompact.BinaryMarshallerNonCompactSe
import org.apache.ignite.internal.binary.noncompact.BinaryObjectBuilderAdditionalNonCompactSelfTest;
import org.apache.ignite.internal.binary.noncompact.BinaryObjectBuilderNonCompactDefaultMappersSelfTest;
import org.apache.ignite.internal.binary.noncompact.BinaryObjectBuilderNonCompactSimpleNameLowerCaseMappersSelfTest;
+import org.apache.ignite.internal.binary.streams.BinaryHeapStreamByteOrderSelfTest;
+import org.apache.ignite.internal.binary.streams.BinaryOffheapStreamByteOrderSelfTest;
+import org.apache.ignite.internal.processors.cache.BinaryObjectOffHeapUnswapTemporaryTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryObjectUserClassloaderSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryStoreBinariesDefaultMappersSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryStoreBinariesSimpleNameMappersSelfTest;
-import org.apache.ignite.internal.binary.streams.BinaryHeapStreamByteOrderSelfTest;
-import org.apache.ignite.internal.binary.streams.BinaryOffheapStreamByteOrderSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheBinaryStoreObjectsSelfTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheClientNodeBinaryObjectMetadataMultinodeTest;
import org.apache.ignite.internal.processors.cache.binary.GridCacheClientNodeBinaryObjectMetadataTest;
@@ -117,6 +118,7 @@ public class IgniteBinaryObjectsTestSuite extends TestSuite {
suite.addTestSuite(GridCacheBinaryObjectsPartitionedNearDisabledSelfTest.class);
suite.addTestSuite(GridCacheBinaryObjectsAtomicSelfTest.class);
suite.addTestSuite(GridCacheBinaryObjectsAtomicNearDisabledSelfTest.class);
+ suite.addTestSuite(BinaryObjectOffHeapUnswapTemporaryTest.class);
suite.addTestSuite(GridCacheBinaryObjectsLocalOffheapTieredSelfTest.class);
suite.addTestSuite(GridCacheBinaryObjectsAtomicOffheapTieredSelfTest.class);
[4/7] ignite git commit: Added test.
Posted by nt...@apache.org.
Added test.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/d09ef6c0
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/d09ef6c0
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/d09ef6c0
Branch: refs/heads/ignite-2630
Commit: d09ef6c0f6acace146cf4012dc40b4de33fb6f61
Parents: c173973
Author: sboikov <sb...@gridgain.com>
Authored: Thu Apr 7 10:26:00 2016 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Apr 7 10:26:00 2016 +0300
----------------------------------------------------------------------
.../discovery/GridDiscoveryManager.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 2 +-
...ffinityAssignmentNodeJoinValidationTest.java | 134 +++++++++++++++++++
.../testsuites/IgniteCacheTestSuite5.java | 2 +
4 files changed, 138 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/d09ef6c0/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 7f02498..dc664a8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -1066,7 +1066,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
boolean rmtLateAssign;
- if (n.version().compareTo(CacheAffinitySharedManager.LATE_AFF_ASSIGN_SINCE) >= 0)
+ if (n.version().compareToIgnoreTimestamp(CacheAffinitySharedManager.LATE_AFF_ASSIGN_SINCE) >= 0)
rmtLateAssign = n.attribute(ATTR_LATE_AFFINITY_ASSIGNMENT);
else
rmtLateAssign = false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/d09ef6c0/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
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 27a31c4..75430c6 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
@@ -3260,7 +3260,7 @@ class ServerImpl extends TcpDiscoveryImpl {
boolean rmtLateAssignBool;
- if (node.version().compareTo(CacheAffinitySharedManager.LATE_AFF_ASSIGN_SINCE) >= 0) {
+ if (node.version().compareToIgnoreTimestamp(CacheAffinitySharedManager.LATE_AFF_ASSIGN_SINCE) >= 0) {
Boolean rmtLateAssign = node.attribute(ATTR_LATE_AFFINITY_ASSIGNMENT);
// Can be null only in tests.
rmtLateAssignBool = rmtLateAssign != null ? rmtLateAssign : false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/d09ef6c0/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentNodeJoinValidationTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentNodeJoinValidationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentNodeJoinValidationTest.java
new file mode 100644
index 0000000..247669b
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentNodeJoinValidationTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class CacheLateAffinityAssignmentNodeJoinValidationTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private boolean lateAff;
+
+ /** */
+ private boolean client;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setLateAffinityAssignment(lateAff);
+
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+ cfg.setClientMode(client);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJoinValidation1() throws Exception {
+ checkNodeJoinValidation(false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testJoinValidation2() throws Exception {
+ checkNodeJoinValidation(true);
+ }
+
+ /**
+ * @param firstEnabled Flag value for first started node.
+ * @throws Exception If failed.
+ */
+ public void checkNodeJoinValidation(boolean firstEnabled) throws Exception {
+ lateAff = firstEnabled;
+
+ Ignite ignite = startGrid(0);
+
+ assertFalse(ignite.configuration().isClientMode());
+
+ lateAff = !firstEnabled;
+
+ try {
+ startGrid(1);
+
+ fail();
+ }
+ catch (Exception e) {
+ checkError(e);
+ }
+
+ client = true;
+
+ try {
+ startGrid(1);
+
+ fail();
+ }
+ catch (Exception e) {
+ checkError(e);
+ }
+
+ assertEquals(1, ignite.cluster().nodes().size());
+
+ lateAff = firstEnabled;
+
+ client = false;
+
+ startGrid(1);
+
+ client = true;
+
+ Ignite client = startGrid(2);
+
+ assertTrue(client.configuration().isClientMode());
+ }
+
+ /**
+ * @param e Error.
+ */
+ private void checkError(Exception e) {
+ IgniteSpiException err = X.cause(e, IgniteSpiException.class);
+
+ assertNotNull(err);
+ assertTrue(err.getMessage().contains("Local node's cache affinity assignment mode differs " +
+ "from the same property on remote node"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/d09ef6c0/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
index 0c24425..35c599e 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite5.java
@@ -24,6 +24,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheSwapSpaceSpiConsiste
import org.apache.ignite.internal.processors.cache.IgniteCachePutStackOverflowSelfTest;
import org.apache.ignite.internal.processors.cache.IgniteCacheStoreCollectionTest;
import org.apache.ignite.internal.processors.cache.distributed.CacheLateAffinityAssignmentFairAffinityTest;
+import org.apache.ignite.internal.processors.cache.distributed.CacheLateAffinityAssignmentNodeJoinValidationTest;
import org.apache.ignite.internal.processors.cache.distributed.CacheLateAffinityAssignmentTest;
import org.apache.ignite.internal.processors.cache.store.IgniteCacheWriteBehindNoUpdateSelfTest;
@@ -47,6 +48,7 @@ public class IgniteCacheTestSuite5 extends TestSuite {
suite.addTestSuite(CacheLateAffinityAssignmentTest.class);
suite.addTestSuite(CacheLateAffinityAssignmentFairAffinityTest.class);
+ suite.addTestSuite(CacheLateAffinityAssignmentNodeJoinValidationTest.class);
return suite;
}