clintropolis commented on code in PR #20197:
URL: https://github.com/apache/druid/pull/20197#discussion_r4128203931
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java:
##########
@@ -1993,32 +1929,111 @@ private File getEffectiveInfoDir()
return infoDir;
}
- private Supplier<ListenableFuture<AcquireSegmentResult>>
makeOnDemandLoadSupplier(
- final CompleteSegmentCacheEntry entry,
- final StorageLocation location
+ /**
+ * Shared scaffolding for on-demand acquires: runs {@code taskBody} on
{@link #virtualStorageLoadingThreadPool},
+ * delivering its result to the returned {@link AcquireSegmentAction}.
Closing the action before delivery cancels a
+ * still-queued task (it never runs); a task already running is deliberately
NOT interrupted (see the
+ * {@code cancel(false)} rationale in the canceler below) and runs to
completion, loses the delivery race, and
+ * closes its own orphaned result.
+ * <p>
+ * Takes ownership of {@code preplacedHold} in all cases:
+ * <ul>
+ * <li>task delivers: the hold is folded into the delivered segment's
close (or closed by the fold on a miss)</li>
+ * <li>task delivers but the action was closed first: the orphaned result
is closed here, releasing the folded
+ * holds along with the segment reference</li>
+ * <li>task throws: the task closes all accumulated holds and the failure
is delivered (or silently absorbed if
+ * the action was closed first)</li>
+ * <li>action closed before the task ever runs (canceled while queued):
the canceler claims and closes the
+ * hold</li>
+ * <li>submitting the task fails synchronously: the hold is closed here
and the failure propagates to the
+ * {@link #acquireSegment} caller</li>
+ * </ul>
+ * Task failures are classified here, where the segment is known: a {@link
DruidException} thrown by the task is
+ * delivered as-is, anything else is delivered as an operator-facing runtime
failure naming the segment. Consumers
+ * of {@link AcquireSegmentAction#release()} therefore always see a properly
categorized {@link DruidException}
+ * rather than {@code AsyncResource.get()}'s generic wrapper around a
checked exception.
+ */
+ private AcquireSegmentAction submitAcquireTask(
+ final SegmentId segmentId,
+ final Closeable preplacedHold,
+ final AcquireTaskBody taskBody
)
{
- return Suppliers.memoize(
- () -> {
- final long startTime = System.nanoTime();
- return virtualStorageLoadingThreadPool.getExecutorService().submit(
- () -> {
- // waitTime is only the executor scheduling delay; when using
virtual threads for the pool, load-slot
- // contention is folded into loadTime now, since mount
acquires the load permit around the deep-storage
- // read.
- final long execStartTime = System.nanoTime();
- final long waitTime = execStartTime - startTime;
- entry.mount(location);
- return new AcquireSegmentResult(
- entry.referenceProvider,
- entry.dataSegment.getSize(),
- waitTime,
- System.nanoTime() - execStartTime
- );
- }
- );
+ final AcquireSegmentAction action = new AcquireSegmentAction();
+ final AtomicBoolean holdClaimed = new AtomicBoolean(false);
+ final long submitNanos = System.nanoTime();
+ final ListenableFuture<?> taskFuture;
+ try {
+ taskFuture =
virtualStorageLoadingThreadPool.getExecutorService().submit(() -> {
+ if (!holdClaimed.compareAndSet(false, true)) {
+ // the canceler won the claim and closed the pre-placed hold,
nothing to do
+ return null;
}
- );
+ final Closer taskHolds = Closer.create();
+ taskHolds.register(preplacedHold);
+ try {
+ // waitNanos is only the executor scheduling delay until the task
body starts; it no longer reflects
+ // load-slot contention when using virtual threads. Load permits are
acquired inside the deep-storage
+ // reads now, so that wait is folded into loadTime instead, and the
query-time bundle/column fetches
+ // (separate permit-bounded tasks) are not reflected here at all. A
meaningful load-wait metric would
+ // have to time the permit acquire at the read sites and aggregate
it across those fetches.
+ final AcquireSegmentResult result = taskBody.load(taskHolds,
System.nanoTime() - submitNanos);
+ if (!action.set(result)) {
+ // the action was closed while the task was completing; close the
orphaned result
+ result.close();
+ }
+ }
+ catch (Throwable t) {
+ CloseableUtils.closeAndSuppressExceptions(taskHolds,
t::addSuppressed);
+ final DruidException failure =
+ t instanceof DruidException druidException
+ ? druidException
+ : DruidException.forPersona(DruidException.Persona.OPERATOR)
+
.ofCategory(DruidException.Category.RUNTIME_FAILURE)
+ .build(t, "Failed to load segment[%s] on
demand", segmentId);
+ // silently absorbed if the action was closed first
+ action.setException(failure);
+ }
+ return null;
+ });
+ }
+ catch (Throwable t) {
+ throw CloseableUtils.closeAndWrapInCatch(t, preplacedHold);
+ }
+ action.setCanceler(() -> {
+ // cancel(false), NOT cancel(true): a still-queued task is prevented
from running (the canceler then wins the
+ // claim CAS below and releases the pre-placed hold); a task already
running is left to finish rather than
+ // interrupted, because interrupting a running mount would also fail any
concurrent query awaiting the same
+ // mount (mounts are deduped). A running task that finishes after close
loses the set() race and closes its
+ // own orphaned result, releasing the holds then.
+ taskFuture.cancel(false);
Review Comment:
yea, i'm planning to look into this in a follow-up, this isn't a behavior
change was mostly just capturing the current reasoning for now of why we aren't
interrupting.
--
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]
