petrov-mg commented on code in PR #13686:
URL: https://github.com/apache/ignite/pull/13686#discussion_r4348037097
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -488,7 +492,11 @@ private Future<Map<PartitionKey, PartitionHashRecord>>
calculatePartitionHashAsy
final GridDhtLocalPartition part,
final BooleanSupplier cancelled
) {
+ checkCancelledOrStopped(cancelled, gctx, true);
Review Comment:
I think it is better to just drop this call. Exceptions from
calcPartitionHashAsync are not handled at all.
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -277,15 +276,20 @@ public
VerifyBackupPartitionsJob(CacheIdleVerifyCommandArg arg) {
return res;
}
- /**
- * Cancels futures from given index.
- *
- * @param start Index to start from.
- * @param partHashCalcFuts Partitions hash calculation futures to
cancel.
- */
- private void cancelFuts(int start, List<Future<Map<PartitionKey,
PartitionHashRecord>>> partHashCalcFuts) {
- for (int i = start; i < partHashCalcFuts.size(); i++)
- partHashCalcFuts.get(i).cancel(false);
+ /** */
+ private void cancelFutures(int startIdx, List<Future<Map<PartitionKey,
PartitionHashRecord>>> futs) {
Review Comment:
cancelFutures -> `cancelJobs` or even `abort`?
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -549,5 +568,25 @@ private Future<Map<PartitionKey, PartitionHashRecord>>
calculatePartitionHashAsy
return res;
});
}
+
+ /**
+ * Checks whether the cancellation is required. If is required and the
exception is requested, throws it.
Review Comment:
Fix typos in the Javadoc, as well as any missing or extra spaces in this
method.
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -277,15 +276,20 @@ public
VerifyBackupPartitionsJob(CacheIdleVerifyCommandArg arg) {
return res;
}
- /**
- * Cancels futures from given index.
- *
- * @param start Index to start from.
- * @param partHashCalcFuts Partitions hash calculation futures to
cancel.
- */
- private void cancelFuts(int start, List<Future<Map<PartitionKey,
PartitionHashRecord>>> partHashCalcFuts) {
- for (int i = start; i < partHashCalcFuts.size(); i++)
- partHashCalcFuts.get(i).cancel(false);
+ /** */
+ private void cancelFutures(int startIdx, List<Future<Map<PartitionKey,
PartitionHashRecord>>> futs) {
+ // Propagate the cancellation to the async partition hash
calculations.
+ cancel();
+
+ // Wait for the rest of the futures to stop.
+ for (int i = startIdx; i < futs.size(); i++) {
+ try {
+ futs.get(i).get();
Review Comment:
We need to handle interrupted exception. It might be worth adding an
explicit timeout.
##########
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java:
##########
@@ -410,6 +410,18 @@ public void storeEntries(GridDhtLocalPartition part,
Iterator<GridCacheEntryInfo
*/
public void preloadPartition(int pardId) throws IgniteCheckedException;
+ /**
+ * Enters "busy" state.
+ *
+ * @return {@code true} if entered to busy state.
+ */
+ public boolean enterBusy();
Review Comment:
We should add proper comments to these methods. They should describe what
the methods are intended to guard.
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -227,11 +226,8 @@ public VerifyBackupPartitionsJob(CacheIdleVerifyCommandArg
arg) {
long lastProgressLogTs = U.currentTimeMillis();
for (int i = 0; i < partHashCalcFuts.size(); ) {
- if (isCancelled()) {
- cancelFuts(i, partHashCalcFuts);
-
+ if (isCancelled())
Review Comment:
Should we restore cancelFutures call here?
##########
modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java:
##########
@@ -499,28 +507,38 @@ private Future<Map<PartitionKey, PartitionHashRecord>>
calculatePartitionHashAsy
PartitionUpdateCounter updateCntrBefore = updCntr == null
? null : updCntr.copy();
if (arg.checkCrc() && gctx.persistenceEnabled()) {
- FilePageStoreManager pageStoreMgr =
-
(FilePageStoreManager)ignite.context().cache().context().pageStore();
+ FilePageStoreManager pageStoreMgr =
(FilePageStoreManager)ignite.context().cache().context().pageStore();
checkPartitionsPageCrcSum(
() ->
(FilePageStore)pageStoreMgr.getStore(gctx.groupId(), part.id()),
part.id(),
FLAG_DATA,
- cancelled
+ () -> checkCancelledOrStopped(cancelled, gctx,
false)
);
}
PartitionKey key = new PartitionKey(gctx.groupId(),
part.id(), gctx.cacheOrGroupName());
- PartitionHashRecord hash = calculatePartitionHash(key,
- updateCntrBefore == null ? 0 :
updateCntrBefore.comparableState(),
-
ignite.context().discovery().localNode().consistentId(),
- part.state(),
- part.primary(gctx.topology().readyTopologyVersion()),
- part.dataStore().fullSize(),
- gctx.offheap().partitionIterator(part.id()),
- cancelled
- );
+ PartitionHashRecord hash;
+
+ // The cache group can be destroyed concurrently.
+ if (!gctx.offheap().enterBusy())
+ throw cancellationException();
+
+ try {
+ hash = calculatePartitionHash(
+ key,
+ updateCntrBefore == null ? 0 :
updateCntrBefore.comparableState(),
+
ignite.context().discovery().localNode().consistentId(),
+ part.state(),
+
part.primary(gctx.topology().readyTopologyVersion()),
+ part.dataStore().fullSize(),
+ gctx.offheap().partitionIterator(part.id()),
+ () -> checkCancelledOrStopped(cancelled, gctx,
false)
+ );
+ } finally {
Review Comment:
Line break is required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
