Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.IntFunction;

Expand Down Expand Up @@ -185,8 +186,13 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
inCreation.countDown();
awaitOrFail(releaseCreation, "test never released query creation");
};
// A 1s creation-wait budget, not 100ms: the interrupt storm below must land at least
// twice inside this window for the deadline-restart property to be exercised at all,
// and a freshly started, yielding interrupter thread is not guaranteed two scheduler
// quanta within 100ms on a saturated CI agent (observed on hosted 3-core mac agents,
// where the post-join count assert failed with the product deadline honored exactly).
QuestDBImpl db = newQuestDB(
SENDER_CFG, 0, 0, 100, slotIndex -> fakeSender(null, null, null), connectHook);
SENDER_CFG, 0, 0, 1000, slotIndex -> fakeSender(null, null, null), connectHook);
QueryClientPool pool = db.getQueryPoolForTesting();
AtomicReference<Throwable> borrowOutcome = new AtomicReference<>();
AtomicBoolean closeReturnedInterrupted = new AtomicBoolean();
Expand Down Expand Up @@ -221,6 +227,8 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringQueryCreation() thr
awaitCreationWaiter(pool,
"facade close did not wait while query construction was internally owned");
interrupter.start();
awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting,
"query close left its creation wait before the interrupt storm landed twice");
closer.join(TimeUnit.SECONDS.toMillis(5));
Assert.assertFalse(
"repeated interrupts restarted the query creation-wait deadline",
Expand Down Expand Up @@ -267,7 +275,9 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
};
String senderConfig = "ws::addr=localhost:1;sf_dir="
+ System.getProperty("java.io.tmpdir") + "/qdb-interrupted-pool-" + System.nanoTime() + ";";
QuestDBImpl db = newQuestDB(senderConfig, 0, 0, 100, senderFactory, client -> {
// 1s creation-wait budget for the same reason as the query-interrupt test above: the
// interrupt storm must land at least twice inside the window even on a saturated agent.
QuestDBImpl db = newQuestDB(senderConfig, 0, 0, 1000, senderFactory, client -> {
});
SenderPool pool = db.getSenderPoolForTesting();
AtomicReference<Throwable> borrowOutcome = new AtomicReference<>();
Expand Down Expand Up @@ -304,6 +314,8 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th
awaitCreationWaiter(pool,
"facade close did not wait while sender construction was internally owned");
interrupter.start();
awaitRepeatedInterrupts(interruptCount, pool::hasCreationWaiterForTesting,
"sender close left its creation wait before the interrupt storm landed twice");
closer.join(TimeUnit.SECONDS.toMillis(5));
Assert.assertFalse(
"repeated interrupts restarted the sender creation-wait deadline",
Expand Down Expand Up @@ -517,6 +529,34 @@ private static void awaitCreationWaiter(SenderPool pool, String message) {
Assert.fail(message);
}

/**
* Holds the test until the interrupt storm has landed at least twice while the facade close is
* still inside its bounded creation wait. The deadline-restart property is only exercised by
* interrupts that arrive during that wait, and the scheduler owes the interrupter thread
* nothing: with a post-join count assert alone, the run races the close budget against thread
* scheduling and can fail with the product invariant intact. Failing here instead separates
* "interrupter starved before the budget expired" from a genuine deadline bug.
*/
private static void awaitRepeatedInterrupts(
AtomicInteger interruptCount,
BooleanSupplier closerStillWaiting,
String message
) {
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
while (System.nanoTime() < deadline) {
// Count first: two interrupts observed while polling means the storm landed no matter
// how quickly the wait ends afterwards, so a budget expiry seen next is not a failure.
if (interruptCount.get() > 1) {
return;
}
if (!closerStillWaiting.getAsBoolean()) {
Assert.fail(message + "; interrupts landed: " + interruptCount.get());
}
Thread.yield();
}
Assert.fail(message + "; interrupts landed: " + interruptCount.get());
}

private static void awaitOrFail(CountDownLatch latch, String message) {
try {
if (!latch.await(10, TimeUnit.SECONDS)) {
Expand Down
Loading