Skip to content

Commit d990470

Browse files
committed
fix(qwp): give capacity-starved borrows a final retired-slot probe before the timeout check
borrow() ran the terminal timeout check before reprobeRetiredSlots(), so a zero-timeout (try-once) borrow threw without its one probe, and a borrower whose awaitNanos budget expired mid-wait timed out on capacity that a deferred engine cleanup had already returned (the delegate-side flock release never signals slotReleased). Hoist the probe above the timeout check so both paths recover the capacity instead of failing. Also pre-size retiredSlots to maxSize: every entry keeps a distinct in-range slot index reserved, so add() can never grow the backing array -- a retire (leakedSlots++ then add, under lock) can no longer fail on allocation and strand a counted-but-untracked slot that reprobeRetiredSlots() could never recover. Tests: - testZeroTimeoutBorrowProbesRetiredSlotBeforeThrowing (red pre-fix) - testParkedBorrowerGetsFinalProbeAfterBudgetExpiry (red pre-fix) - testPoolRetiresAndRecoversSlotThroughRealManagerWorkerWedge: full-stack retire/recover cycle with no forged flags -- real wedged manager worker, real timed-out close handoff, real flock release, and a re-borrow on the recovered index proving the slot dir is genuinely reusable
1 parent 7b9924c commit d990470

2 files changed

Lines changed: 329 additions & 8 deletions

File tree

core/src/main/java/io/questdb/client/impl/SenderPool.java

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -187,8 +187,12 @@ public final class SenderPool implements AutoCloseable {
187187
// the pool's capacity instead of ratcheting it down until process exit.
188188
// Out-of-range startup recoverers are NEVER added: they carry no
189189
// leakedSlots tick and their index has no slotInUse entry to free.
190-
// Guarded by lock.
191-
private final ArrayList<SenderSlot> retiredSlots = new ArrayList<>();
190+
// Pre-sized to maxSize (every entry keeps a distinct in-range slot index
191+
// reserved, so size can never exceed maxSize): add() never grows the
192+
// backing array, so a retire (leakedSlots++ then add, under lock) cannot
193+
// fail on allocation and strand a counted-but-untracked slot that
194+
// reprobeRetiredSlots() could never recover. Guarded by lock.
195+
private final ArrayList<SenderSlot> retiredSlots;
192196
// SF slots currently held by the in-range startup-recovery pass
193197
// (recoverOneSlotStep): each is reserved under `lock` for the
194198
// duration of its drain and counted in the borrow() cap check so a
@@ -306,6 +310,7 @@ public SenderPool(
306310
this.maxLifetimeMillis = maxLifetimeMillis;
307311
this.all = new ArrayList<>(maxSize);
308312
this.available = new ArrayDeque<>(maxSize);
313+
this.retiredSlots = new ArrayList<>(maxSize);
309314
this.slotReleased = lock.newCondition();
310315
// Probe the config once, up front: this validates it eagerly (so a
311316
// bad config fails at construction even when minSize == 0) and tells
@@ -793,16 +798,24 @@ public PooledSender borrow() {
793798
created.bumpGeneration();
794799
return new PooledSender(created, created.generation());
795800
}
801+
// Capacity-starved: re-probe retired slots BEFORE the terminal
802+
// timeout check — a deferred engine cleanup may have released a
803+
// flock since the retire, and the freed index can admit a
804+
// creation right now. The release itself never signals
805+
// slotReleased (it happens in the delegate on a worker/I/O-thread
806+
// exit path, volatile writes only), so this poll is the only way
807+
// a borrower learns of it. Ordering matters twice over: a
808+
// zero-timeout (try-once) borrow must get its one probe before
809+
// throwing, and a borrower whose awaitNanos budget just expired
810+
// must get a final probe on its wake-up pass instead of timing
811+
// out on capacity that has already come back.
812+
if (reprobeRetiredSlots()) {
813+
continue;
814+
}
796815
if (remainingNanos <= 0) {
797816
throw new LineSenderException(
798817
"timed out waiting for a Sender from the pool after " + acquireTimeoutMillis + "ms");
799818
}
800-
// Capacity-starved: before parking, re-probe retired slots — a
801-
// deferred engine cleanup may have released a flock since the
802-
// retire, and the freed index can admit a creation right now.
803-
if (reprobeRetiredSlots()) {
804-
continue;
805-
}
806819
try {
807820
remainingNanos = slotReleased.awaitNanos(remainingNanos);
808821
} catch (InterruptedException e) {

0 commit comments

Comments
 (0)