chungen0126 commented on code in PR #10884:
URL: https://github.com/apache/ozone/pull/10884#discussion_r3725768999
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/QuotaRepairTask.java:
##########
@@ -584,53 +604,110 @@ private <VALUE> void recalculateUsages(
List<Future<?>> tasks = new ArrayList<>();
AtomicBoolean isRunning = new AtomicBoolean(true);
for (int i = 0; i < TASK_THREAD_CNT; ++i) {
- tasks.add(executor.submit(() -> captureCount(
- prefixUsageMap, q, isRunning, haveValue)));
+ tasks.add(executor.submit(() -> captureCount(q, isRunning, kvConsumer)));
}
int count = 0;
long startTime = Time.monotonicNow();
- try (Table.KeyValueIterator<String, VALUE> keyIter
- = table.iterator(null, haveValue ? KEY_AND_VALUE : KEY_ONLY)) {
+ Exception failure = null;
+ try {
while (keyIter.hasNext()) {
count++;
kvList.add(keyIter.next());
if (kvList.size() == BATCH_SIZE) {
- q.put(kvList);
+ putBatch(q, kvList, tasks);
kvList = new ArrayList<>(BATCH_SIZE);
}
}
- q.put(kvList);
+ putBatch(q, kvList, tasks);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ failure = ex;
+ q.clear();
+ } catch (ExecutionException | RuntimeException ex) {
+ failure = ex;
+ q.clear();
+ } finally {
isRunning.set(false);
- for (Future<?> f : tasks) {
- f.get();
+ }
+ // always await workers so none outlives this scan and touches a closed
table
+ failure = awaitAll(tasks, failure);
+ if (failure != null) {
+ throw new UncheckedExecutionException(failure);
+ }
+ LOG.info("Recalculate {} completed, count {} time {}ms", strType,
+ count, (Time.monotonicNow() - startTime));
+ }
+
+ /**
+ * Awaits every task, retrying an interrupted wait so the interrupted future
is not
+ * abandoned mid-run; an interrupt seen while waiting is restored before
returning.
+ * Returns the passed-in failure or the first failure seen, with later ones
suppressed.
+ */
+ private static Exception awaitAll(List<Future<?>> tasks, Exception failure) {
+ boolean interrupted = false;
+ for (Future<?> f : tasks) {
+ boolean done = false;
+ while (!done) {
+ try {
+ f.get();
+ done = true;
+ } catch (InterruptedException ex) {
+ interrupted = true;
+ if (failure == null) {
+ failure = ex;
+ }
+ } catch (ExecutionException ex) {
+ done = true;
+ if (failure == null) {
+ failure = ex;
+ } else {
+ failure.addSuppressed(ex);
+ }
+ }
}
- LOG.info("Recalculate {} completed, count {} time {}ms", strType,
- count, (Time.monotonicNow() - startTime));
- } catch (IOException ex) {
- throw new UncheckedIOException(ex);
- } catch (InterruptedException ex) {
+ }
+ if (interrupted) {
Thread.currentThread().interrupt();
- } catch (ExecutionException ex) {
- throw new UncheckedExecutionException(ex);
}
+ return failure;
}
-
+
+ /**
+ * Blocks until the batch is queued. Fails fast if a worker has already
exited,
+ * otherwise a failed worker set could leave the producer blocked forever on
a full queue.
+ */
+ private static <VALUE> void putBatch(
+ BlockingQueue<List<Table.KeyValue<String, VALUE>>> q,
+ List<Table.KeyValue<String, VALUE>> kvList,
+ List<Future<?>> tasks) throws InterruptedException, ExecutionException {
+ while (!q.offer(kvList, 100, TimeUnit.MILLISECONDS)) {
+ for (Future<?> f : tasks) {
+ if (f.isDone()) {
+ f.get();
+ throw new IllegalStateException("quota repair scan worker exited
prematurely");
Review Comment:
Could you help clarify when this scenario occurs?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
