You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2018/06/27 14:50:06 UTC
[39/50] [abbrv] ignite git commit: IGNITE-8768 Fixed JVM crash caused
by an in-progress partition eviction during cache stop - Fixes #4227.
IGNITE-8768 Fixed JVM crash caused by an in-progress partition eviction during cache stop - Fixes #4227.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/56975c26
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/56975c26
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/56975c26
Branch: refs/heads/ignite-8446
Commit: 56975c266e7019f307bb9da42333a6db4e47365e
Parents: 16a7c98
Author: Pavel Kovalenko <jo...@gmail.com>
Authored: Tue Jun 26 12:23:33 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Tue Jun 26 12:23:33 2018 +0300
----------------------------------------------------------------------
.../processors/cache/CacheGroupContext.java | 2 +
.../cache/distributed/dht/EvictionContext.java | 28 ++
.../distributed/dht/GridDhtLocalPartition.java | 14 +-
.../dht/GridDhtPartitionsEvictor.java | 326 ++++++++++++++-----
4 files changed, 286 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/56975c26/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 8a65038..99f9f97 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -723,6 +723,8 @@ public class CacheGroupContext {
IgniteCheckedException err =
new IgniteCheckedException("Failed to wait for topology update, cache (or node) is stopping.");
+ evictor.stop();
+
aff.cancelFutures(err);
preldr.onKernalStop();
http://git-wip-us.apache.org/repos/asf/ignite/blob/56975c26/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/EvictionContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/EvictionContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/EvictionContext.java
new file mode 100644
index 0000000..0964c3c
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/EvictionContext.java
@@ -0,0 +1,28 @@
+/*
+ * 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.dht;
+
+/**
+ * Additional context for partition eviction process.
+ */
+public interface EvictionContext {
+ /**
+ * @return {@code true} If eviction process should be stopped.
+ */
+ public boolean shouldStop();
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/56975c26/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
index 81249b6..e13c952 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLocalPartition.java
@@ -850,14 +850,14 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
}
/**
- * Tries to start partition clear process {@link GridDhtLocalPartition#clearAll()}).
+ * Tries to start partition clear process {@link GridDhtLocalPartition#clearAll(EvictionContext)}).
* Only one thread is allowed to do such process concurrently.
* At the end of clearing method completes {@code clearFuture}.
*
* @return {@code false} if clearing is not started due to existing reservations.
* @throws NodeStoppingException If node is stopping.
*/
- public boolean tryClear() throws NodeStoppingException {
+ public boolean tryClear(EvictionContext evictionCtx) throws NodeStoppingException {
if (clearFuture.isDone())
return true;
@@ -869,7 +869,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
if (addEvicting()) {
try {
// Attempt to evict partition entries from cache.
- long clearedEntities = clearAll();
+ long clearedEntities = clearAll(evictionCtx);
if (log.isDebugEnabled())
log.debug("Partition is cleared [clearedEntities=" + clearedEntities + ", part=" + this + "]");
@@ -985,7 +985,7 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
* @return Number of rows cleared from page memory.
* @throws NodeStoppingException If node stopping.
*/
- private long clearAll() throws NodeStoppingException {
+ private long clearAll(EvictionContext evictionCtx) throws NodeStoppingException {
GridCacheVersion clearVer = ctx.versions().next();
GridCacheObsoleteEntryExtras extras = new GridCacheObsoleteEntryExtras(clearVer);
@@ -1001,6 +1001,8 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
long cleared = 0;
+ final int stopCheckingFreq = 1000;
+
CacheMapHolder hld = grp.sharedGroup() ? null : singleCacheEntryMap;
try {
@@ -1051,6 +1053,10 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
cleared++;
}
+
+ // For each 'stopCheckingFreq' cleared entities check clearing process to stop.
+ if (cleared % stopCheckingFreq == 0 && evictionCtx.shouldStop())
+ return cleared;
}
catch (GridDhtInvalidPartitionException e) {
assert isEmpty() && state() == EVICTED : "Invalid error [e=" + e + ", part=" + this + ']';
http://git-wip-us.apache.org/repos/asf/ignite/blob/56975c26/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
index 2a28921..7206397 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionsEvictor.java
@@ -16,21 +16,32 @@
*/
package org.apache.ignite.internal.processors.cache.distributed.dht;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.function.Function;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.util.typedef.internal.GPC;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.U;
/**
- * Class that serves asynchronous partition eviction process.
+ * Class that serves asynchronous part eviction process.
+ * Only one partition from group can be evicted at the moment.
*/
public class GridDhtPartitionsEvictor {
- /** Show eviction progress frequency in ms. */
- private static final int SHOW_EVICTION_PROGRESS_FREQ_MS = 2 * 60 * 1000; // 2 Minutes.
+ /** Default eviction progress show frequency. */
+ private static final int DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS = 2 * 60 * 1000; // 2 Minutes.
+
+ /** Eviction progress frequency property name. */
+ private static final String SHOW_EVICTION_PROGRESS_FREQ = "SHOW_EVICTION_PROGRESS_FREQ";
/** */
private final GridCacheSharedContext<?, ?> ctx;
@@ -41,11 +52,31 @@ public class GridDhtPartitionsEvictor {
/** */
private final IgniteLogger log;
+ /** Lock object. */
+ private final Object mux = new Object();
+
/** Queue contains partitions scheduled for eviction. */
- private final ConcurrentHashMap<Integer, GridDhtLocalPartition> evictionQueue = new ConcurrentHashMap<>();
+ private final DeduplicationQueue<Integer, GridDhtLocalPartition> evictionQueue = new DeduplicationQueue<>(GridDhtLocalPartition::id);
+
+ /**
+ * Flag indicates that eviction process is running at the moment.
+ * This is needed to schedule partition eviction if there are no currently running self-scheduling eviction tasks.
+ * Guarded by {@link #mux}.
+ */
+ private boolean evictionRunning;
- /** Flag indicates that eviction process is running at the moment, false in other case. */
- private final AtomicBoolean evictionRunning = new AtomicBoolean();
+ /** Flag indicates that eviction process has stopped. */
+ private volatile boolean stop;
+
+ /** Future for currently running partition eviction task. */
+ private volatile GridFutureAdapter<Boolean> evictionFut;
+
+ /** Eviction progress frequency in ms. */
+ private final long evictionProgressFreqMs = IgniteSystemProperties.getLong(SHOW_EVICTION_PROGRESS_FREQ,
+ DEFAULT_SHOW_EVICTION_PROGRESS_FREQ_MS);
+
+ /** Next time of show eviction progress. */
+ private long nextShowProgressTime;
/**
* Constructor.
@@ -67,78 +98,213 @@ public class GridDhtPartitionsEvictor {
* @param part Partition to evict.
*/
public void evictPartitionAsync(GridDhtLocalPartition part) {
- evictionQueue.putIfAbsent(part.id(), part);
-
- if (evictionRunning.compareAndSet(false, true)) {
- ctx.kernalContext().closure().callLocalSafe(new GPC<Boolean>() {
- @Override public Boolean call() {
- boolean locked = true;
-
- long nextShowProgressTime = U.currentTimeMillis() + SHOW_EVICTION_PROGRESS_FREQ_MS;
-
- while (locked || !evictionQueue.isEmpty()) {
- if (!locked && !evictionRunning.compareAndSet(false, true))
- return false;
-
- try {
- for (GridDhtLocalPartition part : evictionQueue.values()) {
- // Show progress of currently evicting partitions.
- if (U.currentTimeMillis() >= nextShowProgressTime) {
- if (log.isInfoEnabled())
- log.info("Eviction in progress [grp=" + grp.cacheOrGroupName()
- + ", remainingCnt=" + evictionQueue.size() + "]");
-
- nextShowProgressTime = U.currentTimeMillis() + SHOW_EVICTION_PROGRESS_FREQ_MS;
- }
-
- try {
- boolean success = part.tryClear();
-
- if (success) {
- evictionQueue.remove(part.id());
-
- if (part.state() == GridDhtPartitionState.EVICTED && part.markForDestroy())
- part.destroy();
- }
- }
- catch (Throwable ex) {
- if (ctx.kernalContext().isStopping()) {
- LT.warn(log, ex, "Partition eviction failed (current node is stopping).",
- false,
- true);
-
- evictionQueue.clear();
-
- return true;
- }
- else
- LT.error(log, ex, "Partition eviction failed, this can cause grid hang.");
- }
- }
- }
- finally {
- if (!evictionQueue.isEmpty()) {
- if (ctx.kernalContext().isStopping()) {
- evictionQueue.clear();
-
- locked = false;
- }
- else
- locked = true;
- }
- else {
- boolean res = evictionRunning.compareAndSet(true, false);
-
- assert res;
-
- locked = false;
- }
- }
- }
-
- return true;
+ if (stop)
+ return;
+
+ boolean added = evictionQueue.offer(part);
+
+ if (!added)
+ return;
+
+ synchronized (mux) {
+ if (!evictionRunning) {
+ nextShowProgressTime = U.currentTimeMillis() + evictionProgressFreqMs;
+
+ scheduleNextPartitionEviction();
+ }
+ }
+ }
+
+ /**
+ * Stops eviction process.
+ * Method awaits last offered partition eviction.
+ */
+ public void stop() {
+ stop = true;
+
+ synchronized (mux) {
+ // Wait for last offered partition eviction completion.
+ IgniteInternalFuture<Boolean> evictionFut0 = evictionFut;
+
+ if (evictionFut0 != null) {
+ try {
+ evictionFut0.get();
+ }
+ catch (IgniteCheckedException e) {
+ if (log.isDebugEnabled())
+ log.warning("Failed to await partition eviction during stopping", e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Gets next partition from the queue and schedules it for eviction.
+ */
+ private void scheduleNextPartitionEviction() {
+ if (stop)
+ return;
+
+ synchronized (mux) {
+ GridDhtLocalPartition next = evictionQueue.poll();
+
+ if (next != null) {
+ showProgress();
+
+ evictionFut = new GridFutureAdapter<>();
+
+ ctx.kernalContext().closure().callLocalSafe(new PartitionEvictionTask(next, () -> stop), true);
+ }
+ else
+ evictionRunning = false;
+ }
+ }
+
+ /**
+ * Shows progress of eviction.
+ */
+ private void showProgress() {
+ if (U.currentTimeMillis() >= nextShowProgressTime) {
+ int size = evictionQueue.size() + 1; // Queue size plus current partition.
+
+ if (log.isInfoEnabled())
+ log.info("Eviction in progress [grp=" + grp.cacheOrGroupName()
+ + ", remainingPartsCnt=" + size + "]");
+
+ nextShowProgressTime = U.currentTimeMillis() + evictionProgressFreqMs;
+ }
+ }
+
+ /**
+ * Task for self-scheduled partition eviction / clearing.
+ */
+ private class PartitionEvictionTask implements Callable<Boolean> {
+ /** Partition to evict. */
+ private final GridDhtLocalPartition part;
+
+ /** Eviction context. */
+ private final EvictionContext evictionCtx;
+
+ /**
+ * @param part Partition.
+ * @param evictionCtx Eviction context.
+ */
+ public PartitionEvictionTask(GridDhtLocalPartition part, EvictionContext evictionCtx) {
+ this.part = part;
+ this.evictionCtx = evictionCtx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Boolean call() throws Exception {
+ if (stop) {
+ evictionFut.onDone();
+
+ return false;
+ }
+
+ try {
+ boolean success = part.tryClear(evictionCtx);
+
+ if (success) {
+ if (part.state() == GridDhtPartitionState.EVICTED && part.markForDestroy())
+ part.destroy();
}
- }, /*system pool*/ true);
+ else // Re-offer partition if clear was unsuccessful due to partition reservation.
+ evictionQueue.offer(part);
+
+ // Complete eviction future before schedule new to prevent deadlock with
+ // simultaneous eviction stopping and scheduling new eviction.
+ evictionFut.onDone();
+
+ scheduleNextPartitionEviction();
+
+ return true;
+ }
+ catch (Throwable ex) {
+ evictionFut.onDone(ex);
+
+ if (ctx.kernalContext().isStopping()) {
+ LT.warn(log, ex, "Partition eviction failed (current node is stopping).",
+ false,
+ true);
+ }
+ else
+ LT.error(log, ex, "Partition eviction failed, this can cause grid hang.");
+ }
+
+ return false;
+ }
+ }
+
+ /**
+ * Thread-safe blocking queue with items deduplication.
+ *
+ * @param <K> Key type of item used for deduplication.
+ * @param <V> Queue item type.
+ */
+ private static class DeduplicationQueue<K, V> {
+ /** Queue. */
+ private final Queue<V> queue;
+
+ /** Unique items set. */
+ private final Set<K> uniqueItems;
+
+ /** Key mapping function. */
+ private final Function<V, K> keyMappingFunction;
+
+ /**
+ * Constructor.
+ *
+ * @param keyExtractor Function to extract a key from a queue item.
+ * This key is used for deduplication if some item has offered twice.
+ */
+ public DeduplicationQueue(Function<V, K> keyExtractor) {
+ keyMappingFunction = keyExtractor;
+ queue = new LinkedBlockingQueue<>();
+ uniqueItems = new GridConcurrentHashSet<>();
+ }
+
+ /**
+ * Offers item to the queue.
+ *
+ * @param item Item.
+ * @return {@code true} if item has been successfully offered to the queue,
+ * {@code false} if item was rejected because already exists in the queue.
+ */
+ public boolean offer(V item) {
+ K key = keyMappingFunction.apply(item);
+
+ if (uniqueItems.add(key)) {
+ queue.offer(item);
+
+ return true;
+ }
+
+ return false;
+ }
+
+ /**
+ * Polls next item from queue.
+ *
+ * @return Next item or {@code null} if queue is empty.
+ */
+ public V poll() {
+ V item = queue.poll();
+
+ if (item != null) {
+ K key = keyMappingFunction.apply(item);
+
+ uniqueItems.remove(key);
+ }
+
+ return item;
+ }
+
+ /**
+ * @return Size of queue.
+ */
+ public int size() {
+ return queue.size();
}
}
}