This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new ba84c74727d fix: Race between run and cancellation in
FrameProcessorExecutor. (#19006)
ba84c74727d is described below
commit ba84c74727d0f9301409aad8de3849dfc9a33779
Author: Gian Merlino <[email protected]>
AuthorDate: Wed Sep 2 20:14:34 2026 -0700
fix: Race between run and cancellation in FrameProcessorExecutor. (#19006)
Previously, the ExecutorRunnable owned the processor only while running
runProcessorNow(). This created races with cancellation. For example,
if a processor was canceled and cleaned up while the ExecutorRunnable
was calling "isFinished" or "readabilityFuture" on an input channel, it
could lead to calling those operations on a closed channel.
This patch fixes it by having the ExecutorRunnable own the processor
a bit longer, to cover the time that it may need to call methods on
input and output channels. We now also also take care to not call
methods on channels in the debug method logProcessorStatusString.
This patch also makes BlockingQueueFrameChannel more strict about
closing, which helps ensure the above fix is working:
1) Writable channel now rejects writes when closed, rather than when
the reader has finished reading.
2) Writable channel now rejects calls to all methods other than
isClosed() when closed. Behavior changed in: writabilityFuture(),
fail(), and close().
3) Readable channel now rejects calls to all methods when closed.
Behavior changed in: isFinished(), canRead(), read(),
readabilityFuture(), and close().
---
.../output/ChannelStageOutputReaderTest.java | 22 +-
.../frame/channel/BlockingQueueFrameChannel.java | 87 +++++--
.../druid/frame/channel/WritableFrameChannel.java | 15 +-
.../frame/processor/FrameProcessorExecutor.java | 276 ++++++++++++---------
.../druid/frame/processor/ReturnOrAwait.java | 10 +-
.../processor/FrameProcessorExecutorTest.java | 9 +-
.../processor/test/InfiniteFrameProcessor.java | 8 +-
7 files changed, 260 insertions(+), 167 deletions(-)
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/shuffle/output/ChannelStageOutputReaderTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/shuffle/output/ChannelStageOutputReaderTest.java
index d556f185535..a0b0e51f935 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/shuffle/output/ChannelStageOutputReaderTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/shuffle/output/ChannelStageOutputReaderTest.java
@@ -24,6 +24,7 @@ import com.google.common.io.ByteStreams;
import com.google.common.math.IntMath;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.common.guava.FutureUtils;
+import org.apache.druid.error.DruidException;
import org.apache.druid.error.ThrowableMatcher;
import org.apache.druid.frame.Frame;
import org.apache.druid.frame.FrameType;
@@ -167,12 +168,12 @@ public class ChannelStageOutputReaderTest extends
InitializedNullHandlingTest
channel.writable().write(frame);
// See that we can't write another frame.
- final IllegalStateException e = Assertions.assertThrows(
- IllegalStateException.class,
+ final DruidException e = Assertions.assertThrows(
+ DruidException.class,
() -> channel.writable().write(frame)
);
- ThrowableMatcher.of(IllegalStateException.class)
+ ThrowableMatcher.of(DruidException.class)
.expectMessage(message -> message.startsWith("Channel has
no capacity"))
.assertThat(e);
@@ -192,12 +193,12 @@ public class ChannelStageOutputReaderTest extends
InitializedNullHandlingTest
channel.writable().write(frame);
// See that we can't write a fourth frame.
- final IllegalStateException e2 = Assertions.assertThrows(
- IllegalStateException.class,
+ final DruidException e2 = Assertions.assertThrows(
+ DruidException.class,
() -> channel.writable().write(frame)
);
- ThrowableMatcher.of(IllegalStateException.class)
+ ThrowableMatcher.of(DruidException.class)
.expectMessage(message -> message.startsWith("Channel has
no capacity"))
.assertThat(e2);
@@ -277,7 +278,7 @@ public class ChannelStageOutputReaderTest extends
InitializedNullHandlingTest
private static final int EXPECTED_NUM_ROWS = 1209;
private final BlockingQueueFrameChannel channel = new
BlockingQueueFrameChannel(MAX_FRAMES);
- private final ChannelStageOutputReader reader = new
ChannelStageOutputReader(channel.readable(), FrameTestUtil.WT_CONTEXT_LEGACY);
+ private ChannelStageOutputReader reader = new
ChannelStageOutputReader(channel.readable(), FrameTestUtil.WT_CONTEXT_LEGACY);
@RegisterExtension
public final TemporaryFolderExtension temporaryFolder =
TemporaryFolderExtension.testCaseScoped();
@@ -307,7 +308,9 @@ public class ChannelStageOutputReaderTest extends
InitializedNullHandlingTest
@AfterEach
public void tearDown()
{
- reader.close();
+ if (reader != null) {
+ reader.close();
+ }
}
@Test
@@ -334,6 +337,9 @@ public class ChannelStageOutputReaderTest extends
InitializedNullHandlingTest
IllegalStateException.class,
reader::readLocally
);
+
+ // Prevent close() in tearDown, which would fail.
+ reader = null;
}
@Test
diff --git
a/processing/src/main/java/org/apache/druid/frame/channel/BlockingQueueFrameChannel.java
b/processing/src/main/java/org/apache/druid/frame/channel/BlockingQueueFrameChannel.java
index e434b62c209..9c7abb164d1 100644
---
a/processing/src/main/java/org/apache/druid/frame/channel/BlockingQueueFrameChannel.java
+++
b/processing/src/main/java/org/apache/druid/frame/channel/BlockingQueueFrameChannel.java
@@ -24,9 +24,9 @@ import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import it.unimi.dsi.fastutil.objects.ObjectIntPair;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.Either;
import org.apache.druid.java.util.common.IAE;
-import org.apache.druid.java.util.common.ISE;
import org.apache.druid.query.rowsandcols.RowsAndColumns;
import javax.annotation.Nullable;
@@ -52,14 +52,31 @@ public class BlockingQueueFrameChannel
private final Writable writable;
private final Readable readable;
+ /**
+ * Queue of items from the writer. Ends with {@link #END_MARKER} if the
writable channel is closed.
+ * Only ever updated by the writer.
+ */
@GuardedBy("lock")
private final ArrayDeque<Optional<Either<Throwable,
ObjectIntPair<RowsAndColumns>>>> queue;
+ /**
+ * Whether {@link Readable#close()} has been called.
+ */
@GuardedBy("lock")
- private SettableFuture<?> readyForWritingFuture = null;
+ private boolean readerClosed;
+ /**
+ * Future that is set to null by {@link #notifyWriter()} when the reader has
read from {@link #queue}
+ * or been closed.
+ */
@GuardedBy("lock")
- private SettableFuture<?> readyForReadingFuture = null;
+ private SettableFuture<?> readyForWritingFuture;
+
+ /**
+ * Future that is set to null by {@link #notifyReader()} when the writer has
written to {@link #queue}.
+ */
+ @GuardedBy("lock")
+ private SettableFuture<?> readyForReadingFuture;
/**
* Create a channel with a particular buffer size (expressed in number of
frames).
@@ -100,13 +117,6 @@ public class BlockingQueueFrameChannel
return new BlockingQueueFrameChannel(1);
}
- private boolean isFinished()
- {
- synchronized (lock) {
- return END_MARKER.equals(queue.peek());
- }
- }
-
@GuardedBy("lock")
private void notifyWriter()
{
@@ -133,16 +143,14 @@ public class BlockingQueueFrameChannel
public void write(RowsAndColumns rac, int partitionNumber)
{
synchronized (lock) {
- if (isFinished()) {
- throw new ISE("Channel cannot accept new frames");
+ if (isClosed()) {
+ throw DruidException.defensive("Channel cannot accept new frames");
} else if (queue.size() >= maxQueuedFrames) {
// Caller should have checked if this channel was ready for writing.
- throw new ISE("Channel has no capacity");
- } else {
- if (!queue.offer(Optional.of(Either.value(ObjectIntPair.of(rac,
partitionNumber))))) {
- // If this happens, it's a bug in this class's capacity-counting.
- throw new ISE("Channel had capacity, but could not add frame");
- }
+ throw DruidException.defensive("Channel has no capacity");
+ } else if (!queue.offer(Optional.of(Either.value(ObjectIntPair.of(rac,
partitionNumber))))) {
+ // If this happens, it's a bug in this class's capacity-counting.
+ throw DruidException.defensive("Channel had capacity, but could not
add frame");
}
notifyReader();
@@ -153,7 +161,9 @@ public class BlockingQueueFrameChannel
public ListenableFuture<?> writabilityFuture()
{
synchronized (lock) {
- if (queue.size() < maxQueuedFrames) {
+ if (isClosed()) {
+ throw DruidException.defensive("Closed, cannot call
writabilityFuture()");
+ } else if (queue.size() < maxQueuedFrames) {
return Futures.immediateFuture(null);
} else if (readyForWritingFuture != null) {
return readyForWritingFuture;
@@ -167,11 +177,15 @@ public class BlockingQueueFrameChannel
public void fail(@Nullable Throwable cause)
{
synchronized (lock) {
+ if (isClosed()) {
+ throw DruidException.defensive("Closed, cannot call fail()");
+ }
+
queue.clear();
- if (!queue.offer(Optional.of(Either.error(cause != null ? cause : new
ISE("Failed"))))) {
+ if (!queue.offer(Optional.of(Either.error(cause != null ? cause : new
RuntimeException("Failed"))))) {
// If this happens, it's a bug, potentially due to incorrectly using
this class with multiple writers.
- throw new ISE("Could not write error to channel");
+ throw DruidException.defensive("Could not write error to channel");
}
notifyReader();
@@ -183,12 +197,12 @@ public class BlockingQueueFrameChannel
{
synchronized (lock) {
if (isClosed()) {
- throw new ISE("Already closed");
+ throw DruidException.defensive("Closed, cannot call close() again");
}
if (!queue.offer(END_MARKER)) {
// If this happens, it's a bug, potentially due to incorrectly using
this class with multiple writers.
- throw new ISE("Channel had capacity, but could not add end marker");
+ throw DruidException.defensive("Channel had capacity, but could not
add end marker");
}
notifyReader();
@@ -210,13 +224,23 @@ public class BlockingQueueFrameChannel
@Override
public boolean isFinished()
{
- return BlockingQueueFrameChannel.this.isFinished();
+ synchronized (lock) {
+ if (readerClosed) {
+ throw DruidException.defensive("Closed, cannot call isFinished()");
+ }
+
+ return END_MARKER.equals(queue.peek());
+ }
}
@Override
public boolean canRead()
{
synchronized (lock) {
+ if (readerClosed) {
+ throw DruidException.defensive("Closed, cannot call canRead()");
+ }
+
return !queue.isEmpty() && !isFinished();
}
}
@@ -227,6 +251,10 @@ public class BlockingQueueFrameChannel
final Optional<Either<Throwable, ObjectIntPair<RowsAndColumns>>> next;
synchronized (lock) {
+ if (readerClosed) {
+ throw DruidException.defensive("Closed, cannot call read()");
+ }
+
if (isFinished()) {
throw new NoSuchElementException();
}
@@ -247,6 +275,10 @@ public class BlockingQueueFrameChannel
public ListenableFuture<?> readabilityFuture()
{
synchronized (lock) {
+ if (readerClosed) {
+ throw DruidException.defensive("Closed, cannot call
readabilityFuture()");
+ }
+
if (!queue.isEmpty()) {
return Futures.immediateFuture(null);
} else if (readyForReadingFuture != null) {
@@ -261,7 +293,12 @@ public class BlockingQueueFrameChannel
public void close()
{
synchronized (lock) {
- queue.clear();
+ if (readerClosed) {
+ // close() should not be called twice.
+ throw DruidException.defensive("Closed, cannot call close() again");
+ }
+
+ readerClosed = true;
notifyWriter();
}
}
diff --git
a/processing/src/main/java/org/apache/druid/frame/channel/WritableFrameChannel.java
b/processing/src/main/java/org/apache/druid/frame/channel/WritableFrameChannel.java
index 170ce4b6844..155b6cfcd3d 100644
---
a/processing/src/main/java/org/apache/druid/frame/channel/WritableFrameChannel.java
+++
b/processing/src/main/java/org/apache/druid/frame/channel/WritableFrameChannel.java
@@ -92,14 +92,19 @@ public interface WritableFrameChannel extends Closeable
*/
void fail(@Nullable Throwable cause) throws IOException;
+ /**
+ * Returns a future that resolves when {@link #write} is able to receive a
new batch of data without blocking or
+ * throwing an exception. The future never resolves to an exception.
+ */
+ ListenableFuture<?> writabilityFuture();
+
/**
* Finish writing to this channel.
*
* When this method is called without {@link #fail(Throwable)} having
previously been called, the writer is
* understood to have completed successfully.
*
- * After calling this method, no additional calls to {@link #write}, {@link
#fail(Throwable)}, or this method
- * are permitted.
+ * After calling this method, no additional calls to are permitted to any
methods other than {@link #isClosed()}.
*/
@Override
void close() throws IOException;
@@ -108,10 +113,4 @@ public interface WritableFrameChannel extends Closeable
* Whether {@link #close()} has been called on this channel.
*/
boolean isClosed();
-
- /**
- * Returns a future that resolves when {@link #write} is able to receive a
new batch of data without blocking or
- * throwing an exception. The future never resolves to an exception.
- */
- ListenableFuture<?> writabilityFuture();
}
diff --git
a/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
b/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
index f3d378fe576..530232d4e07 100644
---
a/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
+++
b/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
@@ -32,25 +32,23 @@ import com.google.common.util.concurrent.SettableFuture;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.ints.IntSet;
+import org.apache.druid.error.DruidException;
import org.apache.druid.frame.channel.ReadableFrameChannel;
import org.apache.druid.frame.channel.WritableFrameChannel;
import org.apache.druid.frame.processor.manager.ProcessorManager;
import org.apache.druid.java.util.common.Either;
-import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.logger.Logger;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.Executor;
@@ -127,57 +125,82 @@ public class FrameProcessorExecutor
public void run()
{
try {
- final List<ListenableFuture<?>> allWritabilityFutures =
gatherWritabilityFutures();
- final List<ListenableFuture<?>> writabilityFuturesToWaitFor =
- allWritabilityFutures.stream().filter(f ->
!f.isDone()).collect(Collectors.toList());
-
- logProcessorStatusString(processor, finished, allWritabilityFutures);
-
- if (!writabilityFuturesToWaitFor.isEmpty()) {
-
runProcessorAfterFutureResolves(Futures.allAsList(writabilityFuturesToWaitFor),
false);
+ if (!registerRunningProcessor()) {
+ // Processor was canceled. Just exit; cleanup would have been
handled elsewhere.
return;
}
- final Optional<ReturnOrAwait<T>> maybeResult = runProcessorNow();
+ // Set nonnull if the processor should run again.
+ Runnable nextRun = null;
+ Throwable error = null;
- if (!maybeResult.isPresent()) {
- // Processor exited abnormally. Just exit; cleanup would have been
handled elsewhere.
- return;
- }
+ try {
+ final List<ListenableFuture<?>> writabilityFutures =
gatherWritabilityFutures();
+ final List<ListenableFuture<?>> writabilityFuturesToWaitFor =
+ writabilityFutures.stream().filter(f ->
!f.isDone()).collect(Collectors.toList());
+
+ if (!writabilityFuturesToWaitFor.isEmpty()) {
+ logProcessorStatusString(processor, finished.isDone(), null,
null, writabilityFutures);
+ nextRun = () ->
runProcessorAfterFutureResolves(Futures.allAsList(writabilityFuturesToWaitFor),
false);
+ return;
+ }
- final ReturnOrAwait<T> result = maybeResult.get();
- logProcessorStatusString(processor, finished, null);
+ final ReturnOrAwait<T> result = runProcessorNow();
- if (result.isReturn()) {
- succeed(result.value());
- } else if (result.hasAwaitableFutures()) {
-
runProcessorAfterFutureResolves(Futures.allAsList(result.awaitableFutures()),
true);
- } else {
- assert result.hasAwaitableChannels();
+ if (result.isReturn()) {
+ logProcessorStatusString(processor, finished.isDone(), result,
null, null);
+ succeed(result.value());
+ } else if (result.hasAwaitableFutures()) {
+ logProcessorStatusString(processor, finished.isDone(), result,
result.awaitableFutures(), null);
+ final ListenableFuture<List<Object>> combinedFuture =
Futures.allAsList(result.awaitableFutures());
+ nextRun = () -> runProcessorAfterFutureResolves(combinedFuture,
true);
+ } else {
+ assert result.hasAwaitableChannels();
- // Don't retain a reference to this set: it may be mutated the
next time the processor runs.
- final IntSet await = result.awaitableChannels();
+ // Don't retain a reference to this set: it may be mutated the
next time the processor runs.
+ final IntSet await = result.awaitableChannels();
- if (await.isEmpty()) {
- exec.execute(ExecutorRunnable.this);
- } else if (result.isAwaitAllChannels() || await.size() == 1) {
- final List<ListenableFuture<?>> readabilityFutures = new
ArrayList<>();
+ if (await.isEmpty()) {
+ nextRun = () -> exec.execute(ExecutorRunnable.this);
+ } else if (result.isAwaitAllChannels() || await.size() == 1) {
+ final List<ListenableFuture<?>> readabilityFutures = new
ArrayList<>();
- for (final int channelNumber : await) {
- final ReadableFrameChannel channel =
inputChannels.get(channelNumber);
- if (!channel.isFinished() && !channel.canRead()) {
- readabilityFutures.add(channel.readabilityFuture());
+ for (final int channelNumber : await) {
+ final ReadableFrameChannel channel =
inputChannels.get(channelNumber);
+ if (!channel.isFinished() && !channel.canRead()) {
+ readabilityFutures.add(channel.readabilityFuture());
+ }
}
- }
- if (readabilityFutures.isEmpty()) {
- exec.execute(ExecutorRunnable.this);
+ logProcessorStatusString(processor, finished.isDone(), result,
readabilityFutures, null);
+
+ if (readabilityFutures.isEmpty()) {
+ nextRun = () -> exec.execute(ExecutorRunnable.this);
+ } else {
+ final ListenableFuture<List<Object>> combinedFuture =
Futures.allAsList(readabilityFutures);
+ nextRun = () ->
runProcessorAfterFutureResolves(combinedFuture, false);
+ }
} else {
-
runProcessorAfterFutureResolves(Futures.allAsList(readabilityFutures), false);
+ // Await any.
+ final ListenableFuture<?> combinedFuture =
awaitAnyWidget.awaitAny(await);
+ logProcessorStatusString(processor, finished.isDone(), result,
List.of(combinedFuture), null);
+ nextRun = () ->
runProcessorAfterFutureResolves(combinedFuture, false);
}
- } else {
- // Await any.
- runProcessorAfterFutureResolves(awaitAnyWidget.awaitAny(await),
false);
+ }
+ }
+ catch (Throwable e) {
+ error = e;
+ }
+ finally {
+ final boolean canceled = deregisterRunningProcessor();
+
+ if (canceled) {
+ // Processor was canceled while running. Suppress the error and
don't schedule the next run;
+ // the cancel thread handles cleanup.
+ } else if (error != null) {
+ fail(error);
+ } else if (nextRun != null) {
+ nextRun.run();
}
}
}
@@ -198,11 +221,10 @@ public class FrameProcessorExecutor
}
/**
- * Executes {@link FrameProcessor#runIncrementally} on the
currently-readable inputs, while respecting
- * cancellation. Returns an empty Optional if the processor exited
abnormally (canceled or failed). Returns a
- * present Optional if the processor ran successfully. Throws an
exception if the processor does.
+ * Executes {@link FrameProcessor#runIncrementally} on the
currently-readable inputs.
+ * Throws an exception if the processor does.
*/
- private Optional<ReturnOrAwait<T>> runProcessorNow()
+ private ReturnOrAwait<T> runProcessorNow()
{
final IntSet readableInputs = new IntOpenHashSet(inputChannels.size());
@@ -213,21 +235,7 @@ public class FrameProcessorExecutor
}
}
- if (cancellationId != null) {
- // After this synchronized block, our thread may be interrupted by
cancellations, because "cancel"
- // checks "runningProcessors".
- synchronized (lock) {
- if (cancelableProcessors.containsEntry(cancellationId, processor))
{
- runningProcessors.put(processor, Thread.currentThread());
- } else {
- // Processor has been canceled. We don't need to handle cleanup,
because someone else did it.
- return Optional.empty();
- }
- }
- }
-
final String threadName = Thread.currentThread().getName();
- boolean canceled = false;
Either<Throwable, ReturnOrAwait<T>> retVal;
try {
@@ -247,33 +255,11 @@ public class FrameProcessorExecutor
retVal = Either.error(e);
}
finally {
- if (cancellationId != null) {
- // After this synchronized block, our thread will no longer be
interrupted by cancellations,
- // because "cancel" checks "runningProcessors".
- synchronized (lock) {
- if (Thread.interrupted()) {
- // ignore: interrupt was meant for the processor, but came
after the processor already exited.
- }
-
- runningProcessors.remove(processor);
- lock.notifyAll();
-
- if (!cancelableProcessors.containsEntry(cancellationId,
processor)) {
- // Processor has been canceled by one of the "cancel" methods.
They will handle cleanup.
- canceled = true;
- }
- }
-
- // Restore original thread name.
- Thread.currentThread().setName(threadName);
- }
+ // Restore original thread name.
+ Thread.currentThread().setName(threadName);
}
- if (canceled) {
- return Optional.empty();
- } else {
- return Optional.of(retVal.valueOrThrow());
- }
+ return retVal.valueOrThrow();
}
/**
@@ -281,7 +267,7 @@ public class FrameProcessorExecutor
*
* @param future the future
* @param failOnCancel whether the processor should be {@link
#fail(Throwable)} if the future is itself canceled.
- * This is true for futures provided by {@link
ReturnOrAwait#awaitAllFutures(Collection)},
+ * This is true for futures provided by {@link
ReturnOrAwait#awaitAllFutures(List)},
* because the processor has declared it wants to
wait for them; if they are canceled
* the processor must fail. It is false for other
futures, which the processor was not
* directly waiting for.
@@ -384,13 +370,69 @@ public class FrameProcessorExecutor
processor.cleanup();
}
}
+
+ /**
+ * Registers the current thread as running the processor. Returns false
if the processor has been canceled,
+ * in which case the caller should not run it.
+ */
+ private boolean registerRunningProcessor()
+ {
+ if (cancellationId != null) {
+ // After this synchronized block, our thread may be interrupted by
cancellations, because "cancel"
+ // checks "runningProcessors".
+ synchronized (lock) {
+ if (cancelableProcessors.containsEntry(cancellationId, processor))
{
+ final Thread priorThread =
runningProcessors.putIfAbsent(processor, Thread.currentThread());
+ if (priorThread != null) {
+ throw DruidException.defensive(
+ "Already running processor[%s] on thread[%s], cannot also
run on thread[%s]",
+ processor,
+ priorThread.getName(),
+ Thread.currentThread().getName()
+ );
+ }
+ } else {
+ // Processor has been canceled. We don't need to handle cleanup,
because someone else did it.
+ return false;
+ }
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * Deregisters the current thread from running the processor, clearing
any pending interrupt.
+ *
+ * @return true if the processor was canceled while it was running
+ */
+ private boolean deregisterRunningProcessor()
+ {
+ if (cancellationId != null) {
+ // After this synchronized block, our thread will no longer be
interrupted by cancellations,
+ // because "cancel" checks "runningProcessors".
+ synchronized (lock) {
+ if (Thread.interrupted()) {
+ // ignore: interrupt was meant for the processor, but came after
the processor already exited.
+ }
+
+ runningProcessors.remove(processor);
+ lock.notifyAll();
+
+ return !cancelableProcessors.containsEntry(cancellationId,
processor);
+ }
+ }
+
+ return false;
+ }
}
final ExecutorRunnable runnable = new ExecutorRunnable();
finished.addListener(
() -> {
- logProcessorStatusString(processor, finished, null);
+ // finished will be done here, so this log call gets us the "done=y"
log.
+ logProcessorStatusString(processor, finished.isDone(), null, null,
null);
// If the future was canceled, and the processor is cancelable, then
cancel the processor too.
if (finished.isCancelled() && cancellationId != null) {
@@ -408,7 +450,7 @@ public class FrameProcessorExecutor
Execs.directExecutor()
);
- logProcessorStatusString(processor, finished, null);
+ logProcessorStatusString(processor, finished.isDone(), null, null, null);
registerCancelableProcessor(processor, cancellationId);
exec.execute(runnable);
return finished;
@@ -633,9 +675,22 @@ public class FrameProcessorExecutor
}
}
+ /**
+ * Logs a debug-level status string for the given processor, including input
channel readability,
+ * output channel writability, and completion state.
+ *
+ * @param processor the processor
+ * @param finished whether the processor's "finished" future has resolved
+ * @param returnOrAwait if the processor has just finished {@link
FrameProcessor#runIncrementally}, what it returned
+ * @param readabilityFutures if the processor has just finished {@link
FrameProcessor#runIncrementally}, futures for
+ * inputs it is waiting for
+ * @param writabilityFutures futures for output availability
+ */
private static <T> void logProcessorStatusString(
final FrameProcessor<T> processor,
- final ListenableFuture<?> finishedFuture,
+ final boolean finished,
+ @Nullable final ReturnOrAwait<?> returnOrAwait,
+ @Nullable final List<ListenableFuture<?>> readabilityFutures,
@Nullable final List<ListenableFuture<?>> writabilityFutures
)
{
@@ -643,37 +698,36 @@ public class FrameProcessorExecutor
final StringBuilder sb = new StringBuilder()
.append("Processor [")
.append(processor)
- .append("]; in=[");
+ .append("]");
- for (ReadableFrameChannel channel : processor.inputChannels()) {
- if (channel.canRead()) {
- sb.append("R"); // R for readable
- } else if (channel.isFinished()) {
- sb.append("D"); // D for done
- } else {
- sb.append("~"); // ~ for waiting
- }
- }
+ if (returnOrAwait != null) {
+ sb.append("; ").append(returnOrAwait);
- sb.append("]");
+ if (readabilityFutures != null) {
+ // Information about inputs. Note, we can't call methods on the
processor's input channels, because
+ // they might have been closed.
+ sb.append("; in=[");
+ for (final ListenableFuture<?> future : readabilityFutures) {
+ sb.append(future.isDone() ? 'R' : 'B'); // R = ready; B = blocked.
+ }
+ sb.append("]");
+ }
- if (writabilityFutures != null) {
- sb.append("; out=[");
+ if (writabilityFutures != null) {
+ // Information about outputs. Note, we can't call methods on the
processor's output channels, because
+ // they might have been closed.
+ sb.append("; out=[");
- for (final ListenableFuture<?> future : writabilityFutures) {
- if (future.isDone()) {
- sb.append("W"); // W for writable
- } else {
- sb.append("~"); // ~ for waiting
+ for (final ListenableFuture<?> future : writabilityFutures) {
+ sb.append(future.isDone() ? 'R' : 'B'); // R = ready; B = blocked.
}
- }
- sb.append("]");
+ sb.append("]");
+ }
}
- sb.append("; cancel=").append(finishedFuture.isCancelled() ? "y" : "n");
- sb.append("; done=").append(finishedFuture.isDone() ? "y" : "n");
- log.debug(StringUtils.encodeForFormat(sb.toString()));
+ sb.append("; done=").append(finished ? "y" : "n");
+ log.debug("%s", sb);
}
}
diff --git
a/processing/src/main/java/org/apache/druid/frame/processor/ReturnOrAwait.java
b/processing/src/main/java/org/apache/druid/frame/processor/ReturnOrAwait.java
index c154b425851..9ae6c96d7dd 100644
---
a/processing/src/main/java/org/apache/druid/frame/processor/ReturnOrAwait.java
+++
b/processing/src/main/java/org/apache/druid/frame/processor/ReturnOrAwait.java
@@ -26,7 +26,7 @@ import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
import javax.annotation.Nullable;
-import java.util.Collection;
+import java.util.List;
/**
* Instances of this class are returned by {@link
FrameProcessor#runIncrementally}, and are used by
@@ -52,12 +52,12 @@ public class ReturnOrAwait<T>
private final boolean awaitAllChannels;
@Nullable
- private final Collection<ListenableFuture<?>> awaitFutures;
+ private final List<ListenableFuture<?>> awaitFutures;
private ReturnOrAwait(
@Nullable T retVal,
@Nullable IntSet awaitChannels,
- @Nullable Collection<ListenableFuture<?>> awaitFutures,
+ @Nullable List<ListenableFuture<?>> awaitFutures,
final boolean awaitAllChannels
)
{
@@ -106,7 +106,7 @@ public class ReturnOrAwait<T>
/**
* Wait for all of the provided futures.
*/
- public static <T> ReturnOrAwait<T> awaitAllFutures(final
Collection<ListenableFuture<?>> futures)
+ public static <T> ReturnOrAwait<T> awaitAllFutures(final
List<ListenableFuture<?>> futures)
{
return new ReturnOrAwait<>(null, null, futures, true);
}
@@ -161,7 +161,7 @@ public class ReturnOrAwait<T>
}
- public Collection<ListenableFuture<?>> awaitableFutures()
+ public List<ListenableFuture<?>> awaitableFutures()
{
if (!hasAwaitableFutures()) {
throw new ISE("No futures set");
diff --git
a/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
b/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
index d1426a7be40..b728c2b7469 100644
---
a/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
+++
b/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
@@ -211,7 +211,7 @@ public class FrameProcessorExecutorTest
}
@Test
- public void test_registerCancelableFuture() throws InterruptedException
+ public void test_registerCancelableFuture()
{
final SettableFuture<Object> future = SettableFuture.create();
final String cancellationId = "xyzzy";
@@ -372,15 +372,12 @@ public class FrameProcessorExecutorTest
// If we see an unresolved future here, it's a bug in exec.cancel.
Assertions.assertTrue(future.isDone());
Assertions.assertTrue(future.isCancelled());
-
- final Exception e = Assertions.assertThrows(Exception.class,
future::get);
- Assertions.assertInstanceOf(CancellationException.class, e);
}
}
// In both cases, check for cleanup.
for (final InfiniteFrameProcessor generator : generators) {
- Assertions.assertTrue(generator.didCleanup());
+ Assertions.assertEquals(1, generator.getCleanupCount(), "exactly one
cleanup call");
}
Assertions.assertTrue(chomper.didCleanup());
@@ -388,7 +385,7 @@ public class FrameProcessorExecutorTest
}
@Test
- public void test_cancel_nonexistentCancellationId() throws
InterruptedException
+ public void test_cancel_nonexistentCancellationId()
{
// Just making sure no error is thrown when we refer to a nonexistent
cancellationId.
exec.cancel("nonexistent");
diff --git
a/processing/src/test/java/org/apache/druid/frame/processor/test/InfiniteFrameProcessor.java
b/processing/src/test/java/org/apache/druid/frame/processor/test/InfiniteFrameProcessor.java
index fbbf45f81d8..9443106c456 100644
---
a/processing/src/test/java/org/apache/druid/frame/processor/test/InfiniteFrameProcessor.java
+++
b/processing/src/test/java/org/apache/druid/frame/processor/test/InfiniteFrameProcessor.java
@@ -45,7 +45,7 @@ public class InfiniteFrameProcessor implements
FrameProcessor<Long>
private final Frame frame;
private final WritableFrameChannel outChannel;
private final AtomicBoolean stop = new AtomicBoolean(false);
- private final AtomicBoolean didCleanup = new AtomicBoolean(false);
+ private final AtomicLong cleanupCount = new AtomicLong();
private final AtomicLong numFrames = new AtomicLong();
public InfiniteFrameProcessor(
@@ -86,7 +86,7 @@ public class InfiniteFrameProcessor implements
FrameProcessor<Long>
public void cleanup() throws IOException
{
FrameProcessors.closeAll(inputChannels(), outputChannels());
- didCleanup.set(true);
+ cleanupCount.incrementAndGet();
}
public long getNumFrames()
@@ -99,8 +99,8 @@ public class InfiniteFrameProcessor implements
FrameProcessor<Long>
stop.set(true);
}
- public boolean didCleanup()
+ public long getCleanupCount()
{
- return didCleanup.get();
+ return cleanupCount.get();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
