capistrant commented on code in PR #20805:
URL: https://github.com/apache/druid/pull/20805#discussion_r4482475345
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ReadableInputQueue.java:
##########
@@ -331,47 +355,218 @@ private ListenableFuture<NumberedReadableInput>
nextChannelInput()
@Nullable
private ListenableFuture<NumberedReadableInput> nextSegmentInput()
{
- // Pick a loadahead future, preferring ones that are already loaded.
- ListenableFuture<NumberedReadableInput> selectedLoadaheadFuture = null;
+ final List<ListenableFuture<NumberedReadableInput>> skippedFutures = new
ArrayList<>();
+ ListenableFuture<NumberedReadableInput> unskippedFuture = null;
+ LoadableSegment unskippedSegment = null;
+
synchronized (this) {
- for (ListenableFuture<NumberedReadableInput> f : pendingNextInputs) {
- if (selectedLoadaheadFuture == null || f.isDone()) {
- selectedLoadaheadFuture = f;
- if (f.isDone()) {
- break;
- }
+ acquireCachedInNextGroup();
+ while (unskippedFuture == null && !pendingNextInputs.isEmpty()) {
+ final ListenableFuture<NumberedReadableInput> selectedFuture =
selectPendingInput();
+ final PendingInput selectedInput =
pendingNextInputs.remove(selectedFuture);
+ if (!selectedInput.cached()) {
+ addLoadaheadFuture();
+ }
+ // The selected segment may have been pruned since its load was
started, check if it can be skipped once more
+ // before handing it out.
+ if (canSkipSegment(selectedInput.segment())) {
+ skippedFutures.add(selectedFuture);
+ // Skipping may have moved the next input to a new interval group.
+ acquireCachedInNextGroup();
+ } else {
+ unskippedFuture = selectedFuture;
+ unskippedSegment = selectedInput.segment();
}
}
+ }
- if (selectedLoadaheadFuture != null) {
- pendingNextInputs.remove(selectedLoadaheadFuture);
- if (pendingNextInputs.size() < loadahead) {
- addLoadaheadFuture(); // Replace the one we just took out.
- }
- return selectedLoadaheadFuture;
+ releaseSkippedInputs(skippedFutures);
+ if (unskippedFuture != null) {
+ registerCountOnSuccess(unskippedFuture, unskippedSegment);
+ return unskippedFuture;
+ }
+
+ // Acquires on demand when loadahead is 0. With loadahead, no unpruned
segments remain here.
+ final NumberedItem<LoadableSegment> nextItem = pollUnprunedSegment();
+ if (nextItem == null) {
+ return null;
+ }
+ final ListenableFuture<NumberedReadableInput> future =
acquireSegment(nextItem);
+ registerCountOnSuccess(future, nextItem.item());
+ return future;
+ }
+
+ /**
+ * Registers a callback that counts {@code segment} as read once {@code
future} succeeds, which is when the input is
+ * handed off for processing.
+ */
+ private static void registerCountOnSuccess(
+ final ListenableFuture<NumberedReadableInput> future,
+ final LoadableSegment segment
+ )
+ {
+ Futures.addCallback(
+ future,
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(final NumberedReadableInput result)
+ {
+ segment.countHandedOut();
+ }
+
+ @Override
+ public void onFailure(final Throwable t)
+ {
+ // Failed inputs are never processed, so there is nothing to count.
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
+ }
+
+ /**
+ * Picks the pending input to hand out next. Never null; only called when
{@link #pendingNextInputs} is nonempty.
+ */
+ @GuardedBy("this")
+ private ListenableFuture<NumberedReadableInput> selectPendingInput()
+ {
+ final Iterator<Map.Entry<ListenableFuture<NumberedReadableInput>,
PendingInput>> iterator =
+ pendingNextInputs.entrySet().iterator();
+ final Map.Entry<ListenableFuture<NumberedReadableInput>, PendingInput>
head = iterator.next();
+ if (head.getKey().isDone()) {
+ // Eagerly hand out a ready head input.
+ return head.getKey();
+ }
+
+ // head is not ready, seek for the next ready input who compares equal to
head. for ordered inputs that means
+ // matching intervals, for unordered inputs any ready input is fine.
+ final Interval headInterval = head.getValue().interval();
+ while (iterator.hasNext()) {
+ final Map.Entry<ListenableFuture<NumberedReadableInput>, PendingInput>
entry = iterator.next();
+ if (entry.getKey().isDone() && intervalOrder.compare(headInterval,
entry.getValue().interval()) == 0) {
+ return entry.getKey();
+ }
+ }
+
+ // Nothing level with the head is loaded, so hand out the head and let the
caller wait for it.
+ return head.getKey();
+ }
+
+ /**
+ * Acquires the locally-cached segments of the next interval group to be
handed out and adds them to
+ * {@link #pendingNextInputs}, once per group.
+ * <p>
+ * A group is the leading run of {@link #loadableSegments} whose intervals
compare equal under {@link #intervalOrder};
+ * with {@link Order#NONE}, that is every segment. Segments that {@link
#inputPruner} rules out are dropped.
+ */
+ @GuardedBy("this")
+ private void acquireCachedInNextGroup()
+ {
+ final Interval nextInterval;
+ if (!pendingNextInputs.isEmpty()) {
+ nextInterval = pendingNextInputs.values().iterator().next().interval();
+ } else if (!loadableSegments.isEmpty()) {
+ nextInterval = loadableSegments.peek().item().descriptor().getInterval();
+ } else {
+ return;
+ }
+
+ if (probedGroupInterval != null &&
intervalOrder.compare(probedGroupInterval, nextInterval) == 0) {
+ return;
+ }
+ probedGroupInterval = nextInterval;
+
+ final List<NumberedItem<LoadableSegment>> notCached = new ArrayList<>();
+ NumberedItem<LoadableSegment> nextItem;
+ while ((nextItem = loadableSegments.peek()) != null
+ && intervalOrder.compare(nextInterval,
nextItem.item().descriptor().getInterval()) == 0) {
+ loadableSegments.poll();
+ if (canSkipSegment(nextItem.item())) {
+ continue;
+ }
+ final ListenableFuture<NumberedReadableInput> cachedFuture =
acquireIfCached(nextItem);
+ if (cachedFuture != null) {
+ pendingNextInputs.put(cachedFuture, new PendingInput(nextItem.item(),
true));
+ } else {
+ notCached.add(nextItem);
}
}
- return loadNextSegment();
+ // Put non-cached segments back at the front of loadableSegments.
+ for (int i = notCached.size() - 1; i >= 0; i--) {
+ loadableSegments.addFirst(notCached.get(i));
+ }
}
/**
- * Load the next segment from {@link #loadableSegments} and return a future
to its reference. Returns null
- * if {@link #loadableSegments} is empty, or if {@link #canSkipRemaining} is
true.
+ * Removes and returns the next item from {@link #loadableSegments} that
{@link #inputPruner} does not rule out, or
+ * null if none remain.
*/
@Nullable
- private ListenableFuture<NumberedReadableInput> loadNextSegment()
+ private NumberedItem<LoadableSegment> pollUnprunedSegment()
{
- if (canSkipRemaining.getAsBoolean()) {
- return null;
+ synchronized (this) {
+ NumberedItem<LoadableSegment> nextItem;
+ do {
+ nextItem = loadableSegments.poll();
+ } while (nextItem != null && canSkipSegment(nextItem.item()));
+ return nextItem;
}
+ }
+
+ /**
+ * Returns whether processing the given segment can be safely skipped.
+ */
+ private boolean canSkipSegment(final LoadableSegment segment)
+ {
+ return inputPruner.canSkip(segment.descriptor().getInterval());
+ }
+ /**
+ * Acquires the segment in {@code nextItem}: immediately if it is cached,
otherwise by starting to load it.
+ */
+ private ListenableFuture<NumberedReadableInput> acquireSegment(final
NumberedItem<LoadableSegment> nextItem)
+ {
synchronized (this) {
- final NumberedItem<LoadableSegment> nextItem = loadableSegments.poll();
- if (nextItem == null) {
- return null;
+ final ListenableFuture<NumberedReadableInput> cachedFuture =
acquireIfCached(nextItem);
+ if (cachedFuture != null) {
+ return cachedFuture;
}
+ }
+
+ return loadSegment(nextItem);
Review Comment:
hmm. I took a look and tried to consolidate where the monitor is taken. my
first commit in this work split it up and I think that is what exposed these
windows where a close could orphan segments
--
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]
