From 1d77cee2399e8753c5b8d78b94f650ce14fe6c53 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Sat, 3 Oct 2026 10:30:52 -0700 Subject: [PATCH] perf: replace the per-operation orTimeout with a per-client timeout sweeper Every put, get, delete, deleteRange and list, and every read, list and range-scan stream attempt, called CompletableFuture.orTimeout, which schedules and cancels a task on the single JVM-wide delayer, taking its lock twice per operation. TimeoutSweeper replaces it: a future is appended to a list picked by the calling thread, the adding thread drops the completed futures when its list doubles, and a periodic sweep on the client executor fails the expired ones. Timeouts are approximate: never early, and at most two sweep intervals (a tenth of the timeout, up to 100 ms) late. Fixes #394 Signed-off-by: Matteo Merli --- .../client/util/TimeoutSweeperBenchmark.java | 82 ++++++++ .../io/oxia/client/AsyncOxiaClientImpl.java | 27 +-- .../io/oxia/client/grpc/GrpcRpcProvider.java | 23 +-- .../io/oxia/client/util/TimeoutSweeper.java | 193 ++++++++++++++++++ .../oxia/client/util/TimeoutSweeperTest.java | 164 +++++++++++++++ 5 files changed, 462 insertions(+), 27 deletions(-) create mode 100644 benchmarks/src/jmh/java/io/oxia/client/util/TimeoutSweeperBenchmark.java create mode 100644 client/src/main/java/io/oxia/client/util/TimeoutSweeper.java create mode 100644 client/src/test/java/io/oxia/client/util/TimeoutSweeperTest.java diff --git a/benchmarks/src/jmh/java/io/oxia/client/util/TimeoutSweeperBenchmark.java b/benchmarks/src/jmh/java/io/oxia/client/util/TimeoutSweeperBenchmark.java new file mode 100644 index 00000000..a618981a --- /dev/null +++ b/benchmarks/src/jmh/java/io/oxia/client/util/TimeoutSweeperBenchmark.java @@ -0,0 +1,82 @@ +/* + * Copyright © 2026 The Oxia Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.oxia.client.util; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Compares {@link TimeoutSweeper} against the {@link CompletableFuture#orTimeout} it replaced, for + * an operation that completes well before its timeout. Run with {@code ./gradlew :benchmarks:jmh}; + * the threads contend on the JVM-wide delayer with {@code orTimeout}. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@State(Scope.Benchmark) +@Fork(1) +@Threads(8) +@Warmup(iterations = 3, time = 1) +@Measurement(iterations = 5, time = 2) +public class TimeoutSweeperBenchmark { + + private static final Duration TIMEOUT = Duration.ofSeconds(30); + private static final Object RESULT = new Object(); + + private ScheduledExecutorService executor; + private TimeoutSweeper sweeper; + + @Setup + public void setup() { + executor = Executors.newSingleThreadScheduledExecutor(); + sweeper = new TimeoutSweeper(executor, TIMEOUT); + } + + @TearDown + public void tearDown() { + sweeper.close(); + executor.shutdownNow(); + } + + @Benchmark + public CompletableFuture orTimeout() { + var future = + new CompletableFuture().orTimeout(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + future.complete(RESULT); + return future; + } + + @Benchmark + public CompletableFuture timeoutSweeper() { + var future = sweeper.add(new CompletableFuture<>()); + future.complete(RESULT); + return future; + } +} diff --git a/client/src/main/java/io/oxia/client/AsyncOxiaClientImpl.java b/client/src/main/java/io/oxia/client/AsyncOxiaClientImpl.java index b530dc5e..5732ad7e 100644 --- a/client/src/main/java/io/oxia/client/AsyncOxiaClientImpl.java +++ b/client/src/main/java/io/oxia/client/AsyncOxiaClientImpl.java @@ -50,6 +50,7 @@ import io.oxia.client.session.SessionManager; import io.oxia.client.shard.ShardManager; import io.oxia.client.util.PendingBytesLimiter; +import io.oxia.client.util.TimeoutSweeper; import io.oxia.proto.KeyComparisonType; import io.oxia.proto.ListRequest; import io.oxia.proto.ListResponse; @@ -67,7 +68,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -184,7 +184,7 @@ class AsyncOxiaClientImpl implements AsyncOxiaClient { private final @NonNull BatchManager readBatchManager; private final @NonNull BatchManager writeBatchManager; private final @NonNull SessionManager sessionManager; - private final long requestTimeoutMs; + private final @NonNull TimeoutSweeper requestTimeouts; private final @NonNull PendingBytesLimiter pendingBytesLimiter; private volatile boolean closed; @@ -242,7 +242,7 @@ class AsyncOxiaClientImpl implements AsyncOxiaClient { this.sessionManager = sessionManager; this.scheduledExecutor = scheduledExecutor; this.ownsResources = ownsResources; - this.requestTimeoutMs = requestTimeout.toMillis(); + this.requestTimeouts = new TimeoutSweeper(scheduledExecutor, requestTimeout); counterPutBytes = instrumentProvider.newCounter( @@ -375,8 +375,8 @@ class AsyncOxiaClientImpl implements AsyncOxiaClient { callback = CompletableFuture.failedFuture(e); } final long pendingBytes = acquiredBytes; - return callback - .orTimeout(requestTimeoutMs, TimeUnit.MILLISECONDS) + return requestTimeouts + .add(callback) .whenComplete( (putResult, throwable) -> { if (pendingBytes > 0) { @@ -488,8 +488,8 @@ private CompletableFuture internalPut( callback.completeExceptionally(e); } final long pendingBytes = acquiredBytes; - return callback - .orTimeout(requestTimeoutMs, TimeUnit.MILLISECONDS) + return requestTimeouts + .add(callback) .whenComplete( (putResult, throwable) -> { if (pendingBytes > 0) { @@ -552,8 +552,8 @@ private CompletableFuture internalPut( callback = CompletableFuture.failedFuture(e); } final long pendingBytes = acquiredBytes; - return callback - .orTimeout(requestTimeoutMs, TimeUnit.MILLISECONDS) + return requestTimeouts + .add(callback) .whenComplete( (putResult, throwable) -> { if (pendingBytes > 0) { @@ -593,8 +593,8 @@ private CompletableFuture internalPut( callback.completeExceptionally(e); } final long pendingBytes = acquiredBytes; - return callback - .orTimeout(requestTimeoutMs, TimeUnit.MILLISECONDS) + return requestTimeouts + .add(callback) .whenComplete( (getResult, throwable) -> { if (pendingBytes > 0) { @@ -697,8 +697,8 @@ private void internalGetMultiShards( } catch (Exception e) { callback = CompletableFuture.failedFuture(e); } - return callback - .orTimeout(requestTimeoutMs, TimeUnit.MILLISECONDS) + return requestTimeouts + .add(callback) .whenComplete( (listResult, throwable) -> { gaugePendingListRequests.decrement(); @@ -965,6 +965,7 @@ public void close() throws Exception { // In shared mode the RpcProvider does not own the connection pool, so this only closes the // per-client write streams; the shared connections stay open for other clients. rpcProvider.close(); + requestTimeouts.close(); if (ownsResources) { scheduledExecutor.shutdownNow(); } diff --git a/client/src/main/java/io/oxia/client/grpc/GrpcRpcProvider.java b/client/src/main/java/io/oxia/client/grpc/GrpcRpcProvider.java index 31ac3b5c..978f4978 100644 --- a/client/src/main/java/io/oxia/client/grpc/GrpcRpcProvider.java +++ b/client/src/main/java/io/oxia/client/grpc/GrpcRpcProvider.java @@ -27,6 +27,7 @@ import io.oxia.client.ClientConfig; import io.oxia.client.grpc.observer.CancelableStreamObserver; import io.oxia.client.grpc.observer.ManagedObservers; +import io.oxia.client.util.TimeoutSweeper; import io.oxia.proto.CloseSessionRequest; import io.oxia.proto.CloseSessionResponse; import io.oxia.proto.CreateSessionRequest; @@ -71,6 +72,7 @@ final class GrpcRpcProvider implements RpcProvider { private final ScheduledExecutorService asyncExecutor; private final LongFunction shardLeaderProvider; private final Map writeStreams; + private final TimeoutSweeper requestTimeouts; GrpcRpcProvider( @NonNull ClientConfig clientConfig, @@ -104,6 +106,7 @@ private GrpcRpcProvider( this.ownsConnectionManager = ownsConnectionManager; this.shardLeaderProvider = shardLeaderProvider; this.writeStreams = Maps.newConcurrentMap(); + this.requestTimeouts = new TimeoutSweeper(asyncExecutor, clientConfig.requestTimeout()); } @Override @@ -117,9 +120,7 @@ public void getShardAssignments( .with(asyncExecutor) .getStageAsync( () -> { - final var barrierFuture = - new CompletableFuture() - .orTimeout(clientConfig.requestTimeout().toMillis(), TimeUnit.MILLISECONDS); + final var barrierFuture = requestTimeouts.add(new CompletableFuture()); final var barrierObserver = ManagedObservers.toBarrierStreamObserver(guardedObserver, barrierFuture); final var attemptContext = Context.current().withCancellation(); @@ -165,8 +166,7 @@ public void getNotifications( // resumed one gets nothing until a new notification is written, so it is only // bounded by the subscription max age, like the sequence updates. if (!request.hasStartOffsetExclusive()) { - barrierFuture.orTimeout( - clientConfig.requestTimeout().toMillis(), TimeUnit.MILLISECONDS); + requestTimeouts.add(barrierFuture); } final var barrierObserver = ManagedObservers.toBarrierClientResponseObserver(observer, barrierFuture); @@ -284,9 +284,7 @@ public void read(@NonNull ReadRequest request, @NonNull StreamObserver { - final var barrierFuture = - new CompletableFuture() - .orTimeout(clientConfig.requestTimeout().toMillis(), TimeUnit.MILLISECONDS); + final var barrierFuture = requestTimeouts.add(new CompletableFuture()); final var barrierObserver = ManagedObservers.toBarrierStreamObserver(guardedObserver, barrierFuture); final var attemptContext = Context.current().withCancellation(); @@ -364,9 +362,7 @@ public void list( .with(asyncExecutor) .getStageAsync( () -> { - final var barrierFuture = - new CompletableFuture() - .orTimeout(clientConfig.requestTimeout().toMillis(), TimeUnit.MILLISECONDS); + final var barrierFuture = requestTimeouts.add(new CompletableFuture()); final var barrierObserver = ManagedObservers.toBarrierClientResponseObserver(observer, barrierFuture); final var attemptContext = Context.current().withCancellation(); @@ -410,9 +406,7 @@ public void rangeScan( .with(asyncExecutor) .getStageAsync( () -> { - final var barrierFuture = - new CompletableFuture() - .orTimeout(clientConfig.requestTimeout().toMillis(), TimeUnit.MILLISECONDS); + final var barrierFuture = requestTimeouts.add(new CompletableFuture()); final var barrierObserver = ManagedObservers.toBarrierClientResponseObserver(observer, barrierFuture); final var attemptContext = Context.current().withCancellation(); @@ -495,6 +489,7 @@ private OxiaClientGrpc.OxiaClientStub withSubscriptionMaxAge(OxiaClientGrpc.Oxia @Override public void close() throws Exception { + requestTimeouts.close(); try { writeStreams.values().forEach(ManagedWriteStream::close); writeStreams.clear(); diff --git a/client/src/main/java/io/oxia/client/util/TimeoutSweeper.java b/client/src/main/java/io/oxia/client/util/TimeoutSweeper.java new file mode 100644 index 00000000..47b11ffb --- /dev/null +++ b/client/src/main/java/io/oxia/client/util/TimeoutSweeper.java @@ -0,0 +1,193 @@ +/* + * Copyright © 2026 The Oxia Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.oxia.client.util; + +import com.google.common.annotations.VisibleForTesting; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import lombok.NonNull; + +/** + * Fails the futures added to it with a {@link TimeoutException} when they are still pending after + * the timeout, like {@link CompletableFuture#orTimeout}, but without its cost: {@code orTimeout} + * schedules and then cancels a task on the single JVM-wide delayer for every future, taking its + * lock twice. Here a future is appended to a list picked by the calling thread, so that threads + * rarely contend, and a periodic sweep on the given executor fails the futures that expired. + * + *

