chihsuan commented on code in PR #10884:
URL: https://github.com/apache/ozone/pull/10884#discussion_r3758629291
##########
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:
Good question. It's a safety guard, not a case we expect to hit. In theory,
it should be unreachable today, since a worker can't finish normally while the
producer is still feeding it, so a missing worker always comes with an
exception we fail on.
The reason it's still there is that this PR changes the producer to poll
instead of blocking forever, so the loop needs a guaranteed exit. If workers
ever went away without an exception, nobody would drain the queue, and the
producer would just keep polling. Does that make sense to you?
--
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]
