You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by dp...@apache.org on 2019/03/22 11:44:08 UTC
[ignite] branch ignite-2.7.5 updated: IGNITE-1126 Fix of JVM
Crashes on TeamCity, correctly cheking of stop in exchange future;
added assertion for stopping of Page Memory. - Fixes #6092.
This is an automated email from the ASF dual-hosted git repository.
dpavlov pushed a commit to branch ignite-2.7.5
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/ignite-2.7.5 by this push:
new 0d04b02 IGNITE-1126 Fix of JVM Crashes on TeamCity, correctly cheking of stop in exchange future; added assertion for stopping of Page Memory. - Fixes #6092.
0d04b02 is described below
commit 0d04b0236e42ec7a826e8b57e894251ed65ed77c
Author: EdShangGG <es...@gridgain.com>
AuthorDate: Thu Feb 14 20:49:28 2019 +0300
IGNITE-1126 Fix of JVM Crashes on TeamCity, correctly cheking of stop in exchange future; added assertion for stopping of Page Memory. - Fixes #6092.
Signed-off-by: Dmitriy Pavlov <dp...@apache.org>
(cherry picked from commit 3a775313747211869cba9bb828cde87b8fb7c75f)
---
.../pagemem/impl/PageMemoryNoStoreImpl.java | 29 +++++++++++++
.../cache/CacheAffinitySharedManager.java | 7 ++--
.../cache/GridCachePartitionExchangeManager.java | 3 +-
.../dht/GridPartitionedSingleGetFuture.java | 4 +-
.../preloader/GridDhtPartitionsExchangeFuture.java | 47 +++++++++++++++++++++-
.../cache/persistence/pagemem/PageMemoryImpl.java | 40 ++++++++++++++++++
6 files changed, 123 insertions(+), 7 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java
index 66d713c..4ba19c2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/impl/PageMemoryNoStoreImpl.java
@@ -164,6 +164,11 @@ public class PageMemoryNoStoreImpl implements PageMemory {
private final GridCacheSharedContext<?, ?> ctx;
/**
+ * Marker that stop was invoked and memory is not supposed for any usage.
+ */
+ private volatile boolean stopped;
+
+ /**
* @param log Logger.
* @param directMemoryProvider Memory allocator to use.
* @param sharedCtx Cache shared context.
@@ -202,6 +207,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public void start() throws IgniteException {
+ stopped = false;
+
long startSize = dataRegionCfg.getInitialSize();
long maxSize = dataRegionCfg.getMaxSize();
@@ -243,6 +250,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
if (log.isDebugEnabled())
log.debug("Stopping page memory.");
+ stopped = true;
+
directMemoryProvider.shutdown(deallocate);
if (directMemoryProvider instanceof Closeable) {
@@ -262,6 +271,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public long allocatePage(int grpId, int partId, byte flags) {
+ assert !stopped;
+
long relPtr = borrowFreePage();
long absPtr = 0;
@@ -326,6 +337,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public boolean freePage(int cacheId, long pageId) {
+ assert !stopped;
+
releaseFreePage(pageId);
return true;
@@ -445,6 +458,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public long acquirePage(int cacheId, long pageId) {
+ assert !stopped;
+
int pageIdx = PageIdUtils.pageIndex(pageId);
Segment seg = segment(pageIdx);
@@ -454,6 +469,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public void releasePage(int cacheId, long pageId, long page) {
+ assert !stopped;
+
if (trackAcquiredPages) {
Segment seg = segment(PageIdUtils.pageIndex(pageId));
@@ -463,6 +480,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public long readLock(int cacheId, long pageId, long page) {
+ assert !stopped;
+
if (rwLock.readLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId)))
return page + PAGE_OVERHEAD;
@@ -471,6 +490,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public long readLockForce(int cacheId, long pageId, long page) {
+ assert !stopped;
+
if (rwLock.readLock(page + LOCK_OFFSET, -1))
return page + PAGE_OVERHEAD;
@@ -479,11 +500,15 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public void readUnlock(int cacheId, long pageId, long page) {
+ assert !stopped;
+
rwLock.readUnlock(page + LOCK_OFFSET);
}
/** {@inheritDoc} */
@Override public long writeLock(int cacheId, long pageId, long page) {
+ assert !stopped;
+
if (rwLock.writeLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId)))
return page + PAGE_OVERHEAD;
@@ -492,6 +517,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
/** {@inheritDoc} */
@Override public long tryWriteLock(int cacheId, long pageId, long page) {
+ assert !stopped;
+
if (rwLock.tryWriteLock(page + LOCK_OFFSET, PageIdUtils.tag(pageId)))
return page + PAGE_OVERHEAD;
@@ -506,6 +533,8 @@ public class PageMemoryNoStoreImpl implements PageMemory {
Boolean walPlc,
boolean dirtyFlag
) {
+ assert !stopped;
+
long actualId = PageIO.getPageId(page + PAGE_OVERHEAD);
rwLock.writeUnlock(page + LOCK_OFFSET, PageIdUtils.tag(actualId));
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 0dda7ef..5a957f1 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
@@ -1461,7 +1461,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
assert grp != null;
if (affReq != null && affReq.contains(aff.groupId())) {
- assert AffinityTopologyVersion.NONE.equals(aff.lastVersion());
+ assert AffinityTopologyVersion.NONE.equals(aff.lastVersion()) : aff.lastVersion();
CacheGroupAffinityMessage affMsg = receivedAff.get(aff.groupId());
@@ -1469,7 +1469,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
List<List<ClusterNode>> assignments = affMsg.createAssignments(nodesByOrder, evts.discoveryCache());
- assert resTopVer.equals(evts.topologyVersion());
+ assert resTopVer.equals(evts.topologyVersion()) : "resTopVer=" + resTopVer +
+ ", evts.topVer=" + evts.topologyVersion();
List<List<ClusterNode>> idealAssign =
affMsg.createIdealAssignments(nodesByOrder, evts.discoveryCache());
@@ -1477,7 +1478,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
if (idealAssign != null)
aff.idealAssignment(idealAssign);
else {
- assert !aff.centralizedAffinityFunction();
+ assert !aff.centralizedAffinityFunction() : aff;
// Calculate ideal assignments.
aff.calculate(evts.topologyVersion(), evts, evts.discoveryCache());
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 eb89010..f18b404 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
@@ -58,6 +58,7 @@ import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteNeedReconnectException;
+import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
@@ -754,7 +755,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
stopErr = cctx.kernalContext().clientDisconnected() ?
new IgniteClientDisconnectedCheckedException(cctx.kernalContext().cluster().clientReconnectFuture(),
"Client node disconnected: " + cctx.igniteInstanceName()) :
- new IgniteInterruptedCheckedException("Node is stopping: " + cctx.igniteInstanceName());
+ new NodeStoppingException("Node is stopping: " + cctx.igniteInstanceName());
// Stop exchange worker
U.cancel(exchWorker);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 4dc72b2..e93689e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -27,6 +27,7 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteDiagnosticPrepareContext;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -776,7 +777,8 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
if (trackable)
cctx.mvcc().removeFuture(futId);
- cctx.dht().sendTtlUpdateRequest(expiryPlc);
+ if (!(err instanceof NodeStoppingException))
+ cctx.dht().sendTtlUpdateRequest(expiryPlc);
return true;
}
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 6f38757..177fed1 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
@@ -190,7 +190,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
/** Cache context. */
private final GridCacheSharedContext<?, ?> cctx;
- /** Busy lock to prevent activities from accessing exchanger while it's stopping. */
+ /**
+ * Busy lock to prevent activities from accessing exchanger while it's stopping. Stopping uses write lock, so every
+ * {@link #enterBusy()} will be failed as false. But regular operation uses read lock acquired multiple times.
+ */
private ReadWriteLock busyLock;
/** */
@@ -604,7 +607,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
}
/**
- * @return {@code true} if entered to busy state.
+ * @return {@code true} if entered to busy state. {@code false} for stop node.
*/
private boolean enterBusy() {
if (busyLock.readLock().tryLock())
@@ -3202,6 +3205,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
* @param sndResNodes Additional nodes to send finish message to.
*/
private void finishExchangeOnCoordinator(@Nullable Collection<ClusterNode> sndResNodes) {
+ if (isDone() || !enterBusy())
+ return;
+
try {
if (!F.isEmpty(exchangeGlobalExceptions) && dynamicCacheStartExchange() && isRollbackSupported()) {
sendExchangeFailureMessage();
@@ -3489,6 +3495,43 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
else
onDone(e);
}
+ finally {
+ leaveBusy();
+ }
+ }
+
+ /**
+ * @param msg Single message to process.
+ * @param messageAccumulator Message to store message which need to be sent after.
+ */
+ private void processSingleMessageOnCrdFinish(
+ GridDhtPartitionsSingleMessage msg,
+ Map<Integer, CacheGroupAffinityMessage> messageAccumulator
+ ) {
+ for (Map.Entry<Integer, GridDhtPartitionMap> e : msg.partitions().entrySet()) {
+ Integer grpId = e.getKey();
+
+ CacheGroupContext grp = cctx.cache().cacheGroup(grpId);
+
+ GridDhtPartitionTopology top = grp != null
+ ? grp.topology()
+ : cctx.exchange().clientTopology(grpId, events().discoveryCache());
+
+ CachePartitionPartialCountersMap cntrs = msg.partitionUpdateCounters(grpId, top.partitions());
+
+ if (cntrs != null)
+ top.collectUpdateCounters(cntrs);
+ }
+
+ Collection<Integer> affReq = msg.cacheGroupsAffinityRequest();
+
+ if (affReq != null)
+ CacheGroupAffinityMessage.createAffinityMessages(
+ cctx,
+ exchCtx.events().topologyVersion(),
+ affReq,
+ messageAccumulator
+ );
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
index ef87b3a..5c91378 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/PageMemoryImpl.java
@@ -280,6 +280,11 @@ public class PageMemoryImpl implements PageMemoryEx {
private DataRegionMetricsImpl memMetrics;
/**
+ * Marker that stop was invoked and memory is not supposed for any usage.
+ */
+ private volatile boolean stopped;
+
+ /**
* @param directMemoryProvider Memory allocator to use.
* @param sizes segments sizes, last is checkpoint pool size.
* @param ctx Cache shared context.
@@ -341,6 +346,8 @@ public class PageMemoryImpl implements PageMemoryEx {
/** {@inheritDoc} */
@Override public void start() throws IgniteException {
+ stopped = false;
+
directMemoryProvider.initialize(sizes);
List<DirectMemoryRegion> regions = new ArrayList<>(sizes.length);
@@ -416,11 +423,15 @@ public class PageMemoryImpl implements PageMemoryEx {
seg.close();
}
+ stopped = true;
+
directMemoryProvider.shutdown(deallocate);
}
/** {@inheritDoc} */
@Override public void releasePage(int grpId, long pageId, long page) {
+ assert !stopped;
+
Segment seg = segment(grpId, pageId);
seg.readLock().lock();
@@ -435,43 +446,59 @@ public class PageMemoryImpl implements PageMemoryEx {
/** {@inheritDoc} */
@Override public long readLock(int grpId, long pageId, long page) {
+ assert !stopped;
+
return readLockPage(page, new FullPageId(pageId, grpId), false);
}
/** {@inheritDoc} */
@Override public void readUnlock(int grpId, long pageId, long page) {
+ assert !stopped;
+
readUnlockPage(page);
}
/** {@inheritDoc} */
@Override public long writeLock(int grpId, long pageId, long page) {
+ assert !stopped;
+
return writeLock(grpId, pageId, page, false);
}
/** {@inheritDoc} */
@Override public long writeLock(int grpId, long pageId, long page, boolean restore) {
+ assert !stopped;
+
return writeLockPage(page, new FullPageId(pageId, grpId), !restore);
}
/** {@inheritDoc} */
@Override public long tryWriteLock(int grpId, long pageId, long page) {
+ assert !stopped;
+
return tryWriteLockPage(page, new FullPageId(pageId, grpId), true);
}
/** {@inheritDoc} */
@Override public void writeUnlock(int grpId, long pageId, long page, Boolean walPlc,
boolean dirtyFlag) {
+ assert !stopped;
+
writeUnlock(grpId, pageId, page, walPlc, dirtyFlag, false);
}
/** {@inheritDoc} */
@Override public void writeUnlock(int grpId, long pageId, long page, Boolean walPlc,
boolean dirtyFlag, boolean restore) {
+ assert !stopped;
+
writeUnlockPage(page, new FullPageId(pageId, grpId), walPlc, dirtyFlag, restore);
}
/** {@inheritDoc} */
@Override public boolean isDirty(int grpId, long pageId, long page) {
+ assert !stopped;
+
return isDirty(page);
}
@@ -481,6 +508,7 @@ public class PageMemoryImpl implements PageMemoryEx {
flags == PageIdAllocator.FLAG_IDX && partId == PageIdAllocator.INDEX_PARTITION :
"flags = " + flags + ", partId = " + partId;
+ assert !stopped;
assert stateChecker.checkpointLockIsHeldByThread();
if (isThrottlingEnabled())
@@ -636,21 +664,29 @@ public class PageMemoryImpl implements PageMemoryEx {
/** {@inheritDoc} */
@Override public long metaPageId(int grpId) throws IgniteCheckedException {
+ assert !stopped;
+
return storeMgr.metaPageId(grpId);
}
/** {@inheritDoc} */
@Override public long partitionMetaPageId(int grpId, int partId) throws IgniteCheckedException {
+ assert !stopped;
+
return PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, 0);
}
/** {@inheritDoc} */
@Override public long acquirePage(int grpId, long pageId) throws IgniteCheckedException {
+ assert !stopped;
+
return acquirePage(grpId, pageId, false);
}
/** {@inheritDoc} */
@Override public long acquirePage(int grpId, long pageId, boolean restore) throws IgniteCheckedException {
+ assert !stopped;
+
FullPageId fullId = new FullPageId(pageId, grpId);
int partId = PageIdUtils.partId(pageId);
@@ -1389,6 +1425,8 @@ public class PageMemoryImpl implements PageMemoryEx {
* @return Pointer to the page read buffer.
*/
private long readLockPage(long absPtr, FullPageId fullId, boolean force, boolean touch) {
+ assert !stopped;
+
int tag = force ? -1 : PageIdUtils.tag(fullId.pageId());
boolean locked = rwLock.readLock(absPtr + PAGE_LOCK_OFFSET, tag);
@@ -1406,6 +1444,8 @@ public class PageMemoryImpl implements PageMemoryEx {
/** {@inheritDoc} */
@Override public long readLockForce(int grpId, long pageId, long page) {
+ assert !stopped;
+
return readLockPage(page, new FullPageId(pageId, grpId), true);
}