You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2019/02/24 11:58:46 UTC
[ignite] branch ignite-invokeAll-backup created (now f37831b)
This is an automated email from the ASF dual-hosted git repository.
sboikov pushed a change to branch ignite-invokeAll-backup
in repository https://gitbox.apache.org/repos/asf/ignite.git.
at f37831b invokeAll
This branch includes the following new commits:
new f37831b invokeAll
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[ignite] 01/01: invokeAll
Posted by sb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sboikov pushed a commit to branch ignite-invokeAll-backup
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit f37831b307cfc890bb5dc374c0023ccac7be76af
Author: sboikov <sb...@apache.org>
AuthorDate: Sun Feb 24 14:09:23 2019 +0300
invokeAll
---
.../processors/cache/GridCacheMapEntry.java | 11 +-
.../cache/IgniteCacheOffheapManager.java | 7 +-
.../cache/IgniteCacheOffheapManagerImpl.java | 27 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 762 ++++++++++++++-------
.../cache/persistence/GridCacheOffheapManager.java | 9 +-
.../cache/persistence/tree/BPlusTree.java | 6 +-
.../processors/cache/tree/SearchRowEx.java | 45 ++
7 files changed, 599 insertions(+), 268 deletions(-)
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 23b14b6..acd4349 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
@@ -2239,7 +2239,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
) throws IgniteCheckedException, GridCacheEntryRemovedException, GridClosureException {
assert cctx.atomic() && !detached();
- if (!primary && !isNear())
+ if (!primary && !isNear() && c == null)
ensureFreeSpace();
if (!primary) {
@@ -5903,7 +5903,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
private boolean wasIntercepted;
/** */
- AtomicCacheUpdateClosure(
+ public AtomicCacheUpdateClosure(
GridCacheMapEntry entry,
AffinityTopologyVersion topVer,
GridCacheVersion newVer,
@@ -5960,6 +5960,13 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
}
}
+ /**
+ * @return Update result.
+ */
+ public GridCacheUpdateAtomicResult updateResult() {
+ return updateRes;
+ }
+
/** {@inheritDoc} */
@Nullable @Override public CacheDataRow oldRow() {
return oldRow;
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 870d99f..9455c2a 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
@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.function.Function;
import javax.cache.Cache;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
@@ -200,7 +201,7 @@ public interface IgniteCacheOffheapManager {
public void invokeAll(GridCacheContext cctx,
GridDhtLocalPartition part,
Collection<? extends CacheSearchRow> rows,
- Map<? extends CacheSearchRow, ? extends OffheapInvokeClosure> map)
+ Function<CacheSearchRow, OffheapInvokeClosure> closures)
throws IgniteCheckedException;
/**
@@ -899,7 +900,7 @@ public interface IgniteCacheOffheapManager {
* @return Cache search row.
* @throws IgniteCheckedException If failed.
*/
- public CacheSearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException;
+ public CacheSearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key, Object data) throws IgniteCheckedException;
/**
* @return Rows comparator.
@@ -915,7 +916,7 @@ public interface IgniteCacheOffheapManager {
*/
public void invokeAll(GridCacheContext cctx,
Collection<? extends CacheSearchRow> rows,
- Map<? extends CacheSearchRow, ? extends OffheapInvokeClosure> map)
+ Function<CacheSearchRow, OffheapInvokeClosure> closures)
throws IgniteCheckedException;
/**
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 3ae5b81..0a621dc 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 java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
import javax.cache.Cache;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
@@ -80,6 +81,7 @@ import org.apache.ignite.internal.processors.cache.tree.PendingEntriesTree;
import org.apache.ignite.internal.processors.cache.tree.PendingRow;
import org.apache.ignite.internal.processors.cache.tree.RowLinkIO;
import org.apache.ignite.internal.processors.cache.tree.SearchRow;
+import org.apache.ignite.internal.processors.cache.tree.SearchRowEx;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccDataRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateDataRow;
import org.apache.ignite.internal.processors.cache.tree.mvcc.data.MvccUpdateResult;
@@ -456,9 +458,9 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
GridCacheContext cctx,
GridDhtLocalPartition part,
Collection<? extends CacheSearchRow> rows,
- Map<? extends CacheSearchRow, ? extends OffheapInvokeClosure> map)
+ Function<CacheSearchRow, OffheapInvokeClosure> closures)
throws IgniteCheckedException {
- dataStore(part).invokeAll(cctx, rows, map);
+ dataStore(part).invokeAll(cctx, rows, closures);
}
/** {@inheritDoc} */
@@ -1650,8 +1652,10 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public SearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key) {
- return new SearchRow(grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID, key);
+ @Override public SearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key, Object data) {
+ int cacheId = grp.sharedGroup() ? cctx.cacheId() : CU.UNDEFINED_CACHE_ID;
+
+ return data != null ? new SearchRowEx<>(cacheId, key, data) : new SearchRow(cacheId, key);
}
/** {@inheritDoc} */
@@ -1660,7 +1664,10 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
}
/** {@inheritDoc} */
- @Override public void invokeAll(GridCacheContext cctx, Collection<? extends CacheSearchRow> rows, Map<? extends CacheSearchRow, ? extends OffheapInvokeClosure> map) throws IgniteCheckedException {
+ @Override public void invokeAll(GridCacheContext cctx,
+ Collection<? extends CacheSearchRow> rows,
+ Function<CacheSearchRow, OffheapInvokeClosure> closures)
+ throws IgniteCheckedException {
if (!busyLock.enterBusy())
throw new NodeStoppingException("Operation has been cancelled (node is stopping).");
@@ -1668,16 +1675,18 @@ public class IgniteCacheOffheapManagerImpl implements IgniteCacheOffheapManager
RuntimeException err = null;
try {
- dataTree.invokeAll(rows.iterator(), CacheDataRowAdapter.RowData.NO_KEY, map::get);
+ dataTree.invokeAll(rows.iterator(), CacheDataRowAdapter.RowData.NO_KEY, closures);
}
catch (UnregisteredClassException | UnregisteredBinaryTypeException clsErr) {
err = clsErr;
}
- for (Map.Entry<? extends CacheSearchRow, ? extends OffheapInvokeClosure> e : map.entrySet()) {
+ for (CacheSearchRow row : rows) {
+ OffheapInvokeClosure c = closures.apply(row);
+
// Update could be interrupted in the middle, finish update only for processed entries.
- if (e.getValue().operationType() != null)
- finishInvoke(cctx, e.getKey().key(), e.getValue());
+ if (c.operationType() != null)
+ finishInvoke(cctx, row.key(), c);
}
if (err != null)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 329165a..cf23e25 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -63,6 +63,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheMapEntry;
import org.apache.ignite.internal.processors.cache.GridCacheMapEntry.AtomicCacheBatchUpdateClosure;
+import org.apache.ignite.internal.processors.cache.GridCacheMapEntry.AtomicCacheUpdateClosure;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheUpdateAtomicResult;
@@ -95,6 +96,7 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheSearchRow;
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryListener;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
+import org.apache.ignite.internal.processors.cache.tree.SearchRowEx;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionEx;
@@ -1760,6 +1762,17 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
DhtAtomicUpdateResult updDhtRes = new DhtAtomicUpdateResult();
try {
+ Map<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> byPart = null;
+
+ if (req.size() > 1 && // Several keys ...
+ writeThrough() && !req.skipStore() && // and store is enabled ...
+ !ctx.store().isLocal() && // and this is not local store ...
+ // (conflict resolver should be used for local store)
+ !ctx.dr().receiveEnabled() // and no DR.
+ ) {
+ byPart = new HashMap<>();
+ }
+
while (true) {
try {
GridDhtPartitionTopology top = topology();
@@ -1855,7 +1868,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
}
- update(node, locked, req, res, updDhtRes);
+ update(node, locked, req, res, updDhtRes, byPart);
dhtFut = updDhtRes.dhtFuture();
deleted = updDhtRes.deleted();
@@ -1964,15 +1977,16 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param req Request.
* @param res Response.
* @param dhtUpdRes DHT update result
- * @return Operation result.
* @throws GridCacheEntryRemovedException If got obsolete entry.
*/
- private DhtAtomicUpdateResult update(
+ private void update(
ClusterNode node,
List<GridDhtCacheEntry> locked,
GridNearAtomicAbstractUpdateRequest req,
GridNearAtomicUpdateResponse res,
- DhtAtomicUpdateResult dhtUpdRes)
+ DhtAtomicUpdateResult dhtUpdRes,
+ @Nullable Map<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> byPart
+ )
throws GridCacheEntryRemovedException
{
GridDhtPartitionTopology top = topology();
@@ -2006,12 +2020,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
ctx.group().listenerLock().readLock().lock();
try {
- if (req.size() > 1 && // Several keys ...
- writeThrough() && !req.skipStore() && // and store is enabled ...
- !ctx.store().isLocal() && // and this is not local store ...
- // (conflict resolver should be used for local store)
- !ctx.dr().receiveEnabled() // and no DR.
- ) {
+ if (byPart != null) {
// This method can only be used when there are no replicated entries in the batch.
updateWithBatch(node,
hasNear,
@@ -2023,7 +2032,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
taskName,
expiry,
sndPrevVal,
- dhtUpdRes);
+ dhtUpdRes,
+ byPart);
if (req.operation() == TRANSFORM)
retVal = dhtUpdRes.returnValue();
@@ -2077,8 +2087,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
}
dhtUpdRes.expiryPolicy(expiry);
-
- return dhtUpdRes;
}
/**
@@ -2109,7 +2117,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
final String taskName,
@Nullable final IgniteCacheExpiryPolicy expiry,
final boolean sndPrevVal,
- final DhtAtomicUpdateResult dhtUpdRes
+ final DhtAtomicUpdateResult dhtUpdRes,
+ final Map<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> byPart
) throws GridCacheEntryRemovedException {
assert !ctx.dr().receiveEnabled(); // Cannot update in batches during DR due to possible conflicts.
assert !req.returnValue() || req.operation() == TRANSFORM; // Should not request return values for putAll.
@@ -2270,22 +2279,17 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Update previous batch.
if (putMap != null) {
updatePartialBatch(
- hasNear,
firstEntryIdx,
filtered,
ver,
- node,
writeVals,
putMap,
null,
- entryProcessorMap,
req,
res,
- replicate,
dhtUpdRes,
- taskName,
expiry,
- sndPrevVal);
+ byPart);
firstEntryIdx = i;
@@ -2317,22 +2321,17 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Update previous batch.
if (rmvKeys != null) {
updatePartialBatch(
- hasNear,
firstEntryIdx,
filtered,
ver,
- node,
null,
null,
rmvKeys,
- entryProcessorMap,
req,
res,
- replicate,
dhtUpdRes,
- taskName,
expiry,
- sndPrevVal);
+ byPart);
firstEntryIdx = i;
@@ -2440,27 +2439,65 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Store final batch.
if (putMap != null || rmvKeys != null) {
updatePartialBatch(
- hasNear,
firstEntryIdx,
filtered,
ver,
- node,
writeVals,
putMap,
rmvKeys,
- entryProcessorMap,
req,
res,
- replicate,
dhtUpdRes,
- taskName,
expiry,
- sndPrevVal);
+ byPart);
}
else
assert filtered.isEmpty();
dhtUpdRes.returnValue(invokeRes);
+
+ AffinityAssignment affAssignment = ctx.affinity().assignment(req.topologyVersion());
+
+ GridDrType drType = replicate ? DR_PRIMARY : DR_NONE;
+
+ for (Map.Entry<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> e0 : byPart.entrySet()) {
+ try {
+ Map<CacheSearchRow, AtomicCacheBatchUpdateClosure> map = e0.getValue();
+
+ ctx.offheap().invokeAll(ctx, e0.getKey(), map.keySet(), map::get);
+
+ for (Map.Entry<CacheSearchRow, AtomicCacheBatchUpdateClosure> e : map.entrySet()) {
+ AtomicCacheBatchUpdateClosure c = e.getValue();
+
+ updateSingleEntryPartialBatch(
+ (GridDhtCacheEntry)c.entry(),
+ c,
+ ver,
+ (CacheObject)c.writeValue(),
+ entryProcessorMap,
+ node,
+ hasNear,
+ taskName,
+ expiry,
+ drType,
+ req,
+ res,
+ dhtUpdRes,
+ c.reqIdx,
+ affAssignment,
+ sndPrevVal);
+ }
+ }
+ catch (GridCacheEntryRemovedException e) {
+ assert false : "Entry cannot become obsolete while holding lock.";
+
+ e.printStackTrace();
+ }
+ catch (IgniteCheckedException e) {
+ for (CacheSearchRow row : e0.getValue().keySet())
+ res.addFailedKey(row.key(), e);
+ }
+ }
}
/**
@@ -2614,7 +2651,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
ctx.disableTriggeringCacheInterceptorOnConflict()
);
- map.put(dataStore.createSearchRow(ctx, entry.key()), c);
+ map.put(dataStore.createSearchRow(ctx, entry.key(), null), c);
}
catch (IgniteCheckedException e) {
res.addFailedKey(entry.key(), e);
@@ -2628,7 +2665,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
Map<CacheSearchRow, AtomicCacheBatchUpdateClosure> map = e0.getValue();
try {
- ctx.offheap().invokeAll(ctx, e0.getKey(), map.keySet(), map);
+ ctx.offheap().invokeAll(ctx, e0.getKey(), map.keySet(), map::get);
}
catch (UnregisteredClassException | UnregisteredBinaryTypeException e) {
err = e;
@@ -2915,22 +2952,17 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
* @param sndPrevVal If {@code true} sends previous value to backups.
*/
private void updatePartialBatch(
- final boolean hasNear,
final int firstEntryIdx,
final List<GridDhtCacheEntry> entries,
final GridCacheVersion ver,
- final ClusterNode nearNode,
@Nullable final List<CacheObject> writeVals,
@Nullable final Map<KeyCacheObject, CacheObject> putMap,
@Nullable final Collection<KeyCacheObject> rmvKeys,
- @Nullable final Map<KeyCacheObject, EntryProcessor<Object, Object, Object>> entryProcessorMap,
final GridNearAtomicAbstractUpdateRequest req,
final GridNearAtomicUpdateResponse res,
- final boolean replicate,
final DhtAtomicUpdateResult dhtUpdRes,
- final String taskName,
@Nullable final IgniteCacheExpiryPolicy expiry,
- final boolean sndPrevVal
+ Map<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> byPart
) {
assert putMap == null ^ rmvKeys == null;
@@ -2971,17 +3003,10 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
op = DELETE;
}
- AffinityAssignment affAssignment = ctx.affinity().assignment(topVer);
-
- GridDrType drType = replicate ? DR_PRIMARY : DR_NONE;
-
Collection<Object> failedToUnwrapKeys = null;
int cnt = entries.size();
- Map<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> byPart =
- cnt > 1 ? new HashMap<>() : null;
-
// Avoid iterator creation.
for (int i = 0; i < cnt; i++) {
GridDhtCacheEntry entry = entries.get(i);
@@ -3015,119 +3040,49 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
continue;
}
- try {
- // We are holding java-level locks on entries at this point.
- CacheObject writeVal = op == UPDATE ? writeVals.get(i) : null;
+ // We are holding java-level locks on entries at this point.
+ CacheObject writeVal = op == UPDATE ? writeVals.get(i) : null;
- assert writeVal != null || op == DELETE : "null write value found.";
+ assert writeVal != null || op == DELETE : "null write value found.";
- if (byPart != null) {
- GridDhtLocalPartition part = entry.localPartition();
+ GridDhtLocalPartition part = entry.localPartition();
- TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure> map = byPart.get(part);
+ TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure> map = byPart.get(part);
- IgniteCacheOffheapManager.CacheDataStore dataStore = ctx.offheap().dataStore(part);
+ IgniteCacheOffheapManager.CacheDataStore dataStore = ctx.offheap().dataStore(part);
- if (map == null)
- byPart.put(part, map = new TreeMap<>(dataStore.rowsComparator()));
+ if (map == null)
+ byPart.put(part, map = new TreeMap<>(dataStore.rowsComparator()));
- AtomicCacheBatchUpdateClosure c = new AtomicCacheBatchUpdateClosure(
- firstEntryIdx + i,
- entry,
- topVer,
- ver,
- op,
- writeVal,
- req.invokeArguments(),
- /*read-through*/false,
- /*write-through*/false,
- req.keepBinary(),
- expiry,
- /*primary*/true,
- /*verCheck*/false,
- req.filter(),
- CU.TTL_NOT_CHANGED,
- CU.EXPIRE_TIME_CALCULATE,
- null,
- /*conflictResolve*/false,
- false,
- null,
- ctx.disableTriggeringCacheInterceptorOnConflict()
- );
-
- map.put(dataStore.createSearchRow(ctx, entry.key()), c);
- }
- else {
- updateSingleEntryPartialBatch(entry,
- null,
- op,
- ver,
- writeVal,
- entryProcessorMap,
- nearNode,
- hasNear,
- taskName,
- expiry,
- drType,
- req,
- res,
- dhtUpdRes,
- i + firstEntryIdx,
- affAssignment,
- sndPrevVal);
- }
- }
- catch (GridCacheEntryRemovedException e) {
- assert false : "Entry cannot become obsolete while holding lock.";
+ AtomicCacheBatchUpdateClosure c = new AtomicCacheBatchUpdateClosure(
+ firstEntryIdx + i,
+ entry,
+ topVer,
+ ver,
+ op,
+ writeVal,
+ req.invokeArguments(),
+ /*read-through*/false,
+ /*write-through*/false,
+ req.keepBinary(),
+ expiry,
+ /*primary*/true,
+ /*verCheck*/false,
+ req.filter(),
+ CU.TTL_NOT_CHANGED,
+ CU.EXPIRE_TIME_CALCULATE,
+ null,
+ /*conflictResolve*/false,
+ false,
+ null,
+ ctx.disableTriggeringCacheInterceptorOnConflict()
+ );
- e.printStackTrace();
- }
+ map.put(dataStore.createSearchRow(ctx, entry.key(), null), c);
dhtUpdRes.processedEntriesCount(firstEntryIdx + i + 1);
}
- if (byPart != null) {
- for (Map.Entry<GridDhtLocalPartition, TreeMap<CacheSearchRow, AtomicCacheBatchUpdateClosure>> e0 : byPart.entrySet()) {
- try {
- Map<CacheSearchRow, AtomicCacheBatchUpdateClosure> map = e0.getValue();
-
- ctx.offheap().invokeAll(ctx, e0.getKey(), map.keySet(), map);
-
- for (Map.Entry<CacheSearchRow, AtomicCacheBatchUpdateClosure> e : map.entrySet()) {
- AtomicCacheBatchUpdateClosure c = e.getValue();
-
- updateSingleEntryPartialBatch(
- (GridDhtCacheEntry)c.entry(),
- c,
- op,
- ver,
- (CacheObject)c.writeValue(),
- entryProcessorMap,
- nearNode,
- hasNear,
- taskName,
- expiry,
- drType,
- req,
- res,
- dhtUpdRes,
- c.reqIdx,
- affAssignment,
- sndPrevVal);
- }
- }
- catch (GridCacheEntryRemovedException e) {
- assert false : "Entry cannot become obsolete while holding lock.";
-
- e.printStackTrace();
- }
- catch (IgniteCheckedException e) {
- for (CacheSearchRow row : e0.getValue().keySet())
- res.addFailedKey(row.key(), e);
- }
- }
- }
-
if (failedToUnwrapKeys != null) {
log.warning("Failed to get values of keys: " + failedToUnwrapKeys +
" (the binary objects will be used instead).");
@@ -3151,7 +3106,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
private void updateSingleEntryPartialBatch(
GridDhtCacheEntry entry,
@Nullable AtomicCacheBatchUpdateClosure c,
- GridCacheOperation op,
GridCacheVersion ver,
@Nullable final CacheObject writeVal,
@Nullable final Map<KeyCacheObject, EntryProcessor<Object, Object, Object>> entryProcessorMap,
@@ -3177,7 +3131,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
// Get readers before innerUpdate (reader cleared after remove).
GridDhtCacheEntry.ReaderId[] readers = c != null ? c.rdrs : entry.readersLocked();
- assert writeVal != null || op == DELETE : "null write value found.";
+ GridCacheOperation op = writeVal == null ? DELETE : UPDATE;
GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
c,
@@ -3583,135 +3537,449 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
sendNearUpdateReply(nodeId, res);
}
- /**
- * @param nodeId Sender node ID.
- * @param req Dht atomic update request.
- */
- private void processDhtAtomicUpdateRequest(UUID nodeId, GridDhtAtomicAbstractUpdateRequest req) {
- assert Thread.currentThread().getName().startsWith("sys-stripe-") : Thread.currentThread().getName();
+ private void dhtAtomicUpdateRequestUpdateBatch(
+ UUID nodeId,
+ GridDhtAtomicAbstractUpdateRequest req,
+ @Nullable GridDhtAtomicNearResponse nearRes,
+ String taskName,
+ boolean writeThrough,
+ boolean intercept
+ ) throws NodeStoppingException {
+ int curPart = -1;
- if (msgLog.isDebugEnabled()) {
- msgLog.debug("Received DHT atomic update request [futId=" + req.futureId() +
- ", writeVer=" + req.writeVersion() + ", node=" + nodeId + ']');
- }
+ int batchStart = 0;
+ int batchSize = 0;
- assert req.partition() >= 0 : req;
+ for (int i = 0; i < req.size(); i++) {
+ KeyCacheObject key = req.key(i);
- GridCacheVersion ver = req.writeVersion();
+ int part = key.partition();
- GridDhtAtomicNearResponse nearRes = null;
+ assert part >= 0 : key;
- if (req.nearNodeId() != null) {
- nearRes = new GridDhtAtomicNearResponse(ctx.cacheId(),
- req.partition(),
- req.nearFutureId(),
+ if (curPart >=0 && curPart != part) {
+ dhtAtomicUpdateRequestUpdateBatch(
+ batchStart,
+ batchSize,
+ nodeId,
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept);
+
+ batchStart = i;
+ batchSize = 1;
+ }
+ else
+ batchSize++;
+
+ curPart = part;
+ }
+
+ if (batchSize > 0) {
+ dhtAtomicUpdateRequestUpdateBatch(
+ batchStart,
+ batchSize,
nodeId,
- req.flags());
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept);
}
+ }
- boolean replicate = ctx.isDrEnabled();
+ @Nullable private List<SearchRowEx<AtomicCacheUpdateClosure>> dhtAtomicUpdateLockBatch(
+ final int batchStart,
+ final int batchSize,
+ final GridDhtAtomicAbstractUpdateRequest req,
+ boolean writeThrough,
+ boolean intercept
+ ) throws IgniteCheckedException {
+ try {
+ List<SearchRowEx<AtomicCacheUpdateClosure>> locked = new ArrayList<>(batchSize);
- boolean intercept = req.forceTransformBackups() && ctx.config().getInterceptor() != null;
+ SearchRowEx<AtomicCacheUpdateClosure> prev = null;
- String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
+ GridDhtLocalPartition part = null;
- ctx.shared().database().checkpointReadLock();
+ while (true) {
+ for (int i = 0; i < batchSize; i++) {
+ int idx = batchStart + i;
- try {
- for (int i = 0; i < req.size(); i++) {
- KeyCacheObject key = req.key(i);
+ KeyCacheObject key = req.key(idx);
- try {
- while (true) {
- GridDhtCacheEntry entry = null;
+ GridDhtCacheEntry entry = entryExx(key);
- try {
- entry = entryExx(key);
+ if (part == null)
+ part = entry.localPartition();
+ else
+ assert part == entry.localPartition();
- CacheObject val = req.value(i);
- CacheObject prevVal = req.previousValue(i);
+ CacheObject val = req.value(idx);
- EntryProcessor<Object, Object, Object> entryProcessor = req.entryProcessor(i);
- Long updateIdx = req.updateCounter(i);
+ EntryProcessor<Object, Object, Object> entryProcessor = req.entryProcessor(i);
+ Long updateIdx = req.updateCounter(idx);
- GridCacheOperation op = entryProcessor != null ? TRANSFORM :
- (val != null) ? UPDATE : DELETE;
+ GridCacheOperation op = entryProcessor != null ? TRANSFORM : (val != null) ? UPDATE : DELETE;
- long ttl = req.ttl(i);
- long expireTime = req.conflictExpireTime(i);
+ long ttl = req.ttl(idx);
+ long expireTime = req.conflictExpireTime(idx);
- GridCacheUpdateAtomicResult updRes = entry.innerUpdate(
- null,
- ver,
- nodeId,
- nodeId,
- op,
- op == TRANSFORM ? entryProcessor : val,
- op == TRANSFORM ? req.invokeArguments() : null,
- /*write-through*/(ctx.store().isLocal() && !ctx.shared().localStorePrimaryOnly())
- && writeThrough() && !req.skipStore(),
- /*read-through*/false,
- /*retval*/false,
- req.keepBinary(),
- /*expiry policy*/null,
- /*event*/true,
- /*metrics*/true,
- /*primary*/false,
- /*check version*/!req.forceTransformBackups(),
- req.topologyVersion(),
- CU.empty0(),
- replicate ? DR_BACKUP : DR_NONE,
- ttl,
- expireTime,
- req.conflictVersion(i),
- false,
- intercept,
- req.subjectId(),
- taskName,
- prevVal,
- updateIdx,
- null,
- req.transformOperation());
+ AtomicCacheUpdateClosure c = new AtomicCacheUpdateClosure(
+ entry,
+ req.topologyVersion(),
+ req.writeVersion(),
+ op,
+ op == TRANSFORM ? entryProcessor : val,
+ op == TRANSFORM ? req.invokeArguments() : null,
+ /*read-through*/false,
+ writeThrough,
+ req.keepBinary(),
+ /*expiry policy*/null,
+ /*primary*/false,
+ /*check version*/!req.forceTransformBackups(),
+ CU.empty0(),
+ ttl,
+ expireTime,
+ req.conflictVersion(i),
+ false,
+ intercept,
+ updateIdx,
+ ctx.disableTriggeringCacheInterceptorOnConflict());
- if (updRes.removeVersion() != null)
- ctx.onDeferredDelete(entry, updRes.removeVersion());
+ SearchRowEx<AtomicCacheUpdateClosure> row = (SearchRowEx<AtomicCacheUpdateClosure>)
+ part.dataStore().createSearchRow(ctx, key, c);
- entry.onUnlock();
+ locked.add(row);
- break; // While.
- }
- catch (GridCacheEntryRemovedException ignored) {
- if (log.isDebugEnabled())
- log.debug("Got removed entry while updating backup value (will retry): " + key);
+ // Expect keys in request are already sorted.
+ assert prev == null || part.dataStore().rowsComparator().compare(row, prev) >= 0;
- entry = null;
- }
- finally {
- if (entry != null)
- entry.touch();
- }
+ prev = row;
+ }
+
+ boolean retry = false;
+
+ for (int i = 0; i < locked.size(); i++) {
+ SearchRowEx<AtomicCacheUpdateClosure> row = locked.get(i);
+
+ GridCacheMapEntry entry = row.data().entry();
+
+ entry.lockEntry();
+
+ if (entry.obsolete()) {
+ // Unlock all locked.
+ for (int j = 0; j <= i; j++)
+ locked.get(j).data().entry().unlockEntry();
+
+ // Clear entries.
+ locked.clear();
+
+ // Retry.
+ retry = true;
+
+ break;
}
}
- catch (NodeStoppingException e){
- U.warn(log, "Failed to update key on backup (local node is stopping): " + key);
- return;
+ if (!retry)
+ return locked;
+ }
+ }
+ catch (GridDhtInvalidPartitionException e) {
+ // Ignore, do not need update backup.
+ return null;
+ }
+ }
+
+ private static AtomicCacheUpdateClosure rowClosure(CacheSearchRow row) {
+ return ((SearchRowEx<AtomicCacheUpdateClosure>)row).data();
+ }
+
+ private void dhtAtomicUpdateRequestUpdateBatch(
+ final int batchStart,
+ final int batchSize,
+ final UUID nodeId,
+ final GridDhtAtomicAbstractUpdateRequest req,
+ final @Nullable GridDhtAtomicNearResponse nearRes,
+ final String taskName,
+ final boolean writeThrough,
+ final boolean intercept
+ ) throws NodeStoppingException {
+ assert batchSize > 0 : batchSize;
+ assert batchStart >= 0 && batchStart < req.size() : batchStart;
+
+ if (batchSize == 1) {
+ dhtAtomicUpdateRequestUpdateSingle(
+ nodeId,
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept,
+ batchStart);
+
+ return;
+ }
+
+ try {
+ ctx.shared().database().ensureFreeSpace(ctx.dataRegion());
+
+ List<SearchRowEx<AtomicCacheUpdateClosure>> locked = dhtAtomicUpdateLockBatch(
+ batchStart,
+ batchSize,
+ req,
+ writeThrough,
+ intercept);
+
+ if (locked == null)
+ return;
+
+ assert !locked.isEmpty();
+
+ IgniteCacheOffheapManager.CacheDataStore dataStore = ((GridDhtCacheEntry)locked.get(0).data().entry()).localPartition().dataStore();
+
+ try {
+ dataStore.invokeAll(ctx, locked, GridDhtAtomicCache::rowClosure);
+
+ for (int i = 0; i < batchSize; i++) {
+ AtomicCacheUpdateClosure c = locked.get(i).data();
+
+ dhtAtomicUpdateRequestUpdateSingleEntry(
+ c.entry(),
+ c,
+ nodeId,
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept,
+ batchStart + i);
}
- catch (GridDhtInvalidPartitionException ignored) {
- // Ignore.
+ }
+ finally {
+ for (int i = 0; i < batchSize; i++)
+ locked.get(i).data().entry().unlockEntry();
+
+ for (int i = 0; i < batchSize; i++) {
+ AtomicCacheUpdateClosure c = locked.get(i).data();
+
+ if (c.updateResult().removeVersion() != null)
+ ctx.onDeferredDelete(c.entry(), c.updateResult().removeVersion());
+
+ GridCacheMapEntry entry = locked.get(i).data().entry();
+
+ entry.onUnlock();
+
+ entry.touch();
}
- catch (IgniteCheckedException|RuntimeException e) {
- if(e instanceof RuntimeException && !X.hasCause(e, IgniteOutOfMemoryException.class))
- throw (RuntimeException)e;
+ }
+ }
+ catch (GridCacheEntryRemovedException e) {
+ assert false;
+ }
+ catch (NodeStoppingException e) {
+ throw e;
+ }
+ catch (IgniteCheckedException | RuntimeException e) {
+ if (e instanceof RuntimeException && !X.hasCause(e, IgniteOutOfMemoryException.class))
+ throw (RuntimeException)e;
+
+ for (int i = 0; i < batchSize; i++) {
+ KeyCacheObject key = req.key(batchStart + i);
+
+ IgniteCheckedException err = new IgniteCheckedException("Failed to update key on backup node: " + key, e);
+
+ if (nearRes != null)
+ nearRes.addFailedKey(key, err);
+ }
+
+ U.error(log, "Failed to update keys on backup node", e);
+ }
+ }
+
+ private GridCacheUpdateAtomicResult dhtAtomicUpdateRequestUpdateSingleEntry(
+ GridCacheMapEntry entry,
+ AtomicCacheUpdateClosure c,
+ UUID nodeId,
+ GridDhtAtomicAbstractUpdateRequest req,
+ @Nullable GridDhtAtomicNearResponse nearRes,
+ String taskName,
+ boolean writeThrough,
+ boolean intercept,
+ int idx
+ ) throws NodeStoppingException, GridCacheEntryRemovedException {
+ CacheObject val = req.value(idx);
+ CacheObject prevVal = req.previousValue(idx);
+
+ EntryProcessor<Object, Object, Object> entryProcessor = req.entryProcessor(idx);
+ Long updateIdx = req.updateCounter(idx);
+
+ GridCacheOperation op = entryProcessor != null ? TRANSFORM : (val != null) ? UPDATE : DELETE;
+
+ long ttl = req.ttl(idx);
+ long expireTime = req.conflictExpireTime(idx);
+
+ try {
+ return entry.innerUpdate(
+ c,
+ req.writeVersion(),
+ nodeId,
+ nodeId,
+ op,
+ op == TRANSFORM ? entryProcessor : val,
+ op == TRANSFORM ? req.invokeArguments() : null,
+ writeThrough,
+ /*read-through*/false,
+ /*retval*/false,
+ req.keepBinary(),
+ /*expiry policy*/null,
+ /*event*/true,
+ /*metrics*/true,
+ /*primary*/false,
+ /*check version*/!req.forceTransformBackups(),
+ req.topologyVersion(),
+ CU.empty0(),
+ ctx.isDrEnabled() ? DR_BACKUP : DR_NONE,
+ ttl,
+ expireTime,
+ req.conflictVersion(idx),
+ false,
+ intercept,
+ req.subjectId(),
+ taskName,
+ prevVal,
+ updateIdx,
+ null,
+ req.transformOperation());
+ }
+ catch (NodeStoppingException e) {
+ throw e;
+ }
+ catch (IgniteCheckedException | RuntimeException e) {
+ if(e instanceof RuntimeException && !X.hasCause(e, IgniteOutOfMemoryException.class))
+ throw (RuntimeException)e;
+
+ KeyCacheObject key = entry.key();
+
+ IgniteCheckedException err = new IgniteCheckedException("Failed to update key on backup node: " + key, e);
+
+ if (nearRes != null)
+ nearRes.addFailedKey(key, err);
+
+ U.error(log, "Failed to update key on backup node: " + key, e);
+
+ return null;
+ }
+ }
+
+ private void dhtAtomicUpdateRequestUpdateSingle(
+ UUID nodeId,
+ GridDhtAtomicAbstractUpdateRequest req,
+ @Nullable GridDhtAtomicNearResponse nearRes,
+ String taskName,
+ boolean writeThrough,
+ boolean intercept,
+ int i) throws NodeStoppingException
+ {
+ KeyCacheObject key = req.key(i);
+
+ try {
+ while (true) {
+ GridDhtCacheEntry entry = null;
- IgniteCheckedException err = new IgniteCheckedException("Failed to update key on backup node: " + key, e);
+ try {
+ entry = entryExx(key);
+
+ GridCacheUpdateAtomicResult updRes = dhtAtomicUpdateRequestUpdateSingleEntry(
+ entry,
+ null,
+ nodeId,
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept,
+ i);
- if (nearRes != null)
- nearRes.addFailedKey(key, err);
+ if (updRes != null && updRes.removeVersion() != null)
+ ctx.onDeferredDelete(entry, updRes.removeVersion());
- U.error(log, "Failed to update key on backup node: " + key, e);
+ entry.onUnlock();
+
+ break; // While.
}
+ catch (GridCacheEntryRemovedException ignored) {
+ if (log.isDebugEnabled())
+ log.debug("Got removed entry while updating backup value (will retry): " + key);
+
+ entry = null;
+ }
+ finally {
+ if (entry != null)
+ entry.touch();
+ }
+ }
+ }
+ catch (GridDhtInvalidPartitionException ignored) {
+ // Ignore.
+ }
+ }
+
+ /**
+ * @param nodeId Sender node ID.
+ * @param req Dht atomic update request.
+ */
+ private void processDhtAtomicUpdateRequest(UUID nodeId, GridDhtAtomicAbstractUpdateRequest req) {
+ assert Thread.currentThread().getName().startsWith("sys-stripe-") : Thread.currentThread().getName();
+
+ if (msgLog.isDebugEnabled()) {
+ msgLog.debug("Received DHT atomic update request [futId=" + req.futureId() +
+ ", writeVer=" + req.writeVersion() + ", node=" + nodeId + ']');
+ }
+
+ assert req.partition() >= 0 : req;
+
+ GridDhtAtomicNearResponse nearRes = null;
+
+ if (req.nearNodeId() != null) {
+ nearRes = new GridDhtAtomicNearResponse(ctx.cacheId(),
+ req.partition(),
+ req.nearFutureId(),
+ nodeId,
+ req.flags());
+ }
+
+ String taskName = ctx.kernalContext().task().resolveTaskName(req.taskNameHash());
+
+ boolean writeThrough = (ctx.store().isLocal() && !ctx.shared().localStorePrimaryOnly())
+ && writeThrough() && !req.skipStore();
+
+ boolean intercept = req.forceTransformBackups() && ctx.config().getInterceptor() != null;
+
+ ctx.shared().database().checkpointReadLock();
+
+ try {
+ if (req.size() > 1) {
+ dhtAtomicUpdateRequestUpdateBatch(
+ nodeId,
+ req,
+ nearRes,
+ taskName,
+ writeThrough,
+ intercept);
}
+ else {
+ for (int i = 0; i < req.size(); i++)
+ dhtAtomicUpdateRequestUpdateSingle(nodeId, req, nearRes, taskName, writeThrough, intercept, i);
+ }
+ }
+ catch (NodeStoppingException e){
+ U.warn(log, "Failed to update key on backup (local node is stopping)");
+
+ return;
}
finally {
ctx.shared().database().checkpointReadUnlock();
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 a15ff0d..57d3fd2 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
@@ -30,6 +30,7 @@ import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -2103,10 +2104,10 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
}
/** {@inheritDoc} */
- @Override public CacheSearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key) throws IgniteCheckedException {
+ @Override public CacheSearchRow createSearchRow(GridCacheContext cctx, KeyCacheObject key, Object data) throws IgniteCheckedException {
CacheDataStore delegate = init0(false);
- return delegate.createSearchRow(cctx, key);
+ return delegate.createSearchRow(cctx, key, data);
}
/** {@inheritDoc} */
@@ -2119,12 +2120,12 @@ public class GridCacheOffheapManager extends IgniteCacheOffheapManagerImpl imple
/** {@inheritDoc} */
@Override public void invokeAll(GridCacheContext cctx,
Collection<? extends CacheSearchRow> rows,
- Map<? extends CacheSearchRow, ? extends OffheapInvokeClosure> map) throws IgniteCheckedException {
+ Function<CacheSearchRow, OffheapInvokeClosure> closures) throws IgniteCheckedException {
assert ctx.database().checkpointLockIsHeldByThread();
CacheDataStore delegate = init0(false);
- delegate.invokeAll(cctx, rows, map);
+ delegate.invokeAll(cctx, rows, closures);
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
index 63a3d2c..d794840 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/BPlusTree.java
@@ -1844,7 +1844,7 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
* @param closures Provider of closures for each search row.
* @throws IgniteCheckedException If failed.
*/
- public void invokeAll(Iterator<? extends L> sortedRows, Object z, Function<L, InvokeClosure<T>> closures)
+ public void invokeAll(Iterator<? extends L> sortedRows, Object z, Function<L, ? extends InvokeClosure<T>> closures)
throws IgniteCheckedException {
doInvoke(new InvokeAll(sortedRows.next(), sortedRows, z, closures));
}
@@ -4035,7 +4035,7 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
Iterator<? extends L> sortedRows;
/** */
- Function<L, InvokeClosure<T>> closures;
+ Function<L, ? extends InvokeClosure<T>> closures;
/** */
ReuseBag reuseBag;
@@ -4046,7 +4046,7 @@ public abstract class BPlusTree<L, T extends L> extends DataStructure implements
* @param x Implementation specific argument.
* @param clo Closure.
*/
- InvokeAll(L firstRow, Iterator<? extends L> sortedRows, Object x, Function<L, InvokeClosure<T>> closures) {
+ InvokeAll(L firstRow, Iterator<? extends L> sortedRows, Object x, Function<L, ? extends InvokeClosure<T>> closures) {
super(firstRow, x, closures.apply(firstRow));
this.sortedRows = sortedRows;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRowEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRowEx.java
new file mode 100644
index 0000000..469e7ba
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/tree/SearchRowEx.java
@@ -0,0 +1,45 @@
+/*
+ * 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.tree;
+
+import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+
+/**
+ *
+ */
+public class SearchRowEx<T> extends SearchRow {
+ /** */
+ private final T data;
+
+ /**
+ * @param cacheId Cache ID.
+ * @param key Key.
+ */
+ public SearchRowEx(int cacheId, KeyCacheObject key, T data) {
+ super(cacheId, key);
+
+ this.data = data;
+ }
+
+ /**
+ * @return Data.
+ */
+ public T data() {
+ return data;
+ }
+}