You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/05/26 16:49:23 UTC
[1/7] incubator-ignite git commit: # IGNITE-709 Fix
GridEventConsumeSelfTest.testMasterNodeLeave()
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-23 6432ec0e9 -> 9347e6798
# IGNITE-709 Fix GridEventConsumeSelfTest.testMasterNodeLeave()
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/12aa8ab7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/12aa8ab7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/12aa8ab7
Branch: refs/heads/ignite-23
Commit: 12aa8ab7fb107f5fa8067eb36cfc1aee930dbae9
Parents: 092e8c7
Author: sevdokimov <se...@gridgain.com>
Authored: Tue May 26 12:22:35 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Tue May 26 12:22:35 2015 +0300
----------------------------------------------------------------------
.../internal/processors/continuous/GridEventConsumeSelfTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/12aa8ab7/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
index 223c082..2c9e513 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/GridEventConsumeSelfTest.java
@@ -867,7 +867,7 @@ public class GridEventConsumeSelfTest extends GridCommonAbstractTest {
return true;
}
- }, EVT_NODE_LEFT);
+ }, EVT_NODE_LEFT, EVT_NODE_FAILED);
}
g.events().remoteListen(
[3/7] incubator-ignite git commit: IGNITE-709 Fix tests.
Posted by sb...@apache.org.
IGNITE-709 Fix tests.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/e575164b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/e575164b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/e575164b
Branch: refs/heads/ignite-23
Commit: e575164bbfb8f4f5b7c56c3470b234fff52cf8a3
Parents: d59a712
Author: sevdokimov <se...@gridgain.com>
Authored: Tue May 26 14:41:18 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Tue May 26 14:41:18 2015 +0300
----------------------------------------------------------------------
.../internal/GridEventConsumeHandler.java | 106 ++++++++++---------
.../continuous/GridContinuousProcessor.java | 37 ++++++-
.../apache/ignite/internal/GridSelfTest.java | 4 +-
3 files changed, 93 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e575164b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
index bb8366a..f33fa39 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridEventConsumeHandler.java
@@ -137,80 +137,86 @@ class GridEventConsumeHandler implements GridContinuousHandler {
private boolean notificationInProgress;
@Override public void onEvent(Event evt) {
- synchronized (notificationQueue) {
- notificationQueue.add(new T3<>(nodeId, routineId, evt));
+ if (filter != null && !filter.apply(evt))
+ return;
- if (!notificationInProgress) {
- ctx.getSystemExecutorService().submit(new Runnable() {
- @Override public void run() {
- while (true) {
- T3<UUID, UUID, Event> t3;
-
- synchronized (notificationQueue) {
- t3 = notificationQueue.poll();
+ if (loc) {
+ if (!cb.apply(nodeId, evt))
+ ctx.continuous().stopRoutine(routineId);
+ }
+ else {
+ if (ctx.discovery().node(nodeId) == null)
+ return;
- if (t3 == null) {
- notificationInProgress = false;
+ synchronized (notificationQueue) {
+ notificationQueue.add(new T3<>(nodeId, routineId, evt));
- return;
- }
- }
+ if (!notificationInProgress) {
+ ctx.getSystemExecutorService().submit(new Runnable() {
+ @Override public void run() {
+ if (!ctx.continuous().lockStopping())
+ return;
try {
- Event evt = t3.get3();
+ while (true) {
+ T3<UUID, UUID, Event> t3;
- if (filter != null && !filter.apply(evt))
- continue;
+ synchronized (notificationQueue) {
+ t3 = notificationQueue.poll();
- if (loc) {
- if (!cb.apply(nodeId, evt)) {
- ctx.continuous().stopRoutine(routineId);
+ if (t3 == null) {
+ notificationInProgress = false;
- return;
+ return;
+ }
}
- continue;
- }
+ try {
+ Event evt = t3.get3();
- ClusterNode node = ctx.discovery().node(t3.get1());
+ EventWrapper wrapper = new EventWrapper(evt);
- if (node == null)
- continue;
+ if (evt instanceof CacheEvent) {
+ String cacheName = ((CacheEvent)evt).cacheName();
- EventWrapper wrapper = new EventWrapper(evt);
+ ClusterNode node = ctx.discovery().node(t3.get1());
- if (evt instanceof CacheEvent) {
- String cacheName = ((CacheEvent)evt).cacheName();
+ if (node == null)
+ continue;
- if (ctx.config().isPeerClassLoadingEnabled()
- && ctx.discovery().cacheNode(node, cacheName)) {
- wrapper.p2pMarshal(ctx.config().getMarshaller());
+ if (ctx.config().isPeerClassLoadingEnabled()
+ && ctx.discovery().cacheNode(node, cacheName)) {
+ wrapper.p2pMarshal(ctx.config().getMarshaller());
- wrapper.cacheName = cacheName;
+ wrapper.cacheName = cacheName;
- GridCacheDeploymentManager depMgr =
- ctx.cache().internalCache(cacheName).context().deploy();
+ GridCacheDeploymentManager depMgr = ctx.cache()
+ .internalCache(cacheName).context().deploy();
- depMgr.prepare(wrapper);
+ depMgr.prepare(wrapper);
+ }
+ }
+
+ ctx.continuous().addNotification(t3.get1(), t3.get2(), wrapper, null, false,
+ false);
+ }
+ catch (ClusterTopologyCheckedException ignored) {
+ // No-op.
+ }
+ catch (Throwable e) {
+ U.error(ctx.log(GridEventConsumeHandler.class),
+ "Failed to send event notification to node: " + nodeId, e);
}
}
-
- ctx.continuous().addNotification(t3.get1(), t3.get2(), wrapper, null,
- false, false);
- }
- catch (ClusterTopologyCheckedException
- | IgniteInterruptedCheckedException ignored) {
- // No-op.
}
- catch (Throwable e) {
- U.error(ctx.log(GridEventConsumeHandler.class),
- "Failed to send event notification to node: " + nodeId, e);
+ finally {
+ ctx.continuous().unlockStopping();
}
}
- }
- });
+ });
- notificationInProgress = true;
+ notificationInProgress = true;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e575164b/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 71a2a66..67b32a6 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
@@ -93,7 +93,10 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
private int retryCnt = 3;
/** */
- private volatile boolean processorStopped;
+ private final ReentrantReadWriteLock processorStopLock = new ReentrantReadWriteLock();
+
+ /** */
+ private boolean processorStopped;
/**
* @param ctx Kernal context.
@@ -259,9 +262,39 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
log.debug("Continuous processor started.");
}
+ /**
+ * @return {@code true} if lock successful, {@code false} if processor already stopped.
+ */
+ @SuppressWarnings("LockAcquiredButNotSafelyReleased")
+ public boolean lockStopping() {
+ processorStopLock.readLock().lock();
+
+ if (processorStopped) {
+ processorStopLock.readLock().unlock();
+
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ *
+ */
+ public void unlockStopping() {
+ processorStopLock.readLock().unlock();
+ }
+
/** {@inheritDoc} */
@Override public void onKernalStop(boolean cancel) {
- processorStopped = true;
+ processorStopLock.writeLock().lock();
+
+ try {
+ processorStopped = true;
+ }
+ finally {
+ processorStopLock.writeLock().unlock();
+ }
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e575164b/modules/core/src/test/java/org/apache/ignite/internal/GridSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridSelfTest.java
index b4dce6c..7f5ee54 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/GridSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridSelfTest.java
@@ -118,8 +118,8 @@ public class GridSelfTest extends GridProjectionAbstractTest {
g.message().remoteListen(null, new MessagingListenActor<String>() {
@Override protected void receive(UUID nodeId, String rcvMsg) throws Throwable {
- assert locNodeId.equals(nodeId);
- assert msg.equals(rcvMsg);
+ assertEquals(locNodeId, nodeId);
+ assertEquals(msg, rcvMsg);
stop(rcvMsg);
}
[5/7] incubator-ignite git commit: # IGNITE-709 Fix test: use
Cluster.localNode() instead of Cluster.node()
Posted by sb...@apache.org.
# IGNITE-709 Fix test: use Cluster.localNode() instead of Cluster.node()
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/2ca18baa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/2ca18baa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/2ca18baa
Branch: refs/heads/ignite-23
Commit: 2ca18baa6144353f2b0991b1809e6cb634a5de97
Parents: 2e52adf
Author: sevdokimov <se...@gridgain.com>
Authored: Tue May 26 15:40:15 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Tue May 26 15:40:15 2015 +0300
----------------------------------------------------------------------
.../managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2ca18baa/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java
index 0af90c1..03fc01b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java
@@ -264,7 +264,7 @@ public class GridDiscoveryManagerAliveCacheSelfTest extends GridCommonAbstractTe
for (Iterator<Ignite> itr = toRmv.iterator(); itr.hasNext(); ) {
Ignite g = itr.next();
- if (g.cluster().node().isClient()) {
+ if (g.cluster().localNode().isClient()) {
G.stop(g.name(), false);
itr.remove();
[2/7] incubator-ignite git commit: IGNITE-709 Fix NPE
Posted by sb...@apache.org.
IGNITE-709 Fix NPE
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/d59a712a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/d59a712a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/d59a712a
Branch: refs/heads/ignite-23
Commit: d59a712ad75c306ac43f37a9a77401f80821d3a5
Parents: 12aa8ab
Author: sevdokimov <se...@gridgain.com>
Authored: Tue May 26 13:43:12 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Tue May 26 13:43:12 2015 +0300
----------------------------------------------------------------------
.../processors/continuous/GridContinuousProcessor.java | 11 +++++++++++
1 file changed, 11 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d59a712a/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 d67a45a..71a2a66 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
@@ -92,6 +92,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
/** Number of retries using to send messages. */
private int retryCnt = 3;
+ /** */
+ private volatile boolean processorStopped;
+
/**
* @param ctx Kernal context.
*/
@@ -257,6 +260,11 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
}
/** {@inheritDoc} */
+ @Override public void onKernalStop(boolean cancel) {
+ processorStopped = true;
+ }
+
+ /** {@inheritDoc} */
@Override public void stop(boolean cancel) throws IgniteCheckedException {
if (ctx.config().isDaemon())
return;
@@ -551,6 +559,9 @@ public class GridContinuousProcessor extends GridProcessorAdapter {
assert !nodeId.equals(ctx.localNodeId());
+ if (processorStopped)
+ return;
+
RemoteRoutineInfo info = rmtInfos.get(routineId);
if (info != null) {
[4/7] incubator-ignite git commit: # IGNITE-709 Introduce
TcpDiscoverySpiAdapter.injectables()
Posted by sb...@apache.org.
# IGNITE-709 Introduce TcpDiscoverySpiAdapter.injectables()
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/2e52adf6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/2e52adf6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/2e52adf6
Branch: refs/heads/ignite-23
Commit: 2e52adf6a59923695ae444e917cb3283a200a641
Parents: e575164
Author: sevdokimov <se...@gridgain.com>
Authored: Tue May 26 15:20:46 2015 +0300
Committer: sevdokimov <se...@gridgain.com>
Committed: Tue May 26 15:20:46 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java | 5 -----
.../org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 5 -----
.../apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java | 5 +++++
3 files changed, 5 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2e52adf6/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
index de954d8..1cf044e 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpi.java
@@ -390,11 +390,6 @@ public class TcpClientDiscoverySpi extends TcpDiscoverySpiAdapter implements Tcp
}
/** {@inheritDoc} */
- @Override public Collection<Object> injectables() {
- return F.<Object>asList(ipFinder);
- }
-
- /** {@inheritDoc} */
@Override public Collection<ClusterNode> getRemoteNodes() {
return U.arrayList(rmtNodes.values(), TcpDiscoveryNodesRing.VISIBLE_NODES);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2e52adf6/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 34e1ca8..6bad0f2 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
@@ -646,11 +646,6 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov
}
/** {@inheritDoc} */
- @Override public Collection<Object> injectables() {
- return F.<Object>asList(ipFinder);
- }
-
- /** {@inheritDoc} */
@Override public void spiStart(String gridName) throws IgniteSpiException {
spiStart0(false);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/2e52adf6/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
index d922761..5bfd378 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiAdapter.java
@@ -355,6 +355,11 @@ public abstract class TcpDiscoverySpiAdapter extends IgniteSpiAdapter implements
}
/** {@inheritDoc} */
+ @Override public Collection<Object> injectables() {
+ return F.<Object>asList(ipFinder);
+ }
+
+ /** {@inheritDoc} */
@Override protected void onContextInitialized0(IgniteSpiContext spiCtx) throws IgniteSpiException {
super.onContextInitialized0(spiCtx);
[6/7] incubator-ignite git commit: # ignite-23
Posted by sb...@apache.org.
# ignite-23
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/32cb360f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/32cb360f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/32cb360f
Branch: refs/heads/ignite-23
Commit: 32cb360f736cdc84af8d461ad4c49cc2806cf996
Parents: 6432ec0
Author: sboikov <sb...@gridgain.com>
Authored: Tue May 26 15:49:12 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue May 26 17:46:53 2015 +0300
----------------------------------------------------------------------
.../dht/atomic/GridDhtAtomicCache.java | 2 +-
.../GridDhtPartitionsExchangeFuture.java | 2 -
...niteCacheClientNodeChangingTopologyTest.java | 431 ++++++++++++++++---
3 files changed, 380 insertions(+), 55 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 9135f0a..69f5501 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1062,7 +1062,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (ver == null) {
// Assign next version for update inside entries lock.
- ver = ctx.versions().next(req.topologyVersion());
+ ver = ctx.versions().next(topology().topologyVersion());
if (hasNear)
res.nearVersion(ver);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/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 e04432f..1c1ebd5 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
@@ -514,8 +514,6 @@ public class GridDhtPartitionsExchangeFuture extends GridFutureAdapter<AffinityT
if (exchId.isLeft())
cctx.mvcc().removeExplicitNodeLocks(exchId.nodeId(), exchId.topologyVersion());
- cctx.mvcc().finishLocks(exchId.topologyVersion()).get();
-
onDone(exchId.topologyVersion());
skipPreload = true;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/32cb360f/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
index b067797..45fa275 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -211,7 +211,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
putFut.get();
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
ignite3.close();
@@ -248,7 +248,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
putFut.get();
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
for (int i = 0; i < KEYS; i++)
map.put(i, i + 2);
@@ -258,7 +258,106 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
else
cache.put(0, 2);
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicNoRemapClockMode() throws Exception {
+ atomicNoRemap(CLOCK);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testAtomicNoRemapPrimaryMode() throws Exception {
+ atomicNoRemap(PRIMARY);
+ }
+
+ /**
+ * @param writeOrder Write order.
+ * @throws Exception If failed.
+ */
+ private void atomicNoRemap(CacheAtomicWriteOrderMode writeOrder) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(ATOMIC);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAtomicWriteOrderMode(writeOrder);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ map.put(primaryKey(ignite0.cache(null)), 0);
+ map.put(primaryKey(ignite1.cache(null)), 1);
+ map.put(primaryKey(ignite2.cache(null)), 2);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ // Block messages requests for both nodes.
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearAtomicUpdateRequest.class, ignite2.localNode().id());
+
+ spi.record(GridNearAtomicUpdateRequest.class);
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ assertEquals(writeOrder, cache.getConfiguration(CacheConfiguration.class).getAtomicWriteOrderMode());
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ cache.putAll(map);
+
+ return null;
+ }
+ });
+
+ IgniteEx ignite4 = startGrid(4);
+
+ assertTrue(ignite4.configuration().isClientMode());
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ spi.record(null);
+
+ checkData(map, null, cache, 5);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertEquals(3, msgs.size());
+
+ for (Object msg : msgs)
+ assertTrue(((GridNearAtomicUpdateRequest)msg).clientRequest());
+
+ map.put(primaryKey(ignite0.cache(null)), 3);
+ map.put(primaryKey(ignite1.cache(null)), 4);
+ map.put(primaryKey(ignite2.cache(null)), 5);
+
+ cache.putAll(map);
+
+ checkData(map, null, cache, 5);
}
/**
@@ -334,7 +433,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
Integer old = putFut.get();
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
assertEquals((Object)0, old);
}
@@ -394,7 +493,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
putFut.get();
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
map.clear();
@@ -403,7 +502,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
cache.putAll(map);
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
}
/**
* @throws Exception If failed.
@@ -452,6 +551,8 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+ spi.record(GridNearLockRequest.class);
+
final IgniteCache<Integer, Integer> cache = ignite2.cache(null);
IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
@@ -480,7 +581,17 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
putFut.get();
- checkData(map, cache, 4);
+ spi.record(null);
+
+ checkData(map, null, cache, 4);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ assertTrue(((GridNearLockRequest)msgs.get(0)).firstClientRequest());
+ assertTrue(((GridNearLockRequest)msgs.get(1)).firstClientRequest());
+
+ for (int i = 2; i < msgs.size(); i++)
+ assertFalse(((GridNearLockRequest)msgs.get(i)).firstClientRequest());
ignite3.close();
@@ -513,7 +624,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
putFut.get();
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
for (int i = 0; i < 100; i++)
map.put(i, i + 2);
@@ -524,7 +635,192 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
tx.commit();
}
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
+ }
+
+ /**
+ * Tests specific scenario when mapping for first locked keys does not change, but changes for second one.
+ *
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTx2() throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ final Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ AffinityTopologyVersion topVer1 = new AffinityTopologyVersion(4, 0);
+
+ assertEquals(topVer1, ignite0.context().cache().internalCache(null).context().topology().topologyVersion());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ final Integer key1 = 0;
+ final Integer key2 = 7;
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id());
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.put(key1, 1);
+ cache.put(key2, 2);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ client = false;
+
+ IgniteEx ignite4 = startGrid(4);
+
+ AffinityTopologyVersion topVer2 = new AffinityTopologyVersion(5, 0);
+
+ assertEquals(topVer2, ignite0.context().cache().internalCache(null).context().topology().topologyVersion());
+
+ GridCacheAffinityManager aff = ignite0.context().cache().internalCache(null).context().affinity();
+
+ List<ClusterNode> nodes1 = aff.nodes(key1, topVer1);
+ List<ClusterNode> nodes2 = aff.nodes(key1, topVer2);
+
+ assertEquals(nodes1, nodes2);
+
+ nodes1 = aff.nodes(key2, topVer1);
+ nodes2 = aff.nodes(key2, topVer2);
+
+ assertFalse(nodes1.get(0).equals(nodes2.get(0)));
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ checkData(F.asMap(key1, 1, key2, 2), null, cache, 5);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxNearEnabledNoRemap() throws Exception {
+ pessimisticTxNoRemap(new NearCacheConfiguration());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTxNoRemap() throws Exception {
+ pessimisticTxNoRemap(null);
+ }
+
+ /**
+ * @param nearCfg Near cache configuration.
+ * @throws Exception If failed.
+ */
+ private void pessimisticTxNoRemap(@Nullable NearCacheConfiguration nearCfg) throws Exception {
+ ccfg = new CacheConfiguration();
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(1);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+ ccfg.setNearConfiguration(nearCfg);
+
+ IgniteEx ignite0 = startGrid(0);
+ IgniteEx ignite1 = startGrid(1);
+ IgniteEx ignite2 = startGrid(2);
+
+ client = true;
+
+ final Ignite ignite3 = startGrid(3);
+
+ assertTrue(ignite3.configuration().isClientMode());
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+
+ final Map<Integer, Integer> map = new HashMap<>();
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i);
+
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite1.localNode().id());
+ spi.blockMessages(GridNearLockRequest.class, ignite2.localNode().id());
+
+ spi.record(GridNearLockRequest.class);
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(null);
+
+ IgniteInternalFuture<?> putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ for (Map.Entry<Integer, Integer> e : map.entrySet())
+ cache.put(e.getKey(), e.getValue());
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ IgniteEx ignite4 = startGrid(4);
+
+ assertTrue(ignite4.configuration().isClientMode());
+
+ assertFalse(putFut.isDone());
+
+ log.info("Stop block.");
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ spi.record(null);
+
+ checkData(map, null, cache, 5);
+
+ List<Object> msgs = spi.recordedMessages();
+
+ checkClientLockMessages(msgs, map.size());
+
+ for (int i = 0; i < 100; i++)
+ map.put(i, i + 1);
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ checkData(map, null, cache, 5);
}
/**
@@ -775,7 +1071,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
checkClientPrepareMessages(spi.recordedMessages(), 6);
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
cache.putAll(map);
@@ -783,7 +1079,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
spi.record(null);
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
IgniteCache<Integer, Integer> cache0 = ignite0.cache(null);
@@ -802,7 +1098,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
for (Object msg : msgs)
assertFalse(((GridNearTxPrepareRequest)msg).firstClientRequest());
- checkData(map, cache, 4);
+ checkData(map, null, cache, 4);
}
/**
@@ -944,11 +1240,15 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
/**
* @param map Expected data.
+ * @param keys Expected keys (if expected data is not specified).
* @param clientCache Client cache.
* @param expNodes Expected nodes number.
* @throws Exception If failed.
*/
- private void checkData(final Map<Integer, Integer> map, IgniteCache<?, ?> clientCache, final int expNodes)
+ private void checkData(final Map<Integer, Integer> map,
+ final Set<Integer> keys,
+ IgniteCache<?, ?> clientCache,
+ final int expNodes)
throws Exception
{
final List<Ignite> nodes = G.allGrids();
@@ -964,18 +1264,26 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
boolean wait = GridTestUtils.waitForCondition(new PA() {
@Override public boolean apply() {
try {
- for (Map.Entry<Integer, Integer> e : map.entrySet()) {
- Integer key = e.getKey();
+ Set<Integer> keys0 = map != null ? map.keySet() : keys;
+ assertNotNull(keys0);
+
+ for (Integer key : keys0) {
GridCacheVersion ver = null;
+ Object val = null;
for (Ignite node : nodes) {
IgniteCache<Integer, Integer> cache = node.cache(null);
boolean affNode = aff.isPrimaryOrBackup(node.cluster().localNode(), key);
+ Object val0 = cache.localPeek(key);
+
if (affNode || node == nearCacheNode) {
- assertEquals("Unexpected value for " + node.name(), e.getValue(), cache.localPeek(key));
+ if (map != null)
+ assertEquals("Unexpected value for " + node.name(), map.get(key), val0);
+ else
+ assertNotNull("Unexpected value for " + node.name(), val0);
GridCacheAdapter cache0 = ((IgniteKernal)node).internalCache(null);
@@ -991,13 +1299,28 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
assertNotNull("Null version [node=" + node.name() + ", key=" + key + ']', ver0);
- if (ver == null)
+ if (ver == null) {
ver = ver0;
- else
- assertEquals(ver0, ver);
+ val = val0;
+ }
+ else {
+ assertEquals("Version check failed [node=" + node.name() +
+ ", key=" + key +
+ ", affNode=" + affNode +
+ ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']',
+ ver0,
+ ver);
+
+ assertEquals("Value check failed [node=" + node.name() +
+ ", key=" + key +
+ ", affNode=" + affNode +
+ ", primary=" + aff.isPrimary(node.cluster().localNode(), key) + ']',
+ val0,
+ val);
+ }
}
else
- assertNull("Unexpected non-null value for " + node.name(), cache.localPeek(key));
+ assertNull("Unexpected non-null value for " + node.name(), val0);
}
}
}
@@ -1080,10 +1403,6 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
clients.add(ignite);
}
- client = false;
-
- List<IgniteInternalFuture<?>> futs = new ArrayList<>();
-
final AtomicBoolean stop = new AtomicBoolean();
final AtomicInteger threadIdx = new AtomicInteger(0);
@@ -1092,8 +1411,10 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
final ConcurrentHashSet<Integer> putKeys = new ConcurrentHashSet<>();
+ IgniteInternalFuture<?> fut;
+
try {
- GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
+ fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
@Override public Object call() throws Exception {
int clientIdx = threadIdx.getAndIncrement() % CLIENT_CNT;
@@ -1122,7 +1443,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
for (int i = 0; i < 100; i++) {
Integer key = rnd.nextInt(0, 1000);
- map.put(key, key);
+ map.put(key, rnd.nextInt());
}
try {
@@ -1157,28 +1478,31 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
}
}, THREADS, "update-thread");
- for (final Ignite ignite : clients) {
+ long stopTime = System.currentTimeMillis() + 60_000;
- futs.add(GridTestUtils.runAsync(new Callable<Object>() {
- @Override public Object call() throws Exception {
- Thread.currentThread().setName("update-" + ignite.name());
+ while (System.currentTimeMillis() < stopTime) {
+ boolean restartClient = ThreadLocalRandom.current().nextBoolean();
- log.info("Start updates from node: " + ignite.name());
+ Integer idx = null;
+ if (restartClient) {
+ log.info("Start client node.");
- return null;
- }
- }));
- }
+ client = true;
- long stopTime = System.currentTimeMillis() + 60_000;
+ IgniteEx ignite = startGrid(SRV_CNT + CLIENT_CNT);
- while (System.currentTimeMillis() < stopTime) {
- int idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT);
+ IgniteCache<Integer, Integer> cache = ignite.cache(null);
- log.info("Stop node: " + idx);
+ assertNotNull(cache);
+ }
+ else {
+ idx = ThreadLocalRandom.current().nextInt(0, SRV_CNT);
+
+ log.info("Stop server node: " + idx);
- stopGrid(idx);
+ stopGrid(idx);
+ }
updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() {
@Override public void run() {
@@ -1190,7 +1514,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
}
catch (TimeoutException e) {
- log.info("Failed to wait for update.");
+ log.error("Failed to wait for update.");
U.dumpThreads(log);
@@ -1204,9 +1528,18 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
U.sleep(500);
- log.info("Start node: " + idx);
+ if (restartClient) {
+ log.info("Stop client node.");
- startGrid(idx);
+ stopGrid(SRV_CNT + CLIENT_CNT);
+ }
+ else {
+ log.info("Start server node: " + idx);
+
+ client = false;
+
+ startGrid(idx);
+ }
updateBarrier = new CyclicBarrier(THREADS + 1, new Runnable() {
@Override public void run() {
@@ -1218,7 +1551,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
updateBarrier.await(30_000, TimeUnit.MILLISECONDS);
}
catch (TimeoutException e) {
- log.info("Failed to wait for update.");
+ log.error("Failed to wait for update.");
U.dumpThreads(log);
@@ -1237,15 +1570,9 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
stop.set(true);
}
- for (IgniteInternalFuture<?> fut : futs)
- fut.get();
-
- Map<Integer, Integer> map = new HashMap<>();
-
- for (Integer key : putKeys)
- map.put(key, key);
+ fut.get(30_000);
- checkData(map, grid(SRV_CNT).cache(null), SRV_CNT + CLIENT_CNT);
+ checkData(null, putKeys, grid(SRV_CNT).cache(null), SRV_CNT + CLIENT_CNT);
}
/**
[7/7] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-709_2' into ignite-23
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-709_2' into ignite-23
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/9347e679
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/9347e679
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/9347e679
Branch: refs/heads/ignite-23
Commit: 9347e679867b68bb33c24aabd64939abfbd75f2d
Parents: 32cb360 2ca18ba
Author: sboikov <sb...@gridgain.com>
Authored: Tue May 26 17:47:17 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Tue May 26 17:47:17 2015 +0300
----------------------------------------------------------------------
.../internal/GridEventConsumeHandler.java | 106 ++++++++++---------
.../continuous/GridContinuousProcessor.java | 44 ++++++++
.../discovery/tcp/TcpClientDiscoverySpi.java | 5 -
.../spi/discovery/tcp/TcpDiscoverySpi.java | 5 -
.../discovery/tcp/TcpDiscoverySpiAdapter.java | 5 +
.../apache/ignite/internal/GridSelfTest.java | 4 +-
.../GridDiscoveryManagerAliveCacheSelfTest.java | 2 +-
.../continuous/GridEventConsumeSelfTest.java | 2 +-
8 files changed, 109 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9347e679/modules/core/src/test/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManagerAliveCacheSelfTest.java
----------------------------------------------------------------------