Reduced code section size under CP read lock - Fixes
@@ -679,6 +679,9 @@ public class GridDhtPartitionDemander {
GridCacheContext cctx = grp.sharedGroup() ? null : grp.singleCacheContext();
+ ctx.database().checkpointReadLock();
+
+ try {
// Preload.
for (Map.Entry<Integer, CacheEntryInfoCollection> e : supply.infos().entrySet()) {
int p = e.getKey();
@@ -747,6 +750,10 @@ public class GridDhtPartitionDemander {
log.debug("Skipping rebalancing partition (it does not belong on current node): " + p);
}
}
+ }
+ finally {
+ ctx.database().checkpointReadUnlock();
+ }
// Only request partitions based on latest topology version.
for (Integer miss : supply.missed()) {
@@ -803,7 +810,7 @@ public class GridDhtPartitionDemander {
GridCacheEntryInfo entry,
AffinityTopologyVersion topVer
) throws IgniteCheckedException {
- ctx.database().checkpointReadLock();
+ assert ctx.database().checkpointLockIsHeldByThread();
try {
GridCacheEntryEx cached = null;
@@ -816,9 +823,6 @@ public class GridDhtPartitionDemander {
if (log.isDebugEnabled())
log.debug("Rebalancing key [key=" + entry.key() + ", part=" + p + ", node=" + from.id() + ']');
- cctx.shared().database().checkpointReadLock();
-
- try {
if (preloadPred == null || preloadPred.apply(entry)) {
if (cached.initialValue(
entry.value(),
@@ -848,10 +852,6 @@ public class GridDhtPartitionDemander {
else if (log.isDebugEnabled())
log.debug("Rebalance predicate evaluated to false for entry (will ignore): " + entry);
}
- finally {
- cctx.shared().database().checkpointReadUnlock();
- }
- }
catch (GridCacheEntryRemovedException ignored) {
if (log.isDebugEnabled())
log.debug("Entry has been concurrently removed while rebalancing (will ignore) [key=" +
@@ -871,9 +871,6 @@ public class GridDhtPartitionDemander {
throw new IgniteCheckedException("Failed to cache rebalanced entry (will stop rebalancing) [local=" +
ctx.localNode() + ", node=" + from.id() + ", key=" + entry.key() + ", part=" + p + ']', e);
}
- finally {
- ctx.database().checkpointReadUnlock();
- }
return true;
}
