This is an automated email from the ASF dual-hosted git repository. ifesdjeen pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/cassandra-simulator.git
commit cb00523e42288f3ba63e9c3fee956a0fd64d455f Author: Alex Petrov <[email protected]> AuthorDate: Thu Jul 23 17:54:03 2026 +0200 Refactor FUTURE API --- .../simulator/systems/InterceptedExecution.java | 2 +- .../simulator/systems/InterceptingExecutor.java | 63 +++++---- .../simulator/systems/InterceptorOfExecution.java | 2 +- .../simulator/systems/SimulatedExecution.java | 4 +- ...yncFutureTask.java => SimulatorFutureTask.java} | 36 ++--- .../cassandra/simulator/systems/TaskFactory.java | 8 +- .../simulator/utils/concurrent/Future.java | 61 -------- .../utils/concurrent/ImmediateFuture.java | 68 ++++----- .../utils/concurrent/NotScheduledFuture.java | 21 ++- .../simulator/utils/concurrent/RunnableFuture.java | 28 ---- .../simulator/utils/concurrent/SyncFuture.java | 155 --------------------- 11 files changed, 113 insertions(+), 335 deletions(-) diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedExecution.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedExecution.java index 4247eeb..547facd 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedExecution.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptedExecution.java @@ -18,13 +18,13 @@ package org.apache.cassandra.simulator.systems; +import java.util.concurrent.RunnableFuture; import java.util.function.Function; import com.google.common.base.Preconditions; import org.apache.cassandra.simulator.systems.NotifyThreadPaused.AwaitPaused; import org.apache.cassandra.simulator.utils.Shared; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; import org.apache.cassandra.simulator.utils.UncheckedInterruptedException; import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingExecutor.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingExecutor.java index cbde83f..c839e6f 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingExecutor.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptingExecutor.java @@ -20,6 +20,7 @@ package org.apache.cassandra.simulator.systems; import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.IdentityHashMap; import java.util.List; @@ -29,12 +30,17 @@ import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import org.apache.cassandra.simulator.OrderOn; @@ -43,10 +49,8 @@ import org.apache.cassandra.simulator.systems.NotifyThreadPaused.AwaitPaused; import org.apache.cassandra.simulator.utils.Shared; import org.apache.cassandra.simulator.utils.WithResources; import org.apache.cassandra.simulator.utils.concurrent.Condition; -import org.apache.cassandra.simulator.utils.concurrent.Future; import org.apache.cassandra.simulator.utils.concurrent.ImmediateFuture; import org.apache.cassandra.simulator.utils.concurrent.NotScheduledFuture; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; import org.apache.cassandra.simulator.utils.UncheckedInterruptedException; import static java.util.Collections.newSetFromMap; @@ -258,25 +262,25 @@ public interface InterceptingExecutor extends OrderOn // ExecutorService invokeAll/invokeAny - not supported @Override - public <T> List<java.util.concurrent.Future<T>> invokeAll(java.util.Collection<? extends Callable<T>> tasks) throws InterruptedException + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { throw new UnsupportedOperationException(); } @Override - public <T> List<java.util.concurrent.Future<T>> invokeAll(java.util.Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException { throw new UnsupportedOperationException(); } @Override - public <T> T invokeAny(java.util.Collection<? extends Callable<T>> tasks) throws InterruptedException, java.util.concurrent.ExecutionException + public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { throw new UnsupportedOperationException(); } @Override - public <T> T invokeAny(java.util.Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException + public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { throw new UnsupportedOperationException(); } @@ -532,7 +536,7 @@ public interface InterceptingExecutor extends OrderOn @PerClassLoader abstract class AbstractSingleThreadedExecutorPlus extends AbstractInterceptingExecutor { - static class AtLeastOnce extends java.util.concurrent.atomic.AtomicBoolean implements AtLeastOnceTrigger, Runnable + static class AtLeastOnce extends AtomicBoolean implements AtLeastOnceTrigger, Runnable { private static final long serialVersionUID = 0L; protected final AbstractSingleThreadedExecutorPlus executor; @@ -559,7 +563,18 @@ public interface InterceptingExecutor extends OrderOn public void sync() { Future<?> done = executor.submit(() -> {}); - done.awaitThrowUncheckedOnInterrupt(); + try + { + done.get(); + } + catch (InterruptedException e) + { + throw new UncheckedInterruptedException(e); + } + catch (ExecutionException e) + { + throw new AssertionError(e.getCause()); + } } public void run() { set(false); run.run(); } @@ -724,7 +739,7 @@ public interface InterceptingExecutor extends OrderOn @PerClassLoader class InterceptingSequentialExecutor extends AbstractSingleThreadedExecutorPlus implements InterceptingExecutor, ScheduledExecutorService, OrderOn { - static class InterceptableScheduledFutureTask<T> extends SyncFutureTask<T> implements InterceptableScheduledFuture<T> + static class InterceptableScheduledFutureTask<T> extends SimulatorFutureTask<T> implements InterceptableScheduledFuture<T> { final long delayNanos; Runnable onCancel; @@ -948,43 +963,43 @@ public interface InterceptingExecutor extends OrderOn } @Override - public <T> java.util.concurrent.Future<T> submit(Callable<T> task) + public <T> Future<T> submit(Callable<T> task) { return ImmediateFuture.cancelled(); } @Override - public <T> java.util.concurrent.Future<T> submit(Runnable task, T result) + public <T> Future<T> submit(Runnable task, T result) { return ImmediateFuture.cancelled(); } @Override - public java.util.concurrent.Future<?> submit(Runnable task) + public Future<?> submit(Runnable task) { return ImmediateFuture.cancelled(); } @Override - public <T> List<java.util.concurrent.Future<T>> invokeAll(java.util.Collection<? extends Callable<T>> tasks) throws InterruptedException + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException { throw new UnsupportedOperationException(); } @Override - public <T> List<java.util.concurrent.Future<T>> invokeAll(java.util.Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException + public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException { throw new UnsupportedOperationException(); } @Override - public <T> T invokeAny(java.util.Collection<? extends Callable<T>> tasks) throws InterruptedException, java.util.concurrent.ExecutionException + public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException { throw new UnsupportedOperationException(); } @Override - public <T> T invokeAny(java.util.Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, java.util.concurrent.ExecutionException, java.util.concurrent.TimeoutException + public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { throw new UnsupportedOperationException(); } @@ -1018,46 +1033,46 @@ public interface InterceptingExecutor extends OrderOn public ScheduledFuture<?> scheduleSelfRecurring(Runnable run, long delay, TimeUnit units) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } public ScheduledFuture<?> scheduleAt(Runnable run, long deadline) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } public ScheduledFuture<?> scheduleTimeoutAt(Runnable run, long deadline) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } public ScheduledFuture<?> scheduleTimeoutWithDelay(Runnable run, long delay, TimeUnit units) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } @Override public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } @Override public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } @Override public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } } } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfExecution.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfExecution.java index da3c03f..08b0a3c 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfExecution.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/InterceptorOfExecution.java @@ -18,13 +18,13 @@ package org.apache.cassandra.simulator.systems; +import java.util.concurrent.RunnableFuture; import java.util.concurrent.ScheduledFuture; import java.util.function.Function; import org.apache.cassandra.simulator.systems.InterceptingExecutor.InterceptableScheduledFuture; import org.apache.cassandra.simulator.systems.SimulatedAction.Kind; import org.apache.cassandra.simulator.utils.Shared; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedExecution.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedExecution.java index d2e68ae..d06592d 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedExecution.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatedExecution.java @@ -19,6 +19,7 @@ package org.apache.cassandra.simulator.systems; import java.util.concurrent.Callable; +import java.util.concurrent.RunnableFuture; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; @@ -33,7 +34,6 @@ import org.apache.cassandra.simulator.systems.InterceptedExecution.InterceptedTh import org.apache.cassandra.simulator.systems.InterceptingExecutor.InterceptableScheduledFuture; import org.apache.cassandra.simulator.utils.concurrent.Condition; import org.apache.cassandra.simulator.utils.concurrent.NotScheduledFuture; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; import static org.apache.cassandra.simulator.systems.SimulatedAction.Kind.SCHEDULED_DAEMON; import static org.apache.cassandra.simulator.systems.SimulatedAction.Kind.SCHEDULED_TASK; @@ -97,7 +97,7 @@ public class SimulatedExecution implements InterceptorOfExecution public <T> ScheduledFuture<T> schedule(SimulatedAction.Kind kind, long delayNanos, long deadlineNanos, InterceptableScheduledFuture<T> task, InterceptingExecutor executor) { - return new NotScheduledFuture<>(); + return NotScheduledFuture.instance(); } public Thread start(SimulatedAction.Kind kind, Function<Runnable, InterceptibleThread> factory, Runnable run) diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SyncFutureTask.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatorFutureTask.java similarity index 55% rename from simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SyncFutureTask.java rename to simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatorFutureTask.java index 400459c..fc2b7ae 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SyncFutureTask.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/SimulatorFutureTask.java @@ -19,40 +19,28 @@ package org.apache.cassandra.simulator.systems; import java.util.concurrent.Callable; +import java.util.concurrent.FutureTask; import org.apache.cassandra.simulator.utils.Closeable; import org.apache.cassandra.simulator.utils.WithResources; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; -import org.apache.cassandra.simulator.utils.concurrent.SyncFuture; -public class SyncFutureTask<T> extends SyncFuture<T> implements RunnableFuture<T> +/** + * Adapts simulator resource scopes to a standard FutureTask. + */ +class SimulatorFutureTask<T> extends FutureTask<T> { - final Callable<T> call; - - public SyncFutureTask(Callable<T> call) { this.call = call; } - - public SyncFutureTask(WithResources withResources, Callable<T> call) + SimulatorFutureTask(Callable<T> callable) { - this.call = () -> { - try (Closeable close = withResources.get()) { return call.call(); } - }; + super(callable); } - public void run() + SimulatorFutureTask(WithResources withResources, Callable<T> callable) { - try - { - if (!setUncancellable()) + super(() -> { + try (Closeable close = withResources.get()) { - if (isCancelled()) return; - else throw new IllegalStateException(); + return callable.call(); } - if (!trySuccess(call.call())) - throw new IllegalStateException(); - } - catch (Throwable t) - { - tryFailure(t); - } + }); } } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/TaskFactory.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/TaskFactory.java index c517145..448eac1 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/TaskFactory.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/systems/TaskFactory.java @@ -20,9 +20,9 @@ package org.apache.cassandra.simulator.systems; import java.io.Serializable; import java.util.concurrent.Callable; +import java.util.concurrent.RunnableFuture; import org.apache.cassandra.simulator.utils.WithResources; -import org.apache.cassandra.simulator.utils.concurrent.RunnableFuture; public interface TaskFactory { @@ -71,9 +71,9 @@ public interface TaskFactory return withResources.isNoOp() ? toSubmit(callable) : newTask(withResources, callable); } - protected <T> RunnableFuture<T> newTask(Callable<T> call) { return new SyncFutureTask<>(call); } - protected <T> RunnableFuture<T> newTask(WithResources withResources, Callable<T> call) { return new SyncFutureTask<>(withResources, call); } - protected <T> RunnableFuture<T> newTask(WithResources withResources, Runnable run, T result) { return new SyncFutureTask<>(withResources, toCallable(run, result)); } + protected <T> RunnableFuture<T> newTask(Callable<T> call) { return new SimulatorFutureTask<>(call); } + protected <T> RunnableFuture<T> newTask(WithResources withResources, Callable<T> call) { return new SimulatorFutureTask<>(withResources, call); } + protected <T> RunnableFuture<T> newTask(WithResources withResources, Runnable run, T result) { return new SimulatorFutureTask<>(withResources, toCallable(run, result)); } } class LocalAware extends Standard diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/Future.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/Future.java deleted file mode 100644 index 7da16b3..0000000 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/Future.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 org.apache.cassandra.simulator.utils.concurrent; - -import java.util.concurrent.TimeUnit; - -import org.apache.cassandra.simulator.utils.Shared; -import org.apache.cassandra.simulator.utils.UncheckedInterruptedException; - -import static org.apache.cassandra.simulator.utils.Shared.Recursive.INTERFACES; -import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; - -@Shared(scope = SIMULATION, ancestors = INTERFACES, members = INTERFACES) -public interface Future<V> extends java.util.concurrent.Future<V>, Awaitable -{ - boolean isSuccess(); - Throwable cause(); - - @Override - default Future<V> await() throws InterruptedException { return this; } - - @Override - default Future<V> awaitUninterruptibly() { return this; } - - @Override - default Future<V> awaitThrowUncheckedOnInterrupt() throws UncheckedInterruptedException { return this; } - - @Override - default boolean awaitUntil(long nanoTimeDeadline) throws InterruptedException { return true; } - - @Override - default boolean awaitUntilUninterruptibly(long nanoTimeDeadline) { return true; } - - @Override - default boolean awaitUntilThrowUncheckedOnInterrupt(long nanoTimeDeadline) throws UncheckedInterruptedException { return true; } - - @Override - default boolean await(long time, TimeUnit units) throws InterruptedException { return true; } - - @Override - default boolean awaitUninterruptibly(long time, TimeUnit units) { return true; } - - @Override - default boolean awaitThrowUncheckedOnInterrupt(long time, TimeUnit units) throws UncheckedInterruptedException { return true; } -} diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/ImmediateFuture.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/ImmediateFuture.java index 0bf3f63..b37e13d 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/ImmediateFuture.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/ImmediateFuture.java @@ -18,48 +18,54 @@ package org.apache.cassandra.simulator.utils.concurrent; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.CancellationException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -public class ImmediateFuture<V> implements Future<V> +/** + * An already-cancelled Future returned by executors that intentionally discard work. + */ +public final class ImmediateFuture<V> implements Future<V> { - private final V value; - private final Throwable cause; - private final boolean cancelled; + private static final ImmediateFuture<?> CANCELLED = new ImmediateFuture<>(); + + private ImmediateFuture() + { + } - private ImmediateFuture(V value, Throwable cause, boolean cancelled) + @SuppressWarnings("unchecked") + public static <V> ImmediateFuture<V> cancelled() { - this.value = value; - this.cause = cause; - this.cancelled = cancelled; + return (ImmediateFuture<V>) CANCELLED; } - public static <V> ImmediateFuture<V> success(V value) { return new ImmediateFuture<>(value, null, false); } - public static <V> ImmediateFuture<V> failure(Throwable cause) { return new ImmediateFuture<>(null, cause, false); } - public static <V> ImmediateFuture<V> cancelled() { return new ImmediateFuture<>(null, null, true); } + @Override + public boolean cancel(boolean mayInterruptIfRunning) + { + return false; + } - public boolean isSuccess() { return cause == null && !cancelled; } - public Throwable cause() { return cause; } - public boolean cancel(boolean b) { return false; } - public boolean isCancelled() { return cancelled; } - public boolean isDone() { return true; } + @Override + public boolean isCancelled() + { + return true; + } - public V get() throws ExecutionException + @Override + public boolean isDone() { - if (cancelled) throw new java.util.concurrent.CancellationException(); - if (cause != null) throw new ExecutionException(cause); - return value; + return true; } - public V get(long l, TimeUnit u) throws ExecutionException { return get(); } + @Override + public V get() + { + throw new CancellationException(); + } - public boolean awaitUntil(long d) throws InterruptedException { return true; } - public boolean awaitUntilThrowUncheckedOnInterrupt(long d) { return true; } - public boolean awaitUntilUninterruptibly(long d) { return true; } - public boolean await(long t, TimeUnit u) throws InterruptedException { return true; } - public boolean awaitThrowUncheckedOnInterrupt(long t, TimeUnit u) { return true; } - public boolean awaitUninterruptibly(long t, TimeUnit u) { return true; } - public Future<V> await() throws InterruptedException { return this; } - public Future<V> awaitThrowUncheckedOnInterrupt() { return this; } - public Future<V> awaitUninterruptibly() { return this; } + @Override + public V get(long timeout, TimeUnit unit) + { + throw new CancellationException(); + } } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/NotScheduledFuture.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/NotScheduledFuture.java index 7aee7a9..5cdca53 100644 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/NotScheduledFuture.java +++ b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/NotScheduledFuture.java @@ -19,15 +19,28 @@ package org.apache.cassandra.simulator.utils.concurrent; import java.util.concurrent.Delayed; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -public class NotScheduledFuture<T> implements ScheduledFuture<T> +/** + * Sentinel returned when scheduling is intentionally disabled. + */ +public final class NotScheduledFuture<T> implements ScheduledFuture<T> { + private static final NotScheduledFuture<?> INSTANCE = new NotScheduledFuture<>(); + + private NotScheduledFuture() + { + } + + @SuppressWarnings("unchecked") + public static <T> NotScheduledFuture<T> instance() + { + return (NotScheduledFuture<T>) INSTANCE; + } + public long getDelay(TimeUnit unit) { return 0; } - public int compareTo(Delayed o) { return 0; } + public int compareTo(Delayed other) { return 0; } public boolean cancel(boolean mayInterruptIfRunning) { return false; } public boolean isCancelled() { return false; } public boolean isDone() { return false; } diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/RunnableFuture.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/RunnableFuture.java deleted file mode 100644 index dd3bd8a..0000000 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/RunnableFuture.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 org.apache.cassandra.simulator.utils.concurrent; - -import org.apache.cassandra.simulator.utils.Shared; - -import static org.apache.cassandra.simulator.utils.Shared.Scope.SIMULATION; - -@Shared(scope = SIMULATION) -public interface RunnableFuture<V> extends Future<V>, java.util.concurrent.RunnableFuture<V> -{ -} diff --git a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/SyncFuture.java b/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/SyncFuture.java deleted file mode 100644 index 54deb00..0000000 --- a/simulator-core/src/main/java/org/apache/cassandra/simulator/utils/concurrent/SyncFuture.java +++ /dev/null @@ -1,155 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 org.apache.cassandra.simulator.utils.concurrent; - -import org.apache.cassandra.simulator.utils.UncheckedInterruptedException; - -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; - -/** - * Minimal stub of SyncFuture for the simulator. - */ -public class SyncFuture<V> implements RunnableFuture<V> -{ - private static final Object UNSET = new Object(); - private static final Object CANCELLED = new Object(); - - private static final AtomicReferenceFieldUpdater<SyncFuture, Object> resultUpdater = - AtomicReferenceFieldUpdater.newUpdater(SyncFuture.class, Object.class, "result"); - - private volatile Object result = UNSET; - - public boolean trySuccess(V value) - { - if (resultUpdater.compareAndSet(this, UNSET, value)) - { - synchronized (this) { notifyAll(); } - return true; - } - return false; - } - - public boolean tryFailure(Throwable cause) - { - if (resultUpdater.compareAndSet(this, UNSET, new FailureHolder(cause))) - { - synchronized (this) { notifyAll(); } - return true; - } - return false; - } - - public boolean setUncancellable() { return result == UNSET; } - - public boolean isSuccess() { Object r = result; return r != UNSET && r != CANCELLED && !(r instanceof FailureHolder); } - public Throwable cause() { Object r = result; return r instanceof FailureHolder ? ((FailureHolder) r).cause : null; } - - public boolean cancel(boolean b) - { - if (resultUpdater.compareAndSet(this, UNSET, CANCELLED)) - { - synchronized (this) { notifyAll(); } - return true; - } - return false; - } - - public boolean isCancelled() { return result == CANCELLED; } - public boolean isDone() { return result != UNSET; } - - @SuppressWarnings("unchecked") - public V get() throws InterruptedException, ExecutionException - { - synchronized (this) { while (result == UNSET) wait(); } - Object r = result; - if (r == CANCELLED) throw new java.util.concurrent.CancellationException(); - if (r instanceof FailureHolder) throw new ExecutionException(((FailureHolder) r).cause); - return (V) r; - } - - public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, java.util.concurrent.TimeoutException - { - long deadline = System.nanoTime() + unit.toNanos(timeout); - synchronized (this) - { - while (result == UNSET) - { - long remaining = deadline - System.nanoTime(); - if (remaining <= 0) throw new java.util.concurrent.TimeoutException(); - wait(TimeUnit.NANOSECONDS.toMillis(remaining) + 1); - } - } - return get(); - } - - public void run() { throw new UnsupportedOperationException(); } - - public boolean awaitUntil(long nanoTimeDeadline) throws InterruptedException - { - synchronized (this) - { - while (result == UNSET) - { - long remaining = nanoTimeDeadline - System.nanoTime(); - if (remaining <= 0) return false; - wait(TimeUnit.NANOSECONDS.toMillis(remaining) + 1); - } - } - return true; - } - - public boolean awaitUntilThrowUncheckedOnInterrupt(long d) throws UncheckedInterruptedException - { - try { return awaitUntil(d); } catch (InterruptedException e) { throw new UncheckedInterruptedException(e); } - } - - public boolean awaitUntilUninterruptibly(long nanoTimeDeadline) - { - boolean interrupted = false; - try { while (true) { try { return awaitUntil(nanoTimeDeadline); } catch (InterruptedException e) { interrupted = true; } } } - finally { if (interrupted) Thread.currentThread().interrupt(); } - } - - public boolean await(long time, TimeUnit units) throws InterruptedException - { - return awaitUntil(System.nanoTime() + units.toNanos(time)); - } - - public boolean awaitThrowUncheckedOnInterrupt(long t, TimeUnit u) throws UncheckedInterruptedException - { - try { return await(t, u); } catch (InterruptedException e) { throw new UncheckedInterruptedException(e); } - } - - public boolean awaitUninterruptibly(long t, TimeUnit u) - { - return awaitUntilUninterruptibly(System.nanoTime() + u.toNanos(t)); - } - - public synchronized Future<V> await() throws InterruptedException { while (result == UNSET) wait(); return this; } - public Future<V> awaitThrowUncheckedOnInterrupt() throws UncheckedInterruptedException { try { return await(); } catch (InterruptedException e) { throw new UncheckedInterruptedException(e); } } - public synchronized Future<V> awaitUninterruptibly() { boolean i = false; try { while (result == UNSET) { try { wait(); } catch (InterruptedException e) { i = true; } } return this; } finally { if (i) Thread.currentThread().interrupt(); } } - - private static class FailureHolder - { - final Throwable cause; - FailureHolder(Throwable cause) { this.cause = cause; } - } -} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
