xtern commented on a change in pull request #8846:
URL: https://github.com/apache/ignite/pull/8846#discussion_r601228948
##########
File path:
modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/CacheGroupPageScanner.java
##########
@@ -442,74 +485,88 @@ public long remainingPagesCount() {
return remainingPagesCntr.get();
}
- /** {@inheritDoc} */
- @Override public void run() {
- try {
- for (int partId : parts) {
- long state =
ctx.encryption().getEncryptionState(grp.groupId(), partId);
-
- if (state == 0)
- continue;
-
- scanPartition(partId,
ReencryptStateUtils.pageIndex(state), ReencryptStateUtils.pageCount(state));
+ /**
+ * @param partId Partition ID.
+ */
+ private void schedulePartitionScan(int partId) {
+ singleExecSvc.submit(() -> scanPartition(partId));
+ }
- if (isDone())
- return;
- }
+ /**
+ * Check the completeness of the cache group scan.
+ */
+ private synchronized void checkComplete() {
+ if (!isDone() && parts.isEmpty() && !cpWaitGrps.contains(this))
+ cpWaitGrps.add(this);
+ }
- boolean added = cpWaitGrps.add(this);
+ /**
+ * @param partId Partition ID.
+ * @return {@code True} if partition has been evicted.
+ */
+ private boolean evicted(int partId) {
+ if (partId == PageIdAllocator.INDEX_PARTITION)
+ return false;
- assert added;
- }
- catch (Throwable t) {
- if (X.hasCause(t, NodeStoppingException.class))
- onCancelled();
- else
- onDone(t);
- }
+ return !parts.contains(partId) ||
+ grp.topology().localPartition(partId).state() ==
GridDhtPartitionState.EVICTED;
}
/**
* @param partId Partition ID.
- * @param off Start page offset.
- * @param cnt Count of pages to scan.
*/
- private void scanPartition(int partId, int off, int cnt) throws
IgniteCheckedException {
+ private void scanPartition(int partId) {
+ long state = ctx.encryption().getEncryptionState(grp.groupId(),
partId);
+
+ int off = ReencryptStateUtils.pageIndex(state);
+ int cnt = ReencryptStateUtils.pageCount(state);
+
if (log.isDebugEnabled()) {
log.debug("Partition reencryption is started [grpId=" +
grp.groupId() +
", p=" + partId + ", remain=" + (cnt - off) + ", total=" +
cnt + "]");
}
- while (off < cnt) {
- int pagesCnt = Math.min(batchSize, cnt - off);
+ try {
+ while (off < cnt) {
+ int pagesCnt = Math.min(batchSize, cnt - off);
- limiter.acquire(pagesCnt);
+ limiter.acquire(pagesCnt);
- synchronized (this) {
- if (isDone() || !parts.contains(partId))
- break;
+ synchronized (this) {
+ if (isDone() || evicted(partId))
Review comment:
Done
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
