-
Notifications
You must be signed in to change notification settings - Fork 80
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
BarrageSession Subscription/Snapshot Methods now Return `Future<Tab…
…le>` (#4676) This includes breaking changes to the barrage java-client API. The snapshot/subscription methods now return `Future<Table>` instead of `BarrageTable` and users are no longer need to participate via `blockUntilCompletion` as this is forced by `Future#get`. Subscribed tables can now be canceled only through `Liveness` destruction such as using an explicit `LivenessScope` or invoking `ReferenceCounted#forceReferenceCountToZero()`. --------- Co-authored-by: Ryan Caudy <[email protected]>
- Loading branch information
1 parent
8225c90
commit 43c10f4
Showing
12 changed files
with
565 additions
and
538 deletions.
There are no files selected for viewing
20 changes: 20 additions & 0 deletions
20
engine/table/src/main/java/io/deephaven/engine/exceptions/RequestCancelledException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| /** | ||
| * Copyright (c) 2016-2023 Deephaven Data Labs and Patent Pending | ||
| */ | ||
| package io.deephaven.engine.exceptions; | ||
|
|
||
| import io.deephaven.UncheckedDeephavenException; | ||
| import org.jetbrains.annotations.NotNull; | ||
|
|
||
| /** | ||
| * This exception is used when a result cannot be returned because the request was cancelled. | ||
| */ | ||
| public class RequestCancelledException extends UncheckedDeephavenException { | ||
| public RequestCancelledException(@NotNull final String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| public RequestCancelledException(@NotNull final String message, @NotNull final Throwable cause) { | ||
| super(message, cause); | ||
| } | ||
| } |
169 changes: 169 additions & 0 deletions
169
...raph/src/main/java/io/deephaven/engine/updategraph/UpdateGraphAwareCompletableFuture.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,169 @@ | ||
| package io.deephaven.engine.updategraph; | ||
|
|
||
| import io.deephaven.util.SafeCloseable; | ||
| import io.deephaven.util.function.ThrowingSupplier; | ||
| import io.deephaven.util.locks.FunctionalLock; | ||
| import io.deephaven.util.locks.FunctionalReentrantLock; | ||
| import org.jetbrains.annotations.NotNull; | ||
| import org.jetbrains.annotations.Nullable; | ||
|
|
||
| import java.util.Objects; | ||
| import java.util.concurrent.*; | ||
| import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; | ||
| import java.util.concurrent.locks.Condition; | ||
|
|
||
| public class UpdateGraphAwareCompletableFuture<T> implements Future<T> { | ||
|
|
||
| private final UpdateGraph updateGraph; | ||
|
|
||
| /** This condition is used to signal any threads waiting on the UpdateGraph exclusive lock. */ | ||
| private volatile Condition updateGraphCondition; | ||
|
|
||
| private final FunctionalLock lock = new FunctionalReentrantLock(); | ||
| private volatile Condition lockCondition; | ||
|
|
||
| private volatile ThrowingSupplier<T, ExecutionException> resultSupplier; | ||
| @SuppressWarnings("rawtypes") | ||
| private static final AtomicReferenceFieldUpdater<UpdateGraphAwareCompletableFuture, ThrowingSupplier> RESULT_UPDATER = | ||
| AtomicReferenceFieldUpdater.newUpdater( | ||
| UpdateGraphAwareCompletableFuture.class, ThrowingSupplier.class, "resultSupplier"); | ||
|
|
||
| /** The encoding of the cancelled supplier. */ | ||
| private static final ThrowingSupplier<?, ExecutionException> CANCELLATION_SUPPLIER = () -> { | ||
| throw new CancellationException(); | ||
| }; | ||
|
|
||
| public UpdateGraphAwareCompletableFuture(@NotNull final UpdateGraph updateGraph) { | ||
| this.updateGraph = updateGraph; | ||
| } | ||
|
|
||
| //////////////// | ||
| // Future API // | ||
| //////////////// | ||
| @Override | ||
| public boolean cancel(boolean mayInterruptIfRunning) { | ||
| // noinspection unchecked | ||
| return trySignalCompletion((ThrowingSupplier<T, ExecutionException>) CANCELLATION_SUPPLIER); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isCancelled() { | ||
| return resultSupplier == CANCELLATION_SUPPLIER; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isDone() { | ||
| return resultSupplier != null; | ||
| } | ||
|
|
||
| @Override | ||
| public T get() throws InterruptedException, ExecutionException { | ||
| checkSharedLockState(); | ||
|
|
||
| if (resultSupplier != null) { | ||
| return resultSupplier.get(); | ||
| } | ||
| try { | ||
| return getInternal(0, null); | ||
| } catch (TimeoutException toe) { | ||
| throw new IllegalStateException("Unexpected TimeoutException", toe); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public T get(final long timeout, @NotNull final TimeUnit unit) | ||
| throws InterruptedException, ExecutionException, TimeoutException { | ||
| checkSharedLockState(); | ||
|
|
||
| if (resultSupplier != null) { | ||
| return resultSupplier.get(); | ||
| } | ||
| if (timeout <= 0) { | ||
| throw new TimeoutException(); | ||
| } | ||
| return getInternal(timeout, unit); | ||
| } | ||
|
|
||
| private T getInternal(final long timeout, @Nullable final TimeUnit unit) | ||
| throws InterruptedException, ExecutionException, TimeoutException { | ||
| final boolean holdingUpdateGraphLock = updateGraph.exclusiveLock().isHeldByCurrentThread(); | ||
| if (holdingUpdateGraphLock) { | ||
| if (updateGraphCondition == null) { | ||
| updateGraphCondition = updateGraph.exclusiveLock().newCondition(); | ||
| } | ||
| } else if (lockCondition == null) { | ||
| try (final SafeCloseable ignored = lock.lockCloseable()) { | ||
| if (lockCondition == null) { | ||
| lockCondition = lock.newCondition(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if (holdingUpdateGraphLock) { | ||
| waitForResult(updateGraphCondition, timeout, unit); | ||
| } else { | ||
| try (final SafeCloseable ignored = lock.lockCloseable()) { | ||
| waitForResult(lockCondition, timeout, unit); | ||
| } | ||
| } | ||
|
|
||
| return resultSupplier.get(); | ||
| } | ||
|
|
||
| private void checkSharedLockState() { | ||
| if (updateGraph.sharedLock().isHeldByCurrentThread()) { | ||
| throw new UnsupportedOperationException( | ||
| "Cannot Future.get(...) while holding the " + updateGraph + " shared lock"); | ||
| } | ||
| } | ||
|
|
||
| private void waitForResult(final Condition condition, final long timeout, @Nullable final TimeUnit unit) | ||
| throws InterruptedException, TimeoutException { | ||
| if (unit == null) { | ||
| while (resultSupplier == null) { | ||
| condition.await(); | ||
| } | ||
| return; | ||
| } | ||
|
|
||
| long nanosLeft = unit.toNanos(timeout); | ||
| while (resultSupplier == null) { | ||
| nanosLeft = condition.awaitNanos(nanosLeft); | ||
| if (nanosLeft <= 0) { | ||
| throw new TimeoutException(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| //////////////////////////////////////////////////// | ||
| // Completion API modeled after CompletableFuture // | ||
| //////////////////////////////////////////////////// | ||
|
|
||
| public boolean complete(T value) { | ||
| return trySignalCompletion(() -> value); | ||
| } | ||
|
|
||
| public boolean completeExceptionally(Throwable ex) { | ||
| Objects.requireNonNull(ex); | ||
| return trySignalCompletion(() -> { | ||
| throw new ExecutionException(ex); | ||
| }); | ||
| } | ||
|
|
||
| private boolean trySignalCompletion(@NotNull final ThrowingSupplier<T, ExecutionException> result) { | ||
| if (!RESULT_UPDATER.compareAndSet(this, null, result)) { | ||
| return false; | ||
| } | ||
|
|
||
| final Condition localUpdateGraphCondition = updateGraphCondition; | ||
| if (localUpdateGraphCondition != null) { | ||
| updateGraph.requestSignal(localUpdateGraphCondition); | ||
| } | ||
| final Condition localLockCondition = lockCondition; | ||
| if (localLockCondition != null) { | ||
| lock.doLocked(localLockCondition::signalAll); | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.