Fault-tolerant execution: FileSystemExchangeSource busy-polls (100% CPU) while the output selector is not final, starving other queries

[caffein123]

Trino version

483

Please describe the bug

Environment

• Fault-tolerant execution, retry-policy=TASK, filesystem exchange manager on S3.
• 2 workers (32 vCPU / ~205 GB heap each), task.concurrency=32, resource-group hardConcurrencyLimit=10.
• Workload: heavy read with multi-way lookup joins (a daily ETL SELECT), run concurrently at N=10–20.

Symptoms

Under high-concurrency FTE, a subset of queries spend minutes with one worker pinned at ~100% CPU while the other worker is idle, making little progress. It is probabilistic — which/how-many queries are hit varies run to run — so the same query at the same concurrency swings its median exec time ~5× (e.g. 166 s vs 827 s for identical config and image).

Per-slot evidence (identical 1.2 GB input, failedTasks=0, all task attempts = 0, spilledDataSize=0, no Rescheduling task, no S3 503/SlowDown, no credential errors):

• "clean" slots: totalCpuTime ≈ 290 s
• "spinning" slots: totalCpuTime 1,800 – 11,000 s (6–38×), while the scan itself is normal (~200 s CPU).

Live inspection during a stall:

• Upstream scan stages FINISHED quickly.
• The consuming (join) stage is RUNNING with hundreds of blocked drivers; ExchangeOperator and LocalExchangeSourceOperator accumulate huge aggregate blockedWall (tens of thousands of seconds) with tiny CPU — the operator alternates between spinning and blocking on exchange input.

Root cause (analysis)

A contract mismatch between read() and isBlocked() in FileSystemExchangeSource lets the driver busy-poll while no data is available:

• read() (v483 ~L199–210) only reads a reader when reader.isBlocked().isDone() && !reader.isFinished(); otherwise returns null.
• isBlocked() (~L135–139) returns NOT_BLOCKED if any reader's isBlocked().isDone() — including finished readers (no !isFinished() check). The whenAnyComplete(...) branch (~L150–153) has the same issue (a finished reader's future is already done → the combined future is done → NOT_BLOCKED).
• closeAndCreateReadersIfNecessary() (~L265–267) returns early when currentSelector == null || !currentSelector.isFinal() without cleaning finished readers. So while the engine has not yet finalized the output selector, finished readers linger in the readers list.

Net effect: finished readers linger → isBlocked() reports NOT_BLOCKED → read() skips them and returns null → the driver spins (each iteration also runs getMemoryUsage() and the reader loop), burning CPU until the output selector becomes final. Under high concurrency the coordinator finalizes selectors more slowly, so spin windows are long and one spinning query starves co-located queries of CPU (the "one worker pinned, others idle" pattern).

Reproduction

Any FTE (retry-policy=TASK) workload with the filesystem/S3 exchange manager running heavy shuffle queries at high concurrency reproduces it; the spin window correlates with the delay before the exchange source's output selector is finalized.

Proposed direction

Make isBlocked() never return NOT_BLOCKED when read() would yield nothing:

• ignore finished readers in the isBlocked() fast loop and in the whenAnyComplete(...) set;
• clean finished readers (and re-arm blockedOnFiles) even on the "selector not final" path of closeAndCreateReadersIfNecessary(), so the driver parks until setOutputSelector / addSourceHandles re-triggers reader creation (the wake-up is already wired there).

We have a candidate patch and can open a PR if the direction looks right. We'd appreciate maintainers confirming the output-selector-finalization wiring so the change cannot miss a wake-up.

Not a duplicate of

• Per-page flushing of task results can saturate worker HTTP threads under high exchange concurrency #30627 / Avoid flushing each task result page #30625 (per-page task-result flush over HTTP — the streaming/DirectExchange path, not the FTE spooling read).
• The recent SpoolingExchangeDataSource memory-context-close race fix (GENERIC_INTERNAL_ERROR: ... already closed) — different symptom; we observe no errors.