The adding threads drop the completed futures from their list whenever it doubles in size, so + * the memory is bounded by the pending futures rather than by the rate at which they are added. + * + *

The timeout is approximate: a future fails between the timeout and the timeout plus two sweep + * intervals after it was added, never earlier. The sweep interval is a tenth of the timeout, and at + * most 100 ms. + */ +public final class TimeoutSweeper implements AutoCloseable { + + private static final long MAX_SWEEP_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(100); + @VisibleForTesting static final int MIN_COMPACTION_SIZE = 1024; + + private record Generation(long deadlineNanos, List> futures) {} + + /** The futures added since the last sweep by the threads that map to this stripe. */ + private static final class Stripe { + private List> futures = new ArrayList<>(); + private int compactionSize = MIN_COMPACTION_SIZE; + + synchronized void add(CompletableFuture future) { + futures.add(future); + if (futures.size() >= compactionSize) { + futures.removeIf(CompletableFuture::isDone); + compactionSize = Math.max(futures.size() * 2, MIN_COMPACTION_SIZE); + } + } + + synchronized List> takeAll() { + if (futures.isEmpty()) { + return List.of(); + } + final List> taken = futures; + futures = new ArrayList<>(); + compactionSize = MIN_COMPACTION_SIZE; + return taken; + } + + synchronized int size() { + return futures.size(); + } + } + + private final long timeoutNanos; + private final Stripe[] stripes; + private final ScheduledFuture sweepTask; + // The futures taken by each sweep, oldest first. Guarded by this. + private final ArrayDeque generations = new ArrayDeque<>(); + private volatile boolean closed; + + public TimeoutSweeper(@NonNull ScheduledExecutorService executor, @NonNull Duration timeout) { + this.timeoutNanos = timeout.toNanos(); + final int cpus = Runtime.getRuntime().availableProcessors(); + this.stripes = new Stripe[1 << (32 - Integer.numberOfLeadingZeros(cpus - 1))]; + for (int i = 0; i < stripes.length; i++) { + stripes[i] = new Stripe(); + } + final long intervalNanos = Math.min(Math.max(timeoutNanos / 10, 1), MAX_SWEEP_INTERVAL_NANOS); + this.sweepTask = + executor.scheduleWithFixedDelay( + () -> sweep(System.nanoTime()), intervalNanos, intervalNanos, TimeUnit.NANOSECONDS); + } + + /** Fails the future with a {@link TimeoutException} if it is still pending after the timeout. */ + public CompletableFuture add(@NonNull CompletableFuture future) { + if (!future.isDone()) { + stripes[(int) Thread.currentThread().getId() & (stripes.length - 1)].add(future); + if (closed) { + // The sweep no longer runs, and close() may have taken the stripes before this add + handOverToOrTimeout(); + } + } + return future; + } + + @VisibleForTesting + void sweep(long nowNanos) { + List> expired = new ArrayList<>(); + synchronized (this) { + if (closed) { + return; + } + final Iterator it = generations.iterator(); + while (it.hasNext()) { + final Generation generation = it.next(); + if (nowNanos - generation.deadlineNanos() >= 0) { + expired.addAll(generation.futures()); + it.remove(); + } else { + generation.futures().removeIf(CompletableFuture::isDone); + if (generation.futures().isEmpty()) { + it.remove(); + } + } + } + // The futures added since the last sweep start their timeout now: later than they were + // added, so that they never time out early + List> futures = null; + for (Stripe stripe : stripes) { + for (CompletableFuture future : stripe.takeAll()) { + if (!future.isDone()) { + if (futures == null) { + futures = new ArrayList<>(); + } + futures.add(future); + } + } + } + if (futures != null) { + generations.addLast(new Generation(nowNanos + timeoutNanos, futures)); + } + } + // Outside the lock: completing a future runs its callbacks + for (CompletableFuture future : expired) { + if (!future.isDone()) { + future.completeExceptionally(new TimeoutException()); + } + } + } + + @VisibleForTesting + synchronized int pendingCount() { + int count = 0; + for (Stripe stripe : stripes) { + count += stripe.size(); + } + for (Generation generation : generations) { + count += generation.futures().size(); + } + return count; + } + + /** + * Stops the sweep. The futures still pending keep their deadline through {@link + * CompletableFuture#orTimeout}, as the owner usually shuts down the executor next. + */ + @Override + public void close() { + sweepTask.cancel(false); + synchronized (this) { + closed = true; + final long now = System.nanoTime(); + for (Generation generation : generations) { + final long remainingNanos = Math.max(generation.deadlineNanos() - now, 0); + generation.futures().forEach(f -> f.orTimeout(remainingNanos, TimeUnit.NANOSECONDS)); + } + generations.clear(); + } + handOverToOrTimeout(); + } + + private void handOverToOrTimeout() { + for (Stripe stripe : stripes) { + stripe.takeAll().forEach(f -> f.orTimeout(timeoutNanos, TimeUnit.NANOSECONDS)); + } + } +} diff --git a/client/src/test/java/io/oxia/client/util/TimeoutSweeperTest.java b/client/src/test/java/io/oxia/client/util/TimeoutSweeperTest.java new file mode 100644 index 00000000..37f9b3a8 --- /dev/null +++ b/client/src/test/java/io/oxia/client/util/TimeoutSweeperTest.java @@ -0,0 +1,164 @@ +/* + * Copyright © 2026 The Oxia Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.oxia.client.util; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.awaitility.Awaitility.await; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.RETURNS_MOCKS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(30) +class TimeoutSweeperTest { + + private static final long TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(10); + + // The mocked executor never runs the periodic sweep: the tests sweep by hand, at chosen times + private final ScheduledExecutorService executor = + mock(ScheduledExecutorService.class, RETURNS_MOCKS); + private final TimeoutSweeper sweeper = + new TimeoutSweeper(executor, Duration.ofNanos(TIMEOUT_NANOS)); + + @Test + void failsPendingFutureAtTheDeadline() { + var future = sweeper.add(new CompletableFuture()); + + // The timeout starts at the first sweep after the future was added + sweeper.sweep(0); + sweeper.sweep(TIMEOUT_NANOS - 1); + assertThat(future).isNotDone(); + + sweeper.sweep(TIMEOUT_NANOS); + assertThatThrownBy(future::join).hasCauseInstanceOf(TimeoutException.class); + assertThat(sweeper.pendingCount()).isZero(); + } + + @Test + void dropsCompletedFutures() { + sweeper.add(CompletableFuture.completedFuture("already done")); + var completed = sweeper.add(new CompletableFuture()); + var pending = sweeper.add(new CompletableFuture()); + assertThat(sweeper.pendingCount()).isEqualTo(2); + + sweeper.sweep(0); + completed.complete("done"); + sweeper.sweep(1); + assertThat(sweeper.pendingCount()).isEqualTo(1); + + sweeper.sweep(TIMEOUT_NANOS); + assertThat(completed).isCompletedWithValue("done"); + assertThatThrownBy(pending::join).hasCauseInstanceOf(TimeoutException.class); + } + + @Test + void addingThreadDropsCompletedFutures() { + List> pending = new ArrayList<>(); + for (int i = 0; i < 10 * TimeoutSweeper.MIN_COMPACTION_SIZE; i++) { + var future = sweeper.add(new CompletableFuture()); + if (i % 100 == 0) { + pending.add(future); + } else { + future.complete("done"); + } + } + // Bounded by the pending futures, even without a sweep + assertThat(sweeper.pendingCount()).isLessThan(TimeoutSweeper.MIN_COMPACTION_SIZE); + + sweeper.sweep(0); + assertThat(sweeper.pendingCount()).isEqualTo(pending.size()); + sweeper.sweep(TIMEOUT_NANOS); + assertThat(pending) + .allSatisfy(f -> assertThatThrownBy(f::join).hasCauseInstanceOf(TimeoutException.class)); + } + + @Test + void sweepIntervalIsATenthOfTheTimeoutUpTo100Millis() { + new TimeoutSweeper(executor, Duration.ofMillis(500)); + + verify(executor) + .scheduleWithFixedDelay( + any(), + eq(TimeUnit.MILLISECONDS.toNanos(50)), + eq(TimeUnit.MILLISECONDS.toNanos(50)), + eq(TimeUnit.NANOSECONDS)); + // The 10 seconds timeout of the sweeper under test + verify(executor) + .scheduleWithFixedDelay( + any(), + eq(TimeUnit.MILLISECONDS.toNanos(100)), + eq(TimeUnit.MILLISECONDS.toNanos(100)), + eq(TimeUnit.NANOSECONDS)); + } + + @Test + void closeKeepsTheDeadlineOfPendingFutures() { + var future = sweeper.add(new CompletableFuture()); + // Leave 200 ms before the deadline + sweeper.sweep(System.nanoTime() - TIMEOUT_NANOS + TimeUnit.MILLISECONDS.toNanos(200)); + + sweeper.close(); + assertThat(future).isNotDone(); + await() + .atMost(Duration.ofSeconds(5)) + .untilAsserted( + () -> assertThatThrownBy(future::join).hasCauseInstanceOf(TimeoutException.class)); + } + + @Test + void futureAddedAfterCloseStillTimesOut() { + var closedSweeper = new TimeoutSweeper(executor, Duration.ofMillis(200)); + closedSweeper.close(); + + var future = closedSweeper.add(new CompletableFuture()); + assertThat(future).isNotDone(); + await() + .atMost(Duration.ofSeconds(5)) + .untilAsserted( + () -> assertThatThrownBy(future::join).hasCauseInstanceOf(TimeoutException.class)); + } + + @Test + void periodicSweepFailsPendingFutureNoEarlierThanTheTimeout() { + var realExecutor = Executors.newSingleThreadScheduledExecutor(); + try (var realSweeper = new TimeoutSweeper(realExecutor, Duration.ofMillis(200))) { + var failedAt = new AtomicLong(); + long start = System.nanoTime(); + var future = realSweeper.add(new CompletableFuture()); + future.whenComplete((v, e) -> failedAt.set(System.nanoTime())); + + await().atMost(Duration.ofSeconds(5)).until(() -> failedAt.get() != 0); + assertThatThrownBy(future::join).hasCauseInstanceOf(TimeoutException.class); + assertThat(failedAt.get() - start).isGreaterThanOrEqualTo(TimeUnit.MILLISECONDS.toNanos(200)); + } finally { + realExecutor.shutdownNow(); + } + } +}