You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2018/07/03 16:20:20 UTC
ignite git commit: IGNITE-8681 Using ExpiryPolicy with persistence
causes significant slowdown - Fixes #4285.
Repository: ignite
Updated Branches:
refs/heads/master c0cc7d78e -> a06c47f7e
IGNITE-8681 Using ExpiryPolicy with persistence causes significant slowdown - Fixes #4285.
Signed-off-by: Ivan Rakov <ir...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a06c47f7
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a06c47f7
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a06c47f7
Branch: refs/heads/master
Commit: a06c47f7e90005d342ab100270398e9c5ea5423d
Parents: c0cc7d7
Author: Andrey V. Mashenkov <an...@gmail.com>
Authored: Tue Jul 3 19:19:47 2018 +0300
Committer: Ivan Rakov <ir...@apache.org>
Committed: Tue Jul 3 19:19:47 2018 +0300
----------------------------------------------------------------------
.../apache/ignite/IgniteSystemProperties.java | 7 +
.../processors/cache/GridCacheMapEntry.java | 193 ++++++++++++++-----
.../processors/cache/GridCacheTtlManager.java | 3 +
.../cache/IgniteCacheOffheapManagerImpl.java | 52 ++++-
.../colocated/GridDhtDetachedCacheEntry.java | 3 +-
.../distributed/near/GridNearCacheEntry.java | 2 +-
.../persistence/GridCacheOffheapManager.java | 45 ++++-
.../IgniteCacheExpiryStoreLoadSelfTest.java | 2 +-
.../cache/GridCacheAbstractMetricsSelfTest.java | 8 +-
.../IgniteCacheEntryListenerAbstractTest.java | 4 +-
.../IgniteCacheExpiryPolicyTestSuite.java | 2 -
.../expiry/IgniteCacheTtlCleanupSelfTest.java | 87 ---------
.../CacheOperationsWithExpirationTest.java | 6 +-
.../cache/IgniteCacheAbstractQuerySelfTest.java | 4 +-
...achePartitionedQueryP2PDisabledSelfTest.java | 2 +-
15 files changed, 256 insertions(+), 164 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index a7d9199..6ba44ee 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -924,6 +924,13 @@ public final class IgniteSystemProperties {
public static final String IGNITE_DUMP_THREADS_ON_FAILURE = "IGNITE_DUMP_THREADS_ON_FAILURE";
/**
+ * Throttling timeout in millis which avoid excessive PendingTree access on unwind if there is nothing to clean yet.
+ *
+ * Default is 500 ms.
+ */
+ public static final String IGNITE_UNWIND_THROTTLING_TIMEOUT = "IGNITE_UNWIND_THROTTLING_TIMEOUT";
+
+ /**
* Enforces singleton.
*/
private IgniteSystemProperties() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/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 cd984bd..03f6d11 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
@@ -403,9 +403,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
update(val, read.expireTime(), 0, read.version(), false);
- long delta = checkExpire ?
- (read.expireTime() == 0 ? 0 : read.expireTime() - U.currentTimeMillis())
- : 0;
+ long delta = checkExpire && read.expireTime() > 0 ? read.expireTime() - U.currentTimeMillis() : 0;
if (delta >= 0)
return read;
@@ -612,7 +610,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (val != null) {
long expireTime = expireTimeExtras();
- if (expireTime > 0 && (expireTime - U.currentTimeMillis() <= 0)) {
+ if (expireTime > 0 && (expireTime < U.currentTimeMillis())) {
if (onExpired((CacheObject)cctx.unwrapTemporary(val), null)) {
val = null;
evt = false;
@@ -754,7 +752,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
long expTime = CU.toExpireTime(ttl);
// Update indexes before actual write to entry.
- storeValue(ret, expTime, nextVer, null);
+ storeValue(ret, expTime, nextVer);
update(ret, expTime, ttl, nextVer, true);
@@ -865,7 +863,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
// Update indexes.
if (ret != null) {
- storeValue(ret, expTime, nextVer, null);
+ storeValue(ret, expTime, nextVer);
if (cctx.deferredDelete() && !isInternal() && !detached() && deletedUnlocked())
deletedUnlocked(false);
@@ -1028,7 +1026,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert val != null;
- storeValue(val, expireTime, newVer, null);
+ storeValue(val, expireTime, newVer);
if (cctx.deferredDelete() && deletedUnlocked() && !isInternal() && !detached())
deletedUnlocked(false);
@@ -1375,6 +1373,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
lockEntry();
try {
+ checkObsolete();
+
boolean internal = isInternal() || !context().userCache();
Map<UUID, CacheContinuousQueryListener> lsnrCol =
@@ -1386,16 +1386,22 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
!F.isEmpty(filter) ||
lsnrCol != null;
- checkObsolete();
-
- CacheDataRow oldRow = null;
-
// Load and remove from swap if it is new.
if (isNew())
- oldRow = unswap(null, false);
+ unswap(null, false);
old = val;
+ if (expireTimeExtras() > 0 && expireTimeExtras() < U.currentTimeMillis()) {
+ if (onExpired(val, null)) {
+ assert !deletedUnlocked();
+
+ update(null, CU.TTL_ETERNAL, CU.EXPIRE_TIME_ETERNAL, ver, true);
+
+ old = null;
+ }
+ }
+
boolean readFromStore = false;
Object old0 = null;
@@ -1426,7 +1432,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
old = cctx.kernalContext().cacheObjects().prepareForCache(old, cctx);
if (old != null)
- storeValue(old, expireTime, ver, oldRow);
+ storeValue(old, expireTime, ver);
else
removeValue();
@@ -1570,7 +1576,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
// Must persist inside synchronization in non-tx mode.
cctx.store().put(null, key, updated, ver);
- storeValue(updated, expireTime, ver, oldRow);
+ storeValue(updated, expireTime, ver);
assert ttl != CU.TTL_ZERO;
@@ -1758,7 +1764,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
assert updateRes != null : c;
- CacheObject oldVal = c.oldRow != null ? c.oldRow.value() : null;
+ // We should ignore expired old row. Expired oldRow instance is needed for correct row replacement\deletion only.
+ CacheObject oldVal = c.oldRow != null && !c.oldRowExpiredFlag ? c.oldRow.value() : null;
CacheObject updateVal = null;
GridCacheVersion updateVer = c.newVer;
@@ -2179,7 +2186,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (hasValueUnlocked()) {
long expireTime = expireTimeExtras();
- if (expireTime > 0 && (expireTime - U.currentTimeMillis() <= 0)) {
+ if (expireTime > 0 && (expireTime < U.currentTimeMillis())) {
if (obsoleteVer == null)
obsoleteVer = nextVersion();
@@ -2261,8 +2268,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/**
* <p>
- * Note that {@link #onMarkedObsolete()} should always be called after this method
- * returns {@code true}.
+ * Note that {@link #onMarkedObsolete()} should always be called after this method returns {@code true}.
*
* @param ver Version.
* @param clear {@code True} to clear.
@@ -2450,7 +2456,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
cctx.shared().database().checkpointReadLock();
try {
- storeValue(val, expireTime, ver, null);
+ storeValue(val, expireTime, ver);
}
finally {
cctx.shared().database().checkpointReadUnlock();
@@ -2745,9 +2751,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (update) {
// If entry is already unswapped and we are modifying it, we must run deletion callbacks for old value.
long oldExpTime = expireTimeUnlocked();
- long delta = (oldExpTime == 0 ? 0 : oldExpTime - U.currentTimeMillis());
- if (delta < 0) {
+ if (oldExpTime > 0 && oldExpTime < U.currentTimeMillis()) {
if (onExpired(this.val, null)) {
if (cctx.deferredDelete()) {
deferred = true;
@@ -2758,11 +2763,11 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
- storeValue(val, expTime, ver, null);
+ storeValue(val, expTime, ver);
}
}
else // Optimization to access storage only once.
- update = storeValue(val, expTime, ver, null, p);
+ update = storeValue(val, expTime, ver, p);
if (update) {
update(val, expTime, ttl, ver, true);
@@ -2951,7 +2956,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
if (val != null) {
- storeValue(val, expTime, newVer, null);
+ storeValue(val, expTime, newVer);
if (deletedUnlocked())
deletedUnlocked(false);
@@ -3322,7 +3327,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
long expireTime = expireTimeExtras();
- if (expireTime == 0 || (expireTime - U.currentTimeMillis() > 0))
+ if (!(expireTime > 0 && expireTime < U.currentTimeMillis()))
return false;
CacheObject expiredVal = this.val;
@@ -3571,14 +3576,12 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
* @param val Value.
* @param expireTime Expire time.
* @param ver New entry version.
- * @param oldRow Old row if available.
* @throws IgniteCheckedException If update failed.
*/
protected boolean storeValue(@Nullable CacheObject val,
long expireTime,
- GridCacheVersion ver,
- @Nullable CacheDataRow oldRow) throws IgniteCheckedException {
- return storeValue(val, expireTime, ver, oldRow, null);
+ GridCacheVersion ver) throws IgniteCheckedException {
+ return storeValue(val, expireTime, ver, null);
}
/**
@@ -3587,7 +3590,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
* @param val Value.
* @param expireTime Expire time.
* @param ver New entry version.
- * @param oldRow Old row if available.
* @param predicate Optional predicate.
*
* @return {@code True} if storage was modified.
@@ -3597,7 +3599,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable CacheObject val,
long expireTime,
GridCacheVersion ver,
- @Nullable CacheDataRow oldRow,
@Nullable IgnitePredicate<CacheDataRow> predicate) throws IgniteCheckedException {
assert lock.isHeldByCurrentThread();
@@ -4415,6 +4416,12 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** {@inheritDoc} */
@Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
+ if (oldRow != null) {
+ oldRow.key(entry.key);
+
+ oldRow = checkRowExpired(oldRow);
+ }
+
this.oldRow = oldRow;
if (predicate != null && !predicate.apply(oldRow)) {
@@ -4423,9 +4430,6 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
return;
}
- if (oldRow != null)
- oldRow.key(entry.key);
-
if (val != null) {
newRow = entry.cctx.offheap().dataStore(entry.localPartition()).createRow(
entry.cctx,
@@ -4456,6 +4460,53 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Nullable @Override public CacheDataRow oldRow() {
return oldRow;
}
+
+ /**
+ * Checks row for expiration and fire expire events if needed.
+ *
+ * @param row old row.
+ * @return {@code Null} if row was expired, row itself otherwise.
+ * @throws IgniteCheckedException
+ */
+ private CacheDataRow checkRowExpired(CacheDataRow row) throws IgniteCheckedException {
+ assert row != null;
+
+ if (!(row.expireTime() > 0 && row.expireTime() < U.currentTimeMillis()))
+ return row;
+
+ GridCacheContext cctx = entry.context();
+
+ CacheObject expiredVal = row.value();
+
+ if (cctx.deferredDelete() && !entry.detached() && !entry.isInternal()) {
+ entry.update(null, CU.TTL_ETERNAL, CU.EXPIRE_TIME_ETERNAL, entry.ver, true);
+
+ if (!entry.deletedUnlocked() && !entry.isStartVersion())
+ entry.deletedUnlocked(true);
+ }
+ else
+ entry.markObsolete0(cctx.versions().next(), true, null);
+
+ if (cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED)) {
+ cctx.events().addEvent(entry.partition(),
+ entry.key(),
+ cctx.localNodeId(),
+ null,
+ EVT_CACHE_OBJECT_EXPIRED,
+ null,
+ false,
+ expiredVal,
+ expiredVal != null,
+ null,
+ null,
+ null,
+ true);
+ }
+
+ cctx.continuousQueries().onEntryExpired(entry, entry.key(), expiredVal);
+
+ return null;
+ }
}
/**
@@ -4531,6 +4582,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
/** */
private CacheDataRow oldRow;
+ /** OldRow expiration flag. */
+ private boolean oldRowExpiredFlag = false;
+
AtomicCacheUpdateClosure(
GridCacheMapEntry entry,
AffinityTopologyVersion topVer,
@@ -4605,23 +4659,27 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
@Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
assert entry.isNear() || oldRow == null || oldRow.link() != 0 : oldRow;
- if (oldRow != null)
- oldRow.key(entry.key());
-
- this.oldRow = oldRow;
-
GridCacheContext cctx = entry.context();
CacheObject oldVal;
CacheObject storeLoadedVal = null;
+ this.oldRow = oldRow;
+
if (oldRow != null) {
- oldVal = oldRow.value();
+ oldRow.key(entry.key());
+
+ // unswap
+ entry.update(oldRow.value(), oldRow.expireTime(), 0, oldRow.version(), false);
- entry.update(oldVal, oldRow.expireTime(), 0, oldRow.version(), false);
+ if (checkRowExpired(oldRow)) {
+ oldRowExpiredFlag = true;
+
+ oldRow = null;
+ }
}
- else
- oldVal = null;
+
+ oldVal = (oldRow != null) ? oldRow.value() : null;
if (oldVal == null && readThrough) {
storeLoadedVal = cctx.toCacheObject(cctx.store().load(null, entry.key));
@@ -4759,6 +4817,53 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
/**
+ * Check row expiration and fire expire events if needed.
+ *
+ * @param row Old row.
+ * @return {@code True} if row was expired, {@code False} otherwise.
+ * @throws IgniteCheckedException if failed.
+ */
+ private boolean checkRowExpired(CacheDataRow row) throws IgniteCheckedException {
+ assert row != null;
+
+ if (!(row.expireTime() > 0 && row.expireTime() < U.currentTimeMillis()))
+ return false;
+
+ GridCacheContext cctx = entry.context();
+
+ CacheObject expiredVal = row.value();
+
+ if (cctx.deferredDelete() && !entry.detached() && !entry.isInternal()) {
+ entry.update(null, CU.TTL_ETERNAL, CU.EXPIRE_TIME_ETERNAL, entry.ver, true);
+
+ if (!entry.deletedUnlocked())
+ entry.deletedUnlocked(true);
+ }
+ else
+ entry.markObsolete0(cctx.versions().next(), true, null);
+
+ if (cctx.events().isRecordable(EVT_CACHE_OBJECT_EXPIRED)) {
+ cctx.events().addEvent(entry.partition(),
+ entry.key(),
+ cctx.localNodeId(),
+ null,
+ EVT_CACHE_OBJECT_EXPIRED,
+ null,
+ false,
+ expiredVal,
+ expiredVal != null,
+ null,
+ null,
+ null,
+ true);
+ }
+
+ cctx.continuousQueries().onEntryExpired(entry, entry.key(), expiredVal);
+
+ return true;
+ }
+
+ /**
* @param storeLoadedVal Value loaded from store.
* @param updateExpireTime {@code True} if need update expire time.
* @throws IgniteCheckedException If failed.
@@ -5057,7 +5162,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
entry.update(null, CU.TTL_ETERNAL, CU.EXPIRE_TIME_ETERNAL, newVer, true);
- treeOp = (oldVal == null || readFromStore) ? IgniteTree.OperationType.NOOP :
+ treeOp = (oldRow == null || readFromStore) ? IgniteTree.OperationType.NOOP :
IgniteTree.OperationType.REMOVE;
UpdateOutcome outcome = oldVal != null ? UpdateOutcome.SUCCESS : UpdateOutcome.REMOVE_NO_VAL;
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManager.java
index ff6ed65..8277c2a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheTtlManager.java
@@ -201,6 +201,9 @@ public class GridCacheTtlManager extends GridCacheManagerAdapter {
}
}
+ if(!(cctx.affinityNode() && cctx.ttl().eagerTtlEnabled()))
+ return false; /* Pending tree never contains entries for that cache */
+
boolean more = cctx.offheap().expire(dhtCtx, expireC, amount);
if (more)
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
index bf0de02..10281d6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java
@@ -33,6 +33,7 @@ import javax.cache.Cache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.pagemem.FullPageId;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -85,6 +86,13 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDh
*/
@SuppressWarnings("PublicInnerClass")
public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager {
+ /**
+ * Throttling timeout in millis which avoid excessive PendingTree access on unwind
+ * if there is nothing to clean yet.
+ */
+ public static final long UNWIND_THROTTLING_TIMEOUT = Long.getLong(
+ IgniteSystemProperties.IGNITE_UNWIND_THROTTLING_TIMEOUT, 500L);
+
/** */
protected GridCacheSharedContext ctx;
@@ -106,6 +114,9 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
/** */
protected volatile boolean hasPendingEntries;
+ /** Timestamp when next clean try will be allowed. Used for throttling on per-group basis. */
+ protected volatile long nextCleanTime;
+
/** */
private final GridAtomicLong globalRmvId = new GridAtomicLong(U.currentTimeMillis() * 1000_000);
@@ -1029,13 +1040,36 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
) throws IgniteCheckedException {
assert !cctx.isNear() : cctx.name();
- if (!hasPendingEntries || pendingEntries == null)
+ if (!hasPendingEntries || nextCleanTime > U.currentTimeMillis())
return false;
- GridCacheVersion obsoleteVer = null;
+ assert pendingEntries != null;
+
+ int cleared = expireInternal(cctx, c, amount);
+
+ // Throttle if there is nothing to clean anymore.
+ if (cleared < amount)
+ nextCleanTime = U.currentTimeMillis() + UNWIND_THROTTLING_TIMEOUT;
+ return amount != -1 && cleared >= amount;
+ }
+
+ /**
+ * @param cctx Cache context.
+ * @param c Closure.
+ * @param amount Limit of processed entries by single call, {@code -1} for no limit.
+ * @return cleared entries count.
+ * @throws IgniteCheckedException If failed.
+ */
+ private int expireInternal(
+ GridCacheContext cctx,
+ IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
+ int amount
+ ) throws IgniteCheckedException {
long now = U.currentTimeMillis();
+ GridCacheVersion obsoleteVer = null;
+
GridCursor<PendingRow> cur;
if (grp.sharedGroup())
@@ -1044,17 +1078,17 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
cur = pendingEntries.find(null, new PendingRow(CU.UNDEFINED_CACHE_ID, now, 0));
if (!cur.next())
- return false;
-
- int cleared = 0;
+ return 0;
if (!busyLock.enterBusy())
- return false;
+ return 0;
try {
+ int cleared = 0;
+
do {
if (amount != -1 && cleared > amount)
- return true;
+ return cleared;
PendingRow row = cur.get();
@@ -1076,12 +1110,12 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
cleared++;
}
while (cur.next());
+
+ return cleared;
}
finally {
busyLock.leaveBusy();
}
-
- return false;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtDetachedCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtDetachedCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtDetachedCacheEntry.java
index d02015b..346a992 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtDetachedCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtDetachedCacheEntry.java
@@ -67,8 +67,7 @@ public class GridDhtDetachedCacheEntry extends GridDistributedCacheEntry {
/** {@inheritDoc} */
@Override protected boolean storeValue(CacheObject val,
long expireTime,
- GridCacheVersion ver,
- CacheDataRow oldRow) throws IgniteCheckedException {
+ GridCacheVersion ver) throws IgniteCheckedException {
return false;
// No-op for detached entries, index is updated on primary nodes.
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
index fb41f5c..c953beb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearCacheEntry.java
@@ -458,7 +458,7 @@ public class GridNearCacheEntry extends GridDistributedCacheEntry {
}
/** {@inheritDoc} */
- @Override protected boolean storeValue(CacheObject val, long expireTime, GridCacheVersion ver, CacheDataRow oldRow) {
+ @Override protected boolean storeValue(CacheObject val, long expireTime, GridCacheVersion ver) {
return false;
// No-op: queries are disabled for near cache.
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
index 2b6a821..9861ef9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheOffheapManager.java
@@ -776,9 +776,10 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
) throws IgniteCheckedException {
assert !cctx.isNear() : cctx.name();
- if (!hasPendingEntries)
+ if (!hasPendingEntries || nextCleanTime > U.currentTimeMillis())
return false;
+ // Prevent manager being stopped in the middle of pds operation.
if (!busyLock.enterBusy())
return false;
@@ -791,6 +792,10 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
if (amount != -1 && cleared >= amount)
return true;
}
+
+ // Throttle if there is nothing to clean anymore.
+ if (cleared < amount)
+ nextCleanTime = U.currentTimeMillis() + UNWIND_THROTTLING_TIMEOUT;
}
finally {
busyLock.leaveBusy();
@@ -1159,6 +1164,10 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** */
private volatile CacheDataStore delegate;
+ /** Timestamp when next clean try will be allowed for current partition.
+ * Used for fine-grained throttling on per-partition basis. */
+ private volatile long nextStoreCleanTime;
+
/** */
private final boolean exists;
@@ -1721,12 +1730,12 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/**
- * Removes expired entries from data store.
+ * Try to remove expired entries from data store.
*
* @param cctx Cache context.
* @param c Expiry closure that should be applied to expired entry. See {@link GridCacheTtlManager} for details.
* @param amount Limit of processed entries by single call, {@code -1} for no limit.
- * @return {@code True} if unprocessed expired entries remains.
+ * @return cleared entries count.
* @throws IgniteCheckedException If failed.
*/
public int purgeExpired(GridCacheContext cctx,
@@ -1734,13 +1743,39 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
int amount) throws IgniteCheckedException {
CacheDataStore delegate0 = init0(true);
- if (delegate0 == null || pendingTree == null)
+ long now = U.currentTimeMillis();
+
+ if (delegate0 == null || nextStoreCleanTime > now)
return 0;
+ assert pendingTree != null : "Partition data store was not initialized.";
+
+ int cleared = purgeExpiredInternal(cctx, c, amount);
+
+ // Throttle if there is nothing to clean anymore.
+ if (cleared < amount)
+ nextStoreCleanTime = now + UNWIND_THROTTLING_TIMEOUT;
+
+ return cleared;
+ }
+
+ /**
+ * Removes expired entries from data store.
+ *
+ * @param cctx Cache context.
+ * @param c Expiry closure that should be applied to expired entry. See {@link GridCacheTtlManager} for details.
+ * @param amount Limit of processed entries by single call, {@code -1} for no limit.
+ * @return cleared entries count.
+ * @throws IgniteCheckedException If failed.
+ */
+ private int purgeExpiredInternal(GridCacheContext cctx,
+ IgniteInClosure2X<GridCacheEntryEx, GridCacheVersion> c,
+ int amount) throws IgniteCheckedException {
+
GridDhtLocalPartition part = cctx.topology().localPartition(partId, AffinityTopologyVersion.NONE, false, false);
// Skip non-owned partitions.
- if (part == null || part.state() != OWNING || pendingTree.size() == 0)
+ if (part == null || part.state() != OWNING)
return 0;
cctx.shared().database().checkpointReadLock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java b/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
index a6c9997..cdc4277 100644
--- a/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/cache/store/IgniteCacheExpiryStoreLoadSelfTest.java
@@ -51,7 +51,7 @@ public class IgniteCacheExpiryStoreLoadSelfTest extends GridCacheAbstractSelfTes
private static final int TIME_TO_LIVE = 1000;
/** Additional time to wait expiry process in milliseconds. */
- private static final int WAIT_TIME = 1000;
+ private static final int WAIT_TIME = 1500;
/** {@inheritDoc} */
@Override protected int gridCount() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractMetricsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractMetricsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractMetricsSelfTest.java
index 7948569..721989b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractMetricsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractMetricsSelfTest.java
@@ -177,7 +177,7 @@ public abstract class GridCacheAbstractMetricsSelfTest extends GridCacheAbstract
for (int i = 0; i < KEY_CNT; i++)
cache.put(i, i);
- assertEquals(cache.localMetrics().getAverageRemoveTime(), 0.0, 0.0);
+ assertEquals(0.0, cache.localMetrics().getAverageRemoveTime(), 0.0);
for (int i = 0; i < KEY_CNT; i++)
cache.getAndRemoveAsync(i).get();
@@ -221,7 +221,7 @@ public abstract class GridCacheAbstractMetricsSelfTest extends GridCacheAbstract
for (int i = 0; i < KEY_CNT; i++)
cache.put(i, i);
- assertEquals(cache.localMetrics().getAverageRemoveTime(), 0.0, 0.0);
+ assertEquals(0.0, cache.localMetrics().getAverageRemoveTime(), 0.0);
for (int i = 0; i < KEY_CNT; i++)
cache.remove(i);
@@ -239,7 +239,7 @@ public abstract class GridCacheAbstractMetricsSelfTest extends GridCacheAbstract
cache.put(2, 2);
cache.put(3, 3);
- assertEquals(cache.localMetrics().getAverageRemoveTime(), 0.0, 0.0);
+ assertEquals(0.0, cache.localMetrics().getAverageRemoveTime(), 0.0);
Set<Integer> keys = new HashSet<>(4, 1);
keys.add(1);
@@ -272,7 +272,7 @@ public abstract class GridCacheAbstractMetricsSelfTest extends GridCacheAbstract
}
}
- assertEquals(cache.localMetrics().getAverageRemoveTime(), 0.0, 0.0);
+ assertEquals(0.0, cache.localMetrics().getAverageRemoveTime(), 0.0);
IgniteFuture<?> fut = cache.removeAllAsync(keys);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
index e473d52..c6e3552 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheEntryListenerAbstractTest.java
@@ -896,7 +896,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
U.sleep(700);
if (!eagerTtl())
- assertNull(primaryCache(key, cache.getName()).get(key(key))); // Provoke expire event if eager ttl is disabled.
+ assertNull(primaryCache(key(key), cache.getName()).get(key(key))); // Provoke expire event if eager ttl is disabled.
IgniteCache<Object, Object> cache1 = cache;
@@ -917,7 +917,7 @@ public abstract class IgniteCacheEntryListenerAbstractTest extends IgniteCacheAb
U.sleep(200);
if (!eagerTtl())
- assertNull(primaryCache(key, cache.getName()).get(key(key))); // Provoke expire event if eager ttl is disabled.
+ assertNull(primaryCache(key(key), cache.getName()).get(key(key))); // Provoke expire event if eager ttl is disabled.
evtsLatch.await(5000, MILLISECONDS);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
index ebd9bb8..7693260 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheExpiryPolicyTestSuite.java
@@ -51,8 +51,6 @@ public class IgniteCacheExpiryPolicyTestSuite extends TestSuite {
suite.addTestSuite(IgniteCacheExpiryStoreLoadSelfTest.class);
- suite.addTestSuite(IgniteCacheTtlCleanupSelfTest.class);
-
suite.addTestSuite(IgniteCacheClientNearCacheExpiryTest.class);
suite.addTestSuite(IgniteCacheEntryListenerExpiredEventsTest.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTtlCleanupSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTtlCleanupSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTtlCleanupSelfTest.java
deleted file mode 100644
index 227fe1f..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/expiry/IgniteCacheTtlCleanupSelfTest.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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.expiry;
-
-import java.util.concurrent.TimeUnit;
-import javax.cache.expiry.CreatedExpiryPolicy;
-import javax.cache.expiry.Duration;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
-import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.processors.cache.CacheObjectContext;
-import org.apache.ignite.internal.processors.cache.GridCacheAbstractSelfTest;
-import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLocalPartition;
-import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
-
-/**
- *
- */
-public class IgniteCacheTtlCleanupSelfTest extends GridCacheAbstractSelfTest {
- /** Number of partitions. */
- private static final int PART_NUM = 10;
-
- /** {@inheritDoc} */
- @Override protected int gridCount() {
- return 1;
- }
-
- /** {@inheritDoc} */
- @Override protected CacheConfiguration cacheConfiguration(String igniteInstanceName) throws Exception {
- CacheConfiguration ccfg = super.cacheConfiguration(igniteInstanceName);
-
- ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
-
- ccfg.setAffinity(new RendezvousAffinityFunction(false, 10));
-
- ccfg.setNearConfiguration(null);
-
- return ccfg;
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testDeferredDeleteTtl() throws Exception {
- IgniteCache<Object, Object> cache = grid(0).cache(DEFAULT_CACHE_NAME)
- .withExpiryPolicy(new CreatedExpiryPolicy(new Duration(TimeUnit.SECONDS, 5)));
-
- int cnt = GridDhtLocalPartition.MAX_DELETE_QUEUE_SIZE / PART_NUM + 100;
-
- for (long i = 0; i < cnt; i++)
- grid(0).cache(DEFAULT_CACHE_NAME).put(i * PART_NUM, i);
-
- for (int i = 0; i < cnt; i++)
- cache.put(i * PART_NUM, i);
-
- // Wait 5 seconds.
- Thread.sleep(6_000);
-
- assertEquals(cnt, grid(0).cache(DEFAULT_CACHE_NAME).size());
-
- GridCacheAdapter<Object, Object> cacheAdapter = ((IgniteKernal)grid(0)).internalCache(DEFAULT_CACHE_NAME);
-
- IgniteCacheObjectProcessor cacheObjects = cacheAdapter.context().cacheObjects();
-
- CacheObjectContext cacheObjCtx = cacheAdapter.context().cacheObjectContext();
-
- for (int i = 0; i < 100; i++)
- assertNull(cacheAdapter.map().getEntry(cacheAdapter.context(), cacheObjects.toCacheKeyObject(cacheObjCtx, null, i, true)));
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationsWithExpirationTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationsWithExpirationTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationsWithExpirationTest.java
index 76d2376..6a04126 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationsWithExpirationTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationsWithExpirationTest.java
@@ -50,12 +50,10 @@ public class CacheOperationsWithExpirationTest extends GridCommonAbstractTest {
/**
* @param atomicityMode Atomicity mode.
- * @param offheapMem Offheap memory size.
* @param idx Indexing enabled flag.
* @return Cache configuration.
*/
private CacheConfiguration<String, TestIndexedType> cacheConfiguration(CacheAtomicityMode atomicityMode,
- long offheapMem,
boolean idx) {
CacheConfiguration<String, TestIndexedType> ccfg = new CacheConfiguration<>(DEFAULT_CACHE_NAME);
@@ -81,14 +79,14 @@ public class CacheOperationsWithExpirationTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testAtomicIndexEnabled() throws Exception {
- concurrentPutGetRemoveExpireAndQuery(cacheConfiguration(ATOMIC, 0, true));
+ concurrentPutGetRemoveExpireAndQuery(cacheConfiguration(ATOMIC, true));
}
/**
* @throws Exception If failed.
*/
public void testAtomic() throws Exception {
- concurrentPutGetRemoveExpireAndQuery(cacheConfiguration(ATOMIC, 0, false));
+ concurrentPutGetRemoveExpireAndQuery(cacheConfiguration(ATOMIC, false));
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java
index cb038fa..a845aaa 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheAbstractQuerySelfTest.java
@@ -458,7 +458,7 @@ public abstract class IgniteCacheAbstractQuerySelfTest extends GridCommonAbstrac
assertEquals(1, res.getValue().intValue());
- U.sleep(800); // Less than minimal amount of time that must pass before a cache entry is considered expired.
+ U.sleep(300); // Less than minimal amount of time that must pass before a cache entry is considered expired.
qry = cache.query(new SqlQuery<Integer, Integer>(Integer.class, "1=1")).getAll();
@@ -466,7 +466,7 @@ public abstract class IgniteCacheAbstractQuerySelfTest extends GridCommonAbstrac
assertEquals(1, res.getValue().intValue());
- U.sleep(1200); // No expiry guarantee here. Test should be refactored in case of fails.
+ U.sleep(1800); // No expiry guarantee here. Test should be refactored in case of fails.
qry = cache.query(new SqlQuery<Integer, Integer>(Integer.class, "1=1")).getAll();
http://git-wip-us.apache.org/repos/asf/ignite/blob/a06c47f7/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCachePartitionedQueryP2PDisabledSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCachePartitionedQueryP2PDisabledSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCachePartitionedQueryP2PDisabledSelfTest.java
index 7e5d74d..8925b01 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCachePartitionedQueryP2PDisabledSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/IgniteCachePartitionedQueryP2PDisabledSelfTest.java
@@ -27,7 +27,7 @@ public class IgniteCachePartitionedQueryP2PDisabledSelfTest extends IgniteCacheP
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration c = super.getConfiguration(igniteInstanceName);
- c.setPeerClassLoadingEnabled(true);
+ c.setPeerClassLoadingEnabled(false);
return c;
}