diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java index bda9d3fbdc198..ac208bd91ab5e 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java @@ -114,6 +114,11 @@ public interface GridCacheEntryEx { */ public boolean hasValue(); + /** + * @return {@code True} if has value or value bytes and the value is not expired yet. + */ + public boolean hasNonExpiredValue(); + /** * @param val New value. * @param ttl Time to live. 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 021fdd0408d48..7662940494d95 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 @@ -993,6 +993,31 @@ protected void recordNodeId(UUID nodeId, AffinityTopologyVersion topVer) { @Nullable GridCacheVersion dhtVer, @Nullable Long updateCntr ) throws IgniteCheckedException, GridCacheEntryRemovedException { + // Explicit expire time is already reached, the value is expired: remove the entry instead of storing + // an already expired row. A row with expire time in the past must never be written to the row store, + // otherwise the (expireTime, link) pair, already processed by the TTL cleanup worker, can be recreated + // and the pending entries tree becomes inconsistent (see IGNITE-25194). + if (CU.isExpired(drExpireTime)) { + return innerRemove( + tx, + evtNodeId, + affNodeId, + retval, + evt, + metrics, + keepBinary, + keepBinaryInInterceptor, + oldValPresent, + oldVal, + topVer, + drType, + explicitVer, + taskName, + dhtVer, + updateCntr + ); + } + CacheObject old; final boolean valid = valid(tx != null ? tx.topologyVersion() : topVer); @@ -1077,6 +1102,12 @@ else if (interceptorVal != val0) if (ttl == -1L) { ttl = ttlExtras(); expireTime = expireTimeExtras(); + + // The current value is already expired (but not cleaned up yet), so the update is actually + // a creation: calculate a new expire time instead of inheriting the expired one. A row with + // expire time in the past must never be written to the row store (see IGNITE-25194). + if (CU.isExpired(expireTime)) + expireTime = CU.toExpireTime(ttl); } else expireTime = CU.toExpireTime(ttl); @@ -2441,6 +2472,18 @@ private boolean checkExpired() throws IgniteCheckedException { } } + /** {@inheritDoc} */ + @Override public final boolean hasNonExpiredValue() { + lockEntry(); + + try { + return hasValueUnlocked() && !CU.isExpired(expireTimeExtras()); + } + finally { + unlockEntry(); + } + } + /** * @return {@code True} if this entry has value. */ @@ -2526,6 +2569,18 @@ private boolean skipInterceptor(@Nullable GridCacheVersion explicitVer) { long expTime = expireTime < 0 ? CU.toExpireTime(ttl) : expireTime; + CacheDataRow expiredRow = null; + + // The value is already expired: store it as removed instead of storing an already expired row. + // A row with expire time in the past must never be written to the row store (see IGNITE-25194). + if (val != null && CU.isExpired(expTime)) { + val = null; + + // Pre-created row is already inserted to the row store and must be removed. + expiredRow = row; + row = null; + } + val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx); final boolean unswapped = ((flags & IS_UNSWAPPED_MASK) != 0); @@ -2583,6 +2638,12 @@ else if (val == null) else update = storeValue(val, expTime, ver, p, row); + // If update is not applied, the pre-created row is removed by the caller (see CacheDataStore#insertRows). + if (expiredRow != null && update) { + cctx.offheap().dataStore(localPartition()).rowStore() + .removeRow(expiredRow.link(), cctx.group().statisticsHolderData()); + } + if (update) { update(val, expTime, ttl, ver, true); @@ -4354,9 +4415,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI /** */ private CacheDataRow oldRow; - /** */ - private boolean oldRowExpiredFlag; - /** */ private IgniteTree.OperationType treeOp = IgniteTree.OperationType.PUT; @@ -4379,14 +4437,14 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI /** {@inheritDoc} */ @Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException { + this.oldRow = oldRow; + if (oldRow != null) { oldRow.key(entry.key); oldRow = checkRowExpired(oldRow); } - this.oldRow = oldRow; - if (predicate != null && !predicate.apply(oldRow)) { treeOp = IgniteTree.OperationType.NOOP; @@ -4426,11 +4484,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI return oldRow; } - /** {@inheritDoc} */ - @Override public boolean oldRowExpiredFlag() { - return oldRowExpiredFlag; - } - /** * Checks row for expiration and fire expire events if needed. * @@ -4476,8 +4529,6 @@ private CacheDataRow checkRowExpired(CacheDataRow row) throws IgniteCheckedExcep entry.updatePlatformCache(null, null); - oldRowExpiredFlag = true; - return null; } } @@ -4637,11 +4688,6 @@ private static class AtomicCacheUpdateClosure implements IgniteCacheOffheapManag return oldRow; } - /** {@inheritDoc} */ - @Override public boolean oldRowExpiredFlag() { - return oldRowExpiredFlag; - } - /** {@inheritDoc} */ @Override public CacheDataRow newRow() { return newRow; @@ -4994,7 +5040,7 @@ else if (newSysTtl == CU.TTL_ZERO) { newSysExpireTime = newExpireTime = conflictCtx.expireTime(); } - if (newExpireTime > 0 && newExpireTime < U.currentTimeMillis()) { + if (CU.isExpired(newExpireTime)) { op = DELETE; writeObj = null; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java index 7f0b4d36bd11a..8a0422dbe9179 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java @@ -1067,6 +1067,14 @@ public static String cacheOrGroupName(CacheConfiguration ccfg) { return ccfg.getGroupName() == null ? ccfg.getName() : ccfg.getGroupName(); } + /** + * @param expireTime Expire time. + * @return {@code True} if the given expire time is set and already reached. + */ + public static boolean isExpired(long expireTime) { + return expireTime > 0 && expireTime <= U.currentTimeMillis(); + } + /** * Convert TTL to expire time. * diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java index 3247e718ad6cf..52aa494c4ffe3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java @@ -418,12 +418,6 @@ interface OffheapInvokeClosure extends IgniteTree.InvokeClosure { * @return Old row. */ @Nullable public CacheDataRow oldRow(); - - /** - * Flag that indicates if oldRow was expired during invoke. - * @return {@code true} if old row was expired, {@code false} otherwise. - */ - public boolean oldRowExpiredFlag(); } /** 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 c97aefd68213f..d6424e51f7126 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 @@ -997,7 +997,8 @@ private long allocateForTree() throws IgniteCheckedException { try { batch.add(new DataRowCacheAware(info.key(), - info.value(), + // Already expired value is stored as removed, don't insert it to the row store (see IGNITE-25194). + CU.isExpired(info.expireTime()) ? null : info.value(), info.version(), part.id(), info.expireTime(), info.cacheId(), grp.storeCacheIdInDataPage())); @@ -1507,9 +1508,7 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo case PUT: { assert c.newRow() != null : c; - CacheDataRow oldRow = c.oldRow(); - - finishUpdate(cctx, c.newRow(), oldRow, c.oldRowExpiredFlag()); + finishUpdate(cctx, c.newRow(), c.oldRow()); break; } @@ -1666,21 +1665,12 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo * @param oldRow Old row if available. * @throws IgniteCheckedException If failed. */ - private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow) - throws IgniteCheckedException { - finishUpdate(cctx, newRow, oldRow, false); - } - - /** - * @param cctx Cache context. - * @param newRow New row. - * @param oldRow Old row if available. - * @param oldRowExpired Old row expiration flag - * @throws IgniteCheckedException If failed. - */ - private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow, boolean oldRowExpired) - throws IgniteCheckedException { - if (oldRow == null && !oldRowExpired) + private void finishUpdate( + GridCacheContext cctx, + CacheDataRow newRow, + @Nullable CacheDataRow oldRow + ) throws IgniteCheckedException { + if (oldRow == null) incrementSize(cctx.cacheId()); GridCacheQueryManager qryMgr = cctx.queries(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java index 77e5d9aa97ad3..5bace15f75289 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java @@ -584,6 +584,11 @@ else if (conflictCtx.isMerge()) { // Nullify explicit version so that innerSet/innerRemove will work as usual. explicitVer = null; + // Explicit expire time is already reached: remove instead of storing an already + // expired value (the same way as for TTL_ZERO, see IGNITE-25194). + if ((op == CREATE || op == UPDATE) && CU.isExpired(txEntry.conflictExpireTime())) + op = DELETE; + GridCacheVersion dhtVer = cached.isNear() ? writeVersion() : null; if (!near() && cacheCtx.group().logDataRecords() && diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java index 67e72fd1fb162..8321987e6d5db 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java @@ -604,7 +604,7 @@ protected GridCacheEntryEx entryEx(GridCacheContext cacheCtx, IgniteTxKey key, A if (expiry != null) { txEntry.cached().unswap(false); - Duration duration = cached.hasValue() ? + Duration duration = cached.hasNonExpiredValue() ? expiry.getExpiryForUpdate() : expiry.getExpiryForCreation(); txEntry.ttl(CU.toTtl(duration)); @@ -624,7 +624,8 @@ protected GridCacheEntryEx entryEx(GridCacheContext cacheCtx, IgniteTxKey key, A ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry); if (expiry != null) { - Duration duration = cached.hasValue() ? + // Expired (but not cleaned up yet) value is treated as absent. + Duration duration = cached.hasNonExpiredValue() ? expiry.getExpiryForUpdate() : expiry.getExpiryForCreation(); long ttl = CU.toTtl(duration); @@ -676,6 +677,11 @@ else if (conflictCtx.isUseNew()) { txEntry.conflictVersion(explicitVer); } + // Explicit expire time is already reached: remove instead of storing an already + // expired value (the same way as for TTL_ZERO, see IGNITE-25194). + if ((op == CREATE || op == UPDATE) && CU.isExpired(txEntry.conflictExpireTime())) + op = DELETE; + if (dhtVer == null) dhtVer = explicitVer != null ? explicitVer : writeVersion(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java index d4971a8b26ece..ce0905357a88f 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java @@ -298,6 +298,11 @@ void recheckLock() { return val != null; } + /** @inheritDoc */ + @Override public boolean hasNonExpiredValue() { + return hasValue(); + } + /** @inheritDoc */ @Override public CacheObject rawPut(CacheObject val, long ttl) { CacheObject old = this.val; diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/ttl/CacheSizeTtlTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/ttl/CacheSizeTtlTest.java index 30f59229d48fc..39270d1050280 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/ttl/CacheSizeTtlTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/ttl/CacheSizeTtlTest.java @@ -22,16 +22,21 @@ import java.util.stream.IntStream; import javax.cache.expiry.Duration; import javax.cache.expiry.ModifiedExpiryPolicy; +import javax.cache.expiry.TouchedExpiryPolicy; import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.IgniteSystemProperties; import org.apache.ignite.Ignition; import org.apache.ignite.cache.CachePeekMode; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder; import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.WithSystemProperty; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.jetbrains.annotations.NotNull; import org.junit.Test; @@ -39,6 +44,7 @@ import static java.util.Collections.singleton; import static java.util.concurrent.TimeUnit.SECONDS; import static org.apache.ignite.cache.CacheMode.REPLICATED; +import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ; /** * Tests for cache.size() with ttl enabled. @@ -57,6 +63,26 @@ public class CacheSizeTtlTest extends GridCommonAbstractTest { stopAllGrids(); } + /** */ + @Test + @WithSystemProperty(key = IgniteSystemProperties.IGNITE_UNWIND_THROTTLING_TIMEOUT, value = "10000") + public void testEntriesLeak() throws Exception { + IgniteEx srv = startGrid(getConfiguration().setIncludeEventTypes(EVT_CACHE_OBJECT_READ)); + + srv.events().localListen(evt -> { + doSleep(2000L); + return true; + }, EVT_CACHE_OBJECT_READ); + + IgniteCache cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME) + .withExpiryPolicy(new TouchedExpiryPolicy(new Duration(SECONDS, 5))); + + cache.put(1, 1); + cache.get(1); + + assertTrue(GridTestUtils.waitForCondition(() -> cache.size(CachePeekMode.PRIMARY) == 0, 15_000L)); + } + /** * Tests that cache.size() works correctly for massive amount of puts and ttl. */