This is an automated email from the ASF dual-hosted git repository. benedict pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 0c444a75e79da8c157813a72d61f2b2f86e187ba Author: Benedict Elliott Smith <[email protected]> AuthorDate: Wed Jul 28 20:04:03 2021 +0100 [CASSANDRA-16931] CEP-10 Phase 2: Improve DTest @Shared Annotation Functionality --- .../cassandra/concurrent/ExecutorBuilder.java | 4 + .../concurrent/ExecutorBuilderFactory.java | 6 + .../cassandra/concurrent/ExecutorFactory.java | 4 + .../apache/cassandra/concurrent/ExecutorPlus.java | 5 + .../apache/cassandra/concurrent/Interruptible.java | 5 + .../concurrent/LocalAwareExecutorPlus.java | 5 + .../LocalAwareSequentialExecutorPlus.java | 5 + .../cassandra/concurrent/ResizableThreadPool.java | 5 + .../concurrent/ScheduledExecutorPlus.java | 5 + .../concurrent/SequentialExecutorPlus.java | 6 + .../apache/cassandra/concurrent/Shutdownable.java | 5 + .../apache/cassandra/concurrent/TaskFactory.java | 4 + .../cassandra/config/ParameterizedClass.java | 5 + .../cassandra/exceptions/CassandraException.java | 5 + .../apache/cassandra/exceptions/ExceptionCode.java | 4 + .../exceptions/RequestExecutionException.java | 5 + .../cassandra/exceptions/TransportException.java | 5 + .../apache/cassandra/io/util/DataInputPlus.java | 4 + .../apache/cassandra/io/util/DataOutputPlus.java | 6 +- .../org/apache/cassandra/io/util/FileReader.java | 1 - src/java/org/apache/cassandra/io/util/Memory.java | 2 +- .../util/ReadableMemory.java} | 14 +- .../cassandra/streaming/StreamingChannel.java | 5 + .../streaming/StreamingDataInputPlus.java | 4 + .../streaming/StreamingDataOutputPlus.java | 5 + src/java/org/apache/cassandra/utils/Clock.java | 2 + src/java/org/apache/cassandra/utils/Closeable.java | 3 + .../java/org/apache/cassandra/utils}/Isolated.java | 4 +- .../org/apache/cassandra/utils/MonotonicClock.java | 2 + .../cassandra/utils/MonotonicClockTranslation.java | 3 + .../java/org/apache/cassandra/utils}/Shared.java | 8 +- .../org/apache/cassandra/utils/WithResources.java | 3 + .../cassandra/utils/concurrent/Awaitable.java | 3 + .../cassandra/utils/concurrent/Condition.java | 4 + .../cassandra/utils/concurrent/CountDownLatch.java | 4 + .../apache/cassandra/utils/concurrent/Future.java | 4 + .../apache/cassandra/utils/concurrent/Promise.java | 5 + .../cassandra/utils/concurrent/RunnableFuture.java | 5 + .../cassandra/utils/concurrent/Semaphore.java | 3 + .../concurrent/UncheckedInterruptedException.java | 5 + .../cassandra/utils/concurrent/WaitQueue.java | 5 + .../org/apache/cassandra/distributed/Cluster.java | 3 +- .../distributed/impl/AbstractCluster.java | 239 +++++++++++++++++++-- .../cassandra/distributed/impl/Coordinator.java | 2 +- .../distributed/impl/INodeProvisionStrategy.java | 6 +- .../cassandra/distributed/impl/InstanceConfig.java | 2 - .../apache/cassandra/distributed/impl/Query.java | 5 +- .../cassandra/distributed/shared/ClusterUtils.java | 1 + .../test/BootstrapBinaryDisabledTest.java | 2 +- ...AsHibernatingNodeWithoutReplaceAddressTest.java | 8 +- .../upgrade/MixedModeMessageForwardTest.java | 3 +- 51 files changed, 409 insertions(+), 49 deletions(-) diff --git a/src/java/org/apache/cassandra/concurrent/ExecutorBuilder.java b/src/java/org/apache/cassandra/concurrent/ExecutorBuilder.java index 89ca28a..c1d39d5 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutorBuilder.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutorBuilder.java @@ -26,11 +26,15 @@ import java.util.concurrent.TimeUnit; import com.google.common.annotations.VisibleForTesting; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Configure an executor before creating it. * See {@link ThreadPoolExecutorBuilder} */ +@Shared(scope = SIMULATION) public interface ExecutorBuilder<E extends ExecutorService> { /** diff --git a/src/java/org/apache/cassandra/concurrent/ExecutorBuilderFactory.java b/src/java/org/apache/cassandra/concurrent/ExecutorBuilderFactory.java index f96def8..465226a 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutorBuilderFactory.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutorBuilderFactory.java @@ -18,6 +18,11 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * Entry point for configuring and creating new executors. * @@ -29,6 +34,7 @@ package org.apache.cassandra.concurrent; * <li>{@link #configureSequential(String)} * <li>{@link #configurePooled(String, int)} */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface ExecutorBuilderFactory<E extends ExecutorPlus, S extends SequentialExecutorPlus> { /** diff --git a/src/java/org/apache/cassandra/concurrent/ExecutorFactory.java b/src/java/org/apache/cassandra/concurrent/ExecutorFactory.java index 52ba94a..f1acd55 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutorFactory.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutorFactory.java @@ -21,12 +21,15 @@ package org.apache.cassandra.concurrent; import java.util.function.Consumer; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.Shared; import static java.lang.Thread.*; import static org.apache.cassandra.concurrent.NamedThreadFactory.createThread; import static org.apache.cassandra.concurrent.NamedThreadFactory.setupThread; import static org.apache.cassandra.concurrent.ThreadPoolExecutorBuilder.pooledJmx; import static org.apache.cassandra.concurrent.ThreadPoolExecutorBuilder.sequentialJmx; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Entry point for configuring and creating new executors. @@ -52,6 +55,7 @@ import static org.apache.cassandra.concurrent.ThreadPoolExecutorBuilder.sequenti * Supports shared executors via sub-factory {@code localAware().withJMX()} * using {@link LocalAwareSubFactoryWithJMX#shared(String, int, ExecutorPlus.MaximumPoolSizeListener)} */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface ExecutorFactory extends ExecutorBuilderFactory.Jmxable<ExecutorPlus, SequentialExecutorPlus> { public interface LocalAwareSubFactoryWithJMX extends ExecutorBuilderFactory<LocalAwareExecutorPlus, LocalAwareSequentialExecutorPlus> diff --git a/src/java/org/apache/cassandra/concurrent/ExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/ExecutorPlus.java index c42a475..1fca66f 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutorPlus.java @@ -26,14 +26,19 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.WithResources; import org.apache.cassandra.utils.concurrent.Future; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * Cassandra's extension of {@link ExecutorService}, using our own {@link Future}, supporting * {@link #inExecutor()}, and execution with associated resources {@link #execute(WithResources, Runnable)} * (which is primarily used for encapsulating {@link ExecutorLocals} without leaking implementing classes). */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface ExecutorPlus extends ExecutorService, ResizableThreadPool { interface MaximumPoolSizeListener diff --git a/src/java/org/apache/cassandra/concurrent/Interruptible.java b/src/java/org/apache/cassandra/concurrent/Interruptible.java index cc13a63..8641ec8 100644 --- a/src/java/org/apache/cassandra/concurrent/Interruptible.java +++ b/src/java/org/apache/cassandra/concurrent/Interruptible.java @@ -18,8 +18,13 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + import static org.apache.cassandra.concurrent.Interruptible.State.*; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; +@Shared(scope = SIMULATION, inner = INTERFACES) public interface Interruptible extends Shutdownable { public enum State { NORMAL, INTERRUPTED, SHUTTING_DOWN } diff --git a/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorPlus.java index 7509619..743cacc 100644 --- a/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/LocalAwareExecutorPlus.java @@ -18,9 +18,14 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * An {@link ExecutorPlus} that is aware of, and propagates to execution, any ExecutorLocals */ +@Shared(scope = SIMULATION) public interface LocalAwareExecutorPlus extends ExecutorPlus { } diff --git a/src/java/org/apache/cassandra/concurrent/LocalAwareSequentialExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/LocalAwareSequentialExecutorPlus.java index dbcff6c..99e44b0 100644 --- a/src/java/org/apache/cassandra/concurrent/LocalAwareSequentialExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/LocalAwareSequentialExecutorPlus.java @@ -18,9 +18,14 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * A {@link SequentialExecutorPlus} that is aware of, and propagates to execution, any ExecutorLocals */ +@Shared(scope = SIMULATION) public interface LocalAwareSequentialExecutorPlus extends LocalAwareExecutorPlus, SequentialExecutorPlus { } diff --git a/src/java/org/apache/cassandra/concurrent/ResizableThreadPool.java b/src/java/org/apache/cassandra/concurrent/ResizableThreadPool.java index 760c06e..9c1dba6 100644 --- a/src/java/org/apache/cassandra/concurrent/ResizableThreadPool.java +++ b/src/java/org/apache/cassandra/concurrent/ResizableThreadPool.java @@ -18,6 +18,11 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface ResizableThreadPool { /** diff --git a/src/java/org/apache/cassandra/concurrent/ScheduledExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/ScheduledExecutorPlus.java index 0b512ac..ecf073d 100644 --- a/src/java/org/apache/cassandra/concurrent/ScheduledExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/ScheduledExecutorPlus.java @@ -20,6 +20,11 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.ScheduledExecutorService; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface ScheduledExecutorPlus extends ExecutorPlus, ScheduledExecutorService { } diff --git a/src/java/org/apache/cassandra/concurrent/SequentialExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/SequentialExecutorPlus.java index 7ea0e95..2b63f14 100644 --- a/src/java/org/apache/cassandra/concurrent/SequentialExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/SequentialExecutorPlus.java @@ -18,12 +18,18 @@ package org.apache.cassandra.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * An {@link ExecutorPlus} that guarantees the order of execution matches the order of task submission, * and provides a simple mechanism for the recurring pattern of ensuring a job is executed at least once * after some point in time (i.e. ensures that at most one copy of the task is queued, with up to one * copy running as well) */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface SequentialExecutorPlus extends ExecutorPlus { public interface AtLeastOnceTrigger diff --git a/src/java/org/apache/cassandra/concurrent/Shutdownable.java b/src/java/org/apache/cassandra/concurrent/Shutdownable.java index db89217..185875b 100644 --- a/src/java/org/apache/cassandra/concurrent/Shutdownable.java +++ b/src/java/org/apache/cassandra/concurrent/Shutdownable.java @@ -20,6 +20,11 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.TimeUnit; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface Shutdownable { boolean isTerminated(); diff --git a/src/java/org/apache/cassandra/concurrent/TaskFactory.java b/src/java/org/apache/cassandra/concurrent/TaskFactory.java index a25a45f..56087d9 100644 --- a/src/java/org/apache/cassandra/concurrent/TaskFactory.java +++ b/src/java/org/apache/cassandra/concurrent/TaskFactory.java @@ -20,10 +20,13 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.Callable; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.WithResources; import org.apache.cassandra.utils.concurrent.RunnableFuture; import static org.apache.cassandra.concurrent.FutureTask.callable; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * A simple mechanism to impose our desired semantics on the execution of a task without requiring a specialised @@ -32,6 +35,7 @@ import static org.apache.cassandra.concurrent.FutureTask.callable; * The encapsulations handle any exceptions in our standard way, as well as ensuring {@link ExecutorLocals} are * propagated in the case of {@link #localAware()} */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface TaskFactory { Runnable toExecute(Runnable runnable); diff --git a/src/java/org/apache/cassandra/config/ParameterizedClass.java b/src/java/org/apache/cassandra/config/ParameterizedClass.java index 4b8cf5a..9b00178 100644 --- a/src/java/org/apache/cassandra/config/ParameterizedClass.java +++ b/src/java/org/apache/cassandra/config/ParameterizedClass.java @@ -22,6 +22,11 @@ import java.util.Map; import com.google.common.base.Objects; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public class ParameterizedClass { public static final String CLASS_NAME = "class_name"; diff --git a/src/java/org/apache/cassandra/exceptions/CassandraException.java b/src/java/org/apache/cassandra/exceptions/CassandraException.java index 58521df..119daac 100644 --- a/src/java/org/apache/cassandra/exceptions/CassandraException.java +++ b/src/java/org/apache/cassandra/exceptions/CassandraException.java @@ -17,6 +17,11 @@ */ package org.apache.cassandra.exceptions; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public abstract class CassandraException extends RuntimeException implements TransportException { private final ExceptionCode code; diff --git a/src/java/org/apache/cassandra/exceptions/ExceptionCode.java b/src/java/org/apache/cassandra/exceptions/ExceptionCode.java index 1766951..8bb0cfd 100644 --- a/src/java/org/apache/cassandra/exceptions/ExceptionCode.java +++ b/src/java/org/apache/cassandra/exceptions/ExceptionCode.java @@ -21,10 +21,14 @@ import java.util.HashMap; import java.util.Map; import org.apache.cassandra.transport.ProtocolException; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Exceptions code, as defined by the binary protocol. */ +@Shared(scope = SIMULATION) public enum ExceptionCode { SERVER_ERROR (0x0000), diff --git a/src/java/org/apache/cassandra/exceptions/RequestExecutionException.java b/src/java/org/apache/cassandra/exceptions/RequestExecutionException.java index 4db108a..d80559a 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestExecutionException.java +++ b/src/java/org/apache/cassandra/exceptions/RequestExecutionException.java @@ -17,6 +17,11 @@ */ package org.apache.cassandra.exceptions; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public abstract class RequestExecutionException extends CassandraException { protected RequestExecutionException(ExceptionCode code, String msg) diff --git a/src/java/org/apache/cassandra/exceptions/TransportException.java b/src/java/org/apache/cassandra/exceptions/TransportException.java index 70d1da5..9807749 100644 --- a/src/java/org/apache/cassandra/exceptions/TransportException.java +++ b/src/java/org/apache/cassandra/exceptions/TransportException.java @@ -17,6 +17,11 @@ */ package org.apache.cassandra.exceptions; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface TransportException { /** diff --git a/src/java/org/apache/cassandra/io/util/DataInputPlus.java b/src/java/org/apache/cassandra/io/util/DataInputPlus.java index 41b422a..bda8461 100644 --- a/src/java/org/apache/cassandra/io/util/DataInputPlus.java +++ b/src/java/org/apache/cassandra/io/util/DataInputPlus.java @@ -19,11 +19,15 @@ package org.apache.cassandra.io.util; import java.io.*; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.vint.VIntCoding; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * Extension to DataInput that provides support for reading varints */ +@Shared(scope = SIMULATION) public interface DataInputPlus extends DataInput { default long readVInt() throws IOException diff --git a/src/java/org/apache/cassandra/io/util/DataOutputPlus.java b/src/java/org/apache/cassandra/io/util/DataOutputPlus.java index b94d097..a8f545e 100644 --- a/src/java/org/apache/cassandra/io/util/DataOutputPlus.java +++ b/src/java/org/apache/cassandra/io/util/DataOutputPlus.java @@ -21,18 +21,22 @@ import java.io.DataOutput; import java.io.IOException; import java.nio.ByteBuffer; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.vint.VIntCoding; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * Extension to DataOutput that provides for writing ByteBuffer and Memory, potentially with an efficient * implementation that is zero copy or at least has reduced bounds checking overhead. */ +@Shared(scope = SIMULATION) public interface DataOutputPlus extends DataOutput { // write the buffer without modifying its position void write(ByteBuffer buffer) throws IOException; - default void write(Memory memory, long offset, long length) throws IOException + default void write(ReadableMemory memory, long offset, long length) throws IOException { for (ByteBuffer buffer : memory.asByteBuffers(offset, length)) write(buffer); diff --git a/src/java/org/apache/cassandra/io/util/FileReader.java b/src/java/org/apache/cassandra/io/util/FileReader.java index 55b8fbb..86d0388 100644 --- a/src/java/org/apache/cassandra/io/util/FileReader.java +++ b/src/java/org/apache/cassandra/io/util/FileReader.java @@ -20,7 +20,6 @@ package org.apache.cassandra.io.util; import java.io.IOException; import java.io.InputStreamReader; -import java.io.OutputStreamWriter; public class FileReader extends InputStreamReader { diff --git a/src/java/org/apache/cassandra/io/util/Memory.java b/src/java/org/apache/cassandra/io/util/Memory.java index eaa6e91..c55f047 100644 --- a/src/java/org/apache/cassandra/io/util/Memory.java +++ b/src/java/org/apache/cassandra/io/util/Memory.java @@ -32,7 +32,7 @@ import sun.misc.Unsafe; /** * An off-heap region of memory that must be manually free'd when no longer needed. */ -public class Memory implements AutoCloseable +public class Memory implements AutoCloseable, ReadableMemory { private static final Unsafe unsafe; static diff --git a/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java b/src/java/org/apache/cassandra/io/util/ReadableMemory.java similarity index 73% copy from src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java copy to src/java/org/apache/cassandra/io/util/ReadableMemory.java index f3e6111..ccb717d 100644 --- a/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java +++ b/src/java/org/apache/cassandra/io/util/ReadableMemory.java @@ -16,14 +16,16 @@ * limitations under the License. */ -package org.apache.cassandra.streaming; +package org.apache.cassandra.io.util; -import java.io.Closeable; +import java.nio.ByteBuffer; -import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.utils.Shared; -public interface StreamingDataInputPlus extends DataInputPlus, Closeable +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) +public interface ReadableMemory { - @Override - void close(); + ByteBuffer[] asByteBuffers(long offset, long length); } diff --git a/src/java/org/apache/cassandra/streaming/StreamingChannel.java b/src/java/org/apache/cassandra/streaming/StreamingChannel.java index af6e68e..18bb2b7 100644 --- a/src/java/org/apache/cassandra/streaming/StreamingChannel.java +++ b/src/java/org/apache/cassandra/streaming/StreamingChannel.java @@ -24,7 +24,12 @@ import java.util.function.IntFunction; import io.netty.util.concurrent.Future; import org.apache.cassandra.streaming.async.NettyStreamingConnectionFactory; +import org.apache.cassandra.utils.Shared; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION, inner = INTERFACES) public interface StreamingChannel { public interface Factory diff --git a/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java b/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java index f3e6111..0cfcc0d 100644 --- a/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java +++ b/src/java/org/apache/cassandra/streaming/StreamingDataInputPlus.java @@ -21,7 +21,11 @@ package org.apache.cassandra.streaming; import java.io.Closeable; import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.utils.Shared; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface StreamingDataInputPlus extends DataInputPlus, Closeable { @Override diff --git a/src/java/org/apache/cassandra/streaming/StreamingDataOutputPlus.java b/src/java/org/apache/cassandra/streaming/StreamingDataOutputPlus.java index d4a514b..3f68b3a 100644 --- a/src/java/org/apache/cassandra/streaming/StreamingDataOutputPlus.java +++ b/src/java/org/apache/cassandra/streaming/StreamingDataOutputPlus.java @@ -25,7 +25,12 @@ import java.nio.channels.FileChannel; import io.netty.channel.FileRegion; import org.apache.cassandra.io.util.DataOutputPlus; +import org.apache.cassandra.utils.Shared; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION, inner = INTERFACES) public interface StreamingDataOutputPlus extends DataOutputPlus, Closeable { interface BufferSupplier diff --git a/src/java/org/apache/cassandra/utils/Clock.java b/src/java/org/apache/cassandra/utils/Clock.java index 6fd0efd..1f39493 100644 --- a/src/java/org/apache/cassandra/utils/Clock.java +++ b/src/java/org/apache/cassandra/utils/Clock.java @@ -23,6 +23,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOBAL; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Wrapper around time related functions that are either implemented by using the default JVM calls @@ -33,6 +34,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOB * Please note that {@link java.time.Clock} wasn't used, as it would not be possible to provide an * implementation for {@link #nanoTime()} with the exact same properties of {@link System#nanoTime()}. */ +@Shared(scope = SIMULATION) public interface Clock { static final Logger logger = LoggerFactory.getLogger(Clock.class); diff --git a/src/java/org/apache/cassandra/utils/Closeable.java b/src/java/org/apache/cassandra/utils/Closeable.java index ccc33ea..1a8c1e9 100644 --- a/src/java/org/apache/cassandra/utils/Closeable.java +++ b/src/java/org/apache/cassandra/utils/Closeable.java @@ -18,6 +18,9 @@ package org.apache.cassandra.utils; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface Closeable extends java.io.Closeable { public void close(); diff --git a/test/distributed/org/apache/cassandra/distributed/shared/Isolated.java b/src/java/org/apache/cassandra/utils/Isolated.java similarity index 95% rename from test/distributed/org/apache/cassandra/distributed/shared/Isolated.java rename to src/java/org/apache/cassandra/utils/Isolated.java index 898631f..12f88e3 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/Isolated.java +++ b/src/java/org/apache/cassandra/utils/Isolated.java @@ -16,13 +16,15 @@ * limitations under the License. */ -package org.apache.cassandra.distributed.shared; +package org.apache.cassandra.utils; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import org.apache.cassandra.utils.Shared; + /** * Tells jvm-dtest that a class should be isolated and loaded into the instance class loader. * diff --git a/src/java/org/apache/cassandra/utils/MonotonicClock.java b/src/java/org/apache/cassandra/utils/MonotonicClock.java index 2ef563c..e14fd45 100644 --- a/src/java/org/apache/cassandra/utils/MonotonicClock.java +++ b/src/java/org/apache/cassandra/utils/MonotonicClock.java @@ -33,6 +33,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_APPROX; import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_PRECISE; import static org.apache.cassandra.utils.Clock.Global.nanoTime; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Wrapper around time related functions that are either implemented by using the default JVM calls @@ -45,6 +46,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; * * TODO better rationalise MonotonicClock/Clock */ +@Shared(scope = SIMULATION) public interface MonotonicClock { /** diff --git a/src/java/org/apache/cassandra/utils/MonotonicClockTranslation.java b/src/java/org/apache/cassandra/utils/MonotonicClockTranslation.java index f7f83e4..cef8bd8 100644 --- a/src/java/org/apache/cassandra/utils/MonotonicClockTranslation.java +++ b/src/java/org/apache/cassandra/utils/MonotonicClockTranslation.java @@ -18,6 +18,9 @@ package org.apache.cassandra.utils; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface MonotonicClockTranslation { /** accepts millis since epoch, returns nanoTime in the related clock */ diff --git a/test/distributed/org/apache/cassandra/distributed/shared/Shared.java b/src/java/org/apache/cassandra/utils/Shared.java similarity index 82% rename from test/distributed/org/apache/cassandra/distributed/shared/Shared.java rename to src/java/org/apache/cassandra/utils/Shared.java index bb67070..e576c86 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/Shared.java +++ b/src/java/org/apache/cassandra/utils/Shared.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.cassandra.distributed.shared; +package org.apache.cassandra.utils; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; @@ -36,4 +36,10 @@ import java.lang.annotation.Target; @Target({ ElementType.TYPE }) public @interface Shared { + enum Scope { ANY, SIMULATION } + enum Recursive { NONE, INTERFACES /*(and enums and exceptions) */, ALL } + Scope[] scope() default Scope.ANY; + Recursive inner() default Recursive.NONE; + Recursive ancestors() default Recursive.NONE; + Recursive members() default Recursive.NONE; } diff --git a/src/java/org/apache/cassandra/utils/WithResources.java b/src/java/org/apache/cassandra/utils/WithResources.java index 76e218c..0c0bb92 100644 --- a/src/java/org/apache/cassandra/utils/WithResources.java +++ b/src/java/org/apache/cassandra/utils/WithResources.java @@ -20,12 +20,15 @@ package org.apache.cassandra.utils; import org.apache.cassandra.concurrent.ExecutorPlus; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * A generic interface for encapsulating a Runnable task with related work before and after execution, * using the built-in try-with-resources functionality offered by {@link Closeable}. * * See {@link ExecutorPlus#execute(WithResources, Runnable)} */ +@Shared(scope = SIMULATION) public interface WithResources { static class None implements WithResources diff --git a/src/java/org/apache/cassandra/utils/concurrent/Awaitable.java b/src/java/org/apache/cassandra/utils/concurrent/Awaitable.java index 5b3d315..03aab5f 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Awaitable.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Awaitable.java @@ -23,17 +23,20 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import java.util.function.Predicate; import net.nicoulaj.compilecommand.annotations.Inline; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.Intercept; import static org.apache.cassandra.utils.Clock.Global.nanoTime; import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * A generic signal consumer, supporting all of the typical patterns used in Cassandra. * All of the methods defined in {@link Awaitable} may be waited on without a loop, * as this interface declares that there are no spurious wake-ups. */ +@Shared(scope = SIMULATION) public interface Awaitable { /** diff --git a/src/java/org/apache/cassandra/utils/concurrent/Condition.java b/src/java/org/apache/cassandra/utils/concurrent/Condition.java index f47e20f..eb97848 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Condition.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Condition.java @@ -19,6 +19,9 @@ package org.apache.cassandra.utils.concurrent; import org.apache.cassandra.utils.Intercept; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * Simpler API than java.util.concurrent.Condition; would be nice to extend it, but also nice @@ -26,6 +29,7 @@ import org.apache.cassandra.utils.Intercept; * * {@link Awaitable} for explicit external signals. */ +@Shared(scope = SIMULATION) public interface Condition extends Awaitable { /** diff --git a/src/java/org/apache/cassandra/utils/concurrent/CountDownLatch.java b/src/java/org/apache/cassandra/utils/concurrent/CountDownLatch.java index 5988375..976abb7 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/CountDownLatch.java +++ b/src/java/org/apache/cassandra/utils/concurrent/CountDownLatch.java @@ -21,7 +21,11 @@ package org.apache.cassandra.utils.concurrent; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import org.apache.cassandra.utils.Intercept; +import org.apache.cassandra.utils.Shared; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface CountDownLatch extends Awaitable { /** diff --git a/src/java/org/apache/cassandra/utils/concurrent/Future.java b/src/java/org/apache/cassandra/utils/concurrent/Future.java index e21a654..4454a2e 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Future.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Future.java @@ -30,13 +30,17 @@ import com.google.common.util.concurrent.ListenableFuture; import io.netty.util.concurrent.GenericFutureListener; import io.netty.util.internal.PlatformDependent; +import org.apache.cassandra.utils.Shared; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * A Future that integrates several different (but equivalent) APIs used within Cassandra into a single concept, * integrating also with our {@link Awaitable} abstraction, to overall improve coherency and clarity in the codebase. */ +@Shared(scope = SIMULATION, ancestors = INTERFACES) public interface Future<V> extends io.netty.util.concurrent.Future<V>, ListenableFuture<V>, Awaitable { /** diff --git a/src/java/org/apache/cassandra/utils/concurrent/Promise.java b/src/java/org/apache/cassandra/utils/concurrent/Promise.java index 10d6275..d9e4623 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Promise.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Promise.java @@ -25,11 +25,16 @@ import java.util.function.Consumer; import com.google.common.util.concurrent.FutureCallback; import io.netty.util.concurrent.GenericFutureListener; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; /** * A Promise that integrates {@link io.netty.util.concurrent.Promise} with our {@link Future} API * to improve clarity and coherence in the codebase. */ +@Shared(scope = SIMULATION, ancestors = INTERFACES) public interface Promise<V> extends io.netty.util.concurrent.Promise<V>, Future<V> { public static <V> GenericFutureListener<? extends Future<V>> listener(FutureCallback<V> callback) diff --git a/src/java/org/apache/cassandra/utils/concurrent/RunnableFuture.java b/src/java/org/apache/cassandra/utils/concurrent/RunnableFuture.java index 74d8678..6ec1daa4 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/RunnableFuture.java +++ b/src/java/org/apache/cassandra/utils/concurrent/RunnableFuture.java @@ -18,6 +18,11 @@ package org.apache.cassandra.utils.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + +@Shared(scope = SIMULATION) public interface RunnableFuture<V> extends Future<V>, java.util.concurrent.RunnableFuture<V> { } diff --git a/src/java/org/apache/cassandra/utils/concurrent/Semaphore.java b/src/java/org/apache/cassandra/utils/concurrent/Semaphore.java index f16ffdd..66dd543 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Semaphore.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Semaphore.java @@ -24,10 +24,13 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import net.openhft.chronicle.core.util.ThrowingConsumer; import org.apache.cassandra.utils.Intercept; +import org.apache.cassandra.utils.Shared; import static java.lang.System.nanoTime; import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; +@Shared(scope = SIMULATION) public interface Semaphore { /** diff --git a/src/java/org/apache/cassandra/utils/concurrent/UncheckedInterruptedException.java b/src/java/org/apache/cassandra/utils/concurrent/UncheckedInterruptedException.java index 8e85b84..d7248e8 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/UncheckedInterruptedException.java +++ b/src/java/org/apache/cassandra/utils/concurrent/UncheckedInterruptedException.java @@ -18,9 +18,14 @@ package org.apache.cassandra.utils.concurrent; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * Unchecked {@link InterruptedException}, to be thrown in places where an interrupt is unexpected */ +@Shared(scope = SIMULATION) public class UncheckedInterruptedException extends RuntimeException { public UncheckedInterruptedException() diff --git a/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java b/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java index 4fbe7c6..e9dcdf8 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java +++ b/src/java/org/apache/cassandra/utils/concurrent/WaitQueue.java @@ -26,9 +26,13 @@ import java.util.function.BooleanSupplier; import java.util.function.Consumer; import org.apache.cassandra.utils.Intercept; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.utils.concurrent.Awaitable.AbstractAwaitable; import static org.apache.cassandra.utils.Clock.Global.nanoTime; +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; +import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; + /** * <p>A relatively easy to use utility for general purpose thread signalling.</p> * <p>Usage on a thread awaiting a state change using a WaitQueue q is:</p> @@ -73,6 +77,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; * * TODO: this class should not be backed by CLQ (should use an intrusive linked-list with lower overhead) */ +@Shared(scope = SIMULATION, inner = INTERFACES) public interface WaitQueue { /** diff --git a/test/distributed/org/apache/cassandra/distributed/Cluster.java b/test/distributed/org/apache/cassandra/distributed/Cluster.java index 05ea799..116dcd6 100644 --- a/test/distributed/org/apache/cassandra/distributed/Cluster.java +++ b/test/distributed/org/apache/cassandra/distributed/Cluster.java @@ -24,14 +24,13 @@ import java.util.function.Consumer; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.impl.AbstractCluster; import org.apache.cassandra.distributed.api.IInvokableInstance; -import org.apache.cassandra.distributed.shared.Shared; import org.apache.cassandra.distributed.shared.Versions; +import org.apache.cassandra.utils.Shared; /** * A simple cluster supporting only the 'current' Cassandra version, offering easy access to the convenience methods * of IInvokableInstance on each node. */ -@Shared public class Cluster extends AbstractCluster<IInvokableInstance> { private Cluster(Builder builder) diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index 2f146bf..be49daa 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@ -19,6 +19,9 @@ package org.apache.cassandra.distributed.impl; import java.lang.annotation.Annotation; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; import java.net.InetSocketAddress; import java.nio.file.FileSystem; import java.nio.file.Files; @@ -27,12 +30,14 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executor; import java.util.concurrent.Future; @@ -40,6 +45,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiPredicate; import java.util.function.Consumer; +import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -47,6 +53,7 @@ import java.util.stream.Stream; import javax.annotation.concurrent.GuardedBy; +import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,22 +82,29 @@ import org.apache.cassandra.distributed.api.LogAction; import org.apache.cassandra.distributed.api.NodeToolResult; import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.shared.InstanceClassLoader; -import org.apache.cassandra.distributed.shared.Isolated; +import org.apache.cassandra.utils.Isolated; import org.apache.cassandra.distributed.shared.MessageFilters; import org.apache.cassandra.distributed.shared.Metrics; import org.apache.cassandra.distributed.shared.NetworkTopology; -import org.apache.cassandra.distributed.shared.Shared; +import org.apache.cassandra.utils.Shared; import org.apache.cassandra.distributed.shared.ShutdownException; import org.apache.cassandra.distributed.shared.Versions; import org.apache.cassandra.io.util.PathUtils; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.utils.Shared.Recursive; import org.apache.cassandra.utils.concurrent.Condition; import org.apache.cassandra.utils.FBUtilities; import org.reflections.Reflections; +import org.reflections.scanners.TypeAnnotationsScanner; import org.reflections.util.ConfigurationBuilder; +import static java.util.stream.Stream.of; import static org.apache.cassandra.distributed.shared.NetworkTopology.addressAndPort; +import static org.apache.cassandra.utils.Shared.Recursive.ALL; +import static org.apache.cassandra.utils.Shared.Recursive.NONE; +import static org.apache.cassandra.utils.Shared.Scope.ANY; import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition; +import static org.reflections.ReflectionUtils.forNames; /** * AbstractCluster creates, initializes and manages Cassandra instances ({@link Instance}. @@ -116,7 +130,6 @@ import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeConditio * handlers for internode to have more control over it. Messaging is wired by passing verbs manually. * coordinator-handling code and hooks to the callbacks can be found in {@link Coordinator}. */ -@Shared public abstract class AbstractCluster<I extends IInstance> implements ICluster<I>, AutoCloseable { public static Versions.Version CURRENT_VERSION = new Versions.Version(FBUtilities.getReleaseVersionString(), Versions.getClassPath()); @@ -128,16 +141,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I private static final AtomicInteger GENERATION = new AtomicInteger(); // include byteman so tests can use - private static final Set<String> SHARED_CLASSES = findClassesMarkedForSharedClassLoader(); - private static final Set<String> ISOLATED_CLASSES = findClassesMarkedForInstanceClassLoader(); - public static final Predicate<String> SHARED_PREDICATE = s -> { - if (ISOLATED_CLASSES.contains(s)) - return false; - - return SHARED_CLASSES.contains(s) || - InstanceClassLoader.getDefaultLoadSharedFilter().test(s) || - s.startsWith("org.jboss.byteman"); - }; + public static final Predicate<String> SHARED_PREDICATE = getSharedClassPredicate(ANY); private final UUID clusterId = UUID.randomUUID(); private final Path root; @@ -996,22 +1000,213 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I .collect(Collectors.toList()); } - private static Set<String> findClassesMarkedForSharedClassLoader() + private static Set<String> findClassesMarkedForSharedClassLoader(Class<?>[] share, Shared.Scope ... scopes) + { + return findClassesMarkedForSharedClassLoader(share, ImmutableSet.copyOf(scopes)::contains); + } + + private static Set<String> findClassesMarkedForSharedClassLoader(Class<?>[] share, Predicate<Shared.Scope> scopes) { - return findClassesMarkedWith(Shared.class); + Set<Class<?>> classes = findClassesMarkedWith(Shared.class, a -> of(a.scope()).anyMatch(scopes)); + Collections.addAll(classes, share); + assertTransitiveClosure(classes); + return toNames(classes); } - private static Set<String> findClassesMarkedForInstanceClassLoader() + private static Set<String> findClassesMarkedForInstanceClassLoader(Class<?>[] isolate) { - return findClassesMarkedWith(Isolated.class); + Set<Class<?>> classes = findClassesMarkedWith(Isolated.class, ignore -> true); + Collections.addAll(classes, isolate); + return toNames(classes); } - private static Set<String> findClassesMarkedWith(Class<? extends Annotation> annotation) + public static Predicate<String> getSharedClassPredicate(Shared.Scope ... scopes) { - return new Reflections(ConfigurationBuilder.build("org.apache.cassandra").setExpandSuperTypes(false)) - .getTypesAnnotatedWith(annotation).stream() - .map(Class::getName) + return getSharedClassPredicate(new Class[0], new Class[0], scopes); + } + + public static Predicate<String> getSharedClassPredicate(Class<?>[] isolate, Class<?>[] share, Shared.Scope ... scopes) + { + Set<String> shared = findClassesMarkedForSharedClassLoader(share, scopes); + Set<String> isolated = findClassesMarkedForInstanceClassLoader(isolate); + return s -> { + if (isolated.contains(s)) + return false; + + return shared.contains(s) || + InstanceClassLoader.getDefaultLoadSharedFilter().test(s) || + s.startsWith("org.jboss.byteman"); + }; + } + + private static <A extends Annotation> Set<Class<?>> findClassesMarkedWith(Class<A> annotation, Predicate<A> testAnnotation) + { + Reflections reflections = new Reflections(ConfigurationBuilder.build("org.apache.cassandra").setExpandSuperTypes(false)); + return forNames(reflections.getStore().getAll(TypeAnnotationsScanner.class, annotation.getName()), + reflections.getConfiguration().getClassLoaders()) + .stream() + .filter(testAnnotation(annotation, testAnnotation)) + .flatMap(expander()) .collect(Collectors.toSet()); } + + private static Set<String> toNames(Set<Class<?>> classes) + { + return classes.stream().map(Class::getName).collect(Collectors.toSet()); + } + + private static <A extends Annotation> Predicate<Class<?>> testAnnotation(Class<A> annotation, Predicate<A> test) + { + return clazz -> { + A[] annotations = clazz.getDeclaredAnnotationsByType(annotation); + for (A a : annotations) + { + if (!test.test(a)) + return false; + } + return true; + }; + } + + private static void assertTransitiveClosure(Set<Class<?>> classes) + { + Set<Class<?>> tested = new HashSet<>(); + for (Class<?> clazz : classes) + { + forEach(test -> { + if (!classes.contains(test)) + throw new AssertionError(clazz.getName() + " is shared, but its dependency " + test + " is not"); + }, new SharedParams(ALL, ALL, NONE), clazz, tested); + } + } + + private static class SharedParams + { + final Recursive ancestors, members, inner; + + private SharedParams(Recursive ancestors, Recursive members, Recursive inner) + { + this.ancestors = ancestors; + this.members = members; + this.inner = inner; + } + + private SharedParams(Shared shared) + { + this.ancestors = shared.ancestors(); + this.members = shared.members(); + this.inner = shared.inner(); + } + } + + private static void forEach(Consumer<Class<?>> forEach, SharedParams shared, Class<?> cur, Set<Class<?>> done) + { + if (null == (cur = consider(cur, done))) + return; + + forEach.accept(cur); + + switch (shared.ancestors) + { + case ALL: + forEach(forEach, shared, cur.getSuperclass(), done); + case INTERFACES: + for (Class<?> i : cur.getInterfaces()) + forEach(forEach, shared, i, done); + } + + if (shared.members != NONE) + { + for (Field field : cur.getDeclaredFields()) + { + if ((field.getModifiers() & Modifier.PRIVATE) == 0) + forEachMatch(shared.members, forEach, shared, field.getType(), done); + } + + for (Method method : cur.getDeclaredMethods()) + { + if ((method.getModifiers() & Modifier.PRIVATE) == 0) + { + forEachMatch(shared.members, forEach, shared, method.getReturnType(), done); + forEachMatch(shared.members, forEach, shared, method.getParameterTypes(), done); + } + } + } + + if (shared.inner != NONE) + forEachMatch(shared.inner, forEach, shared, cur.getDeclaredClasses(), done); + } + + private static void forEachMatch(Recursive ifMatches, Consumer<Class<?>> forEach, SharedParams shared, Class<?>[] classes, Set<Class<?>> done) + { + for (Class<?> cur : classes) + forEachMatch(ifMatches, forEach, shared, cur, done); + } + + private static void forEachMatch(Recursive ifMatches, Consumer<Class<?>> forEach, SharedParams shared, Class<?> cur, Set<Class<?>> done) + { + if (ifMatches == ALL || isInterface(cur)) + forEach(forEach, shared, cur, done); + } + + private static boolean isInterface(Class<?> test) + { + return test.isInterface() || test.isEnum() || Throwable.class.isAssignableFrom(test); + } + + private static Function<Class<?>, Stream<Class<?>>> expander() + { + Set<Class<?>> done = new HashSet<>(); + return clazz -> expand(clazz, done); + } + + private static Stream<Class<?>> expand(Class<?> clazz, Set<Class<?>> done) + { + Optional<Shared> maybeShared = of(clazz.getDeclaredAnnotationsByType(Shared.class)).findFirst(); + if (!maybeShared.isPresent()) + return Stream.of(clazz); + + Shared shared = maybeShared.get(); + if (shared.inner() == NONE && shared.members() == NONE && shared.ancestors() == NONE) + return Stream.of(clazz); + + Set<Class<?>> closure = new HashSet<>(); + forEach(closure::add, new SharedParams(shared), clazz, done); + return closure.stream(); + } + + private static Class<?> consider(Class<?> consider, Set<Class<?>> considered) + { + if (consider == null) return null; + while (consider.isArray()) // TODO (future): this is inadequate handling of array types (fine for now) + consider = consider.getComponentType(); + + if (consider.isPrimitive()) return null; + if (consider.getPackage() != null && consider.getPackage().getName().startsWith("java.")) return null; + if (!considered.add(consider)) return null; + if (InstanceClassLoader.getDefaultLoadSharedFilter().test(consider.getName())) return null; + + return consider; + } + + // 3.0 and earlier clusters must have unique InetAddressAndPort for each InetAddress + public static <I extends IInstance> Map<InetSocketAddress, I> getUniqueAddressLookup(ICluster<I> cluster) + { + return getUniqueAddressLookup(cluster, Function.identity()); + } + + public static <I extends IInstance, V> Map<InetSocketAddress, V> getUniqueAddressLookup(ICluster<I> cluster, Function<I, V> function) + { + Map<InetSocketAddress, V> lookup = new HashMap<>(); + cluster.stream().forEach(instance -> { + InetSocketAddress address = instance.broadcastAddress(); + if (!address.equals(instance.config().broadcastAddress())) + throw new IllegalStateException("addressAndPort mismatch: " + address + " vs " + instance.config().broadcastAddress()); + V prev = lookup.put(address, function.apply(instance)); + if (null != prev) + throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress: " + address + " vs " + prev); + }); + return lookup; + } } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java index 9fdebed..b409c88 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java @@ -210,7 +210,7 @@ public class Coordinator implements ICoordinator }).call(); } - static ClientState makeFakeClientState() + public static ClientState makeFakeClientState() { return ClientState.forExternalCalls(new InetSocketAddress(FBUtilities.getJustLocalAddress(), 9042)); } diff --git a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java index 32f82c0..1ec844e 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/INodeProvisionStrategy.java @@ -18,9 +18,13 @@ package org.apache.cassandra.distributed.impl; +import org.apache.cassandra.utils.Shared; + +import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES; + +@Shared(inner = INTERFACES) public interface INodeProvisionStrategy { - public enum Strategy { OneNetworkInterface diff --git a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 9978697..c7ae775 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@ -33,12 +33,10 @@ import com.vdurmont.semver4j.Semver; import org.apache.cassandra.distributed.api.Feature; import org.apache.cassandra.distributed.api.IInstanceConfig; import org.apache.cassandra.distributed.shared.NetworkTopology; -import org.apache.cassandra.distributed.shared.Shared; import org.apache.cassandra.distributed.upgrade.UpgradeTestBase; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.SimpleSeedProvider; -@Shared public class InstanceConfig implements IInstanceConfig { public final int num; diff --git a/test/distributed/org/apache/cassandra/distributed/impl/Query.java b/test/distributed/org/apache/cassandra/distributed/impl/Query.java index 1b03996..7950d75 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/Query.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Query.java @@ -35,11 +35,10 @@ import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; -import static org.apache.cassandra.distributed.impl.Coordinator.toCassandraCL; - -// TODO: maybe just keep with Simulator? public class Query implements IIsolatedExecutor.SerializableCallable<Object[][]> { + private static final long serialVersionUID = 1L; + final String query; final long timestamp; final org.apache.cassandra.distributed.api.ConsistencyLevel commitConsistencyOrigin; diff --git a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java index 1821f9c..35d6323 100644 --- a/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java +++ b/test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java @@ -53,6 +53,7 @@ import org.apache.cassandra.distributed.impl.AbstractCluster; import org.apache.cassandra.distributed.impl.InstanceConfig; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Isolated; import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS; diff --git a/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java b/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java index a7ac605..55b6c86 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/BootstrapBinaryDisabledTest.java @@ -36,7 +36,7 @@ import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.shared.Byteman; import org.apache.cassandra.distributed.shared.NetworkTopology; -import org.apache.cassandra.distributed.shared.Shared; +import org.apache.cassandra.utils.Shared; /** * Replaces python dtest bootstrap_test.py::TestBootstrap::test_bootstrap_binary_disabled diff --git a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest.java b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest.java index 78d55e3..7d69248 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest.java @@ -24,7 +24,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; -import com.google.common.util.concurrent.Uninterruptibles; import org.junit.Assert; import org.junit.Test; @@ -35,14 +34,15 @@ import net.bytebuddy.implementation.bind.annotation.SuperCall; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.Constants; import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.ICluster; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.api.TokenSupplier; import org.apache.cassandra.distributed.impl.InstanceIDDefiner; import org.apache.cassandra.distributed.shared.ClusterUtils; -import org.apache.cassandra.distributed.shared.Shared; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.service.PendingRangeCalculatorService; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.Shared; import org.assertj.core.api.Assertions; import static net.bytebuddy.matcher.ElementMatchers.named; @@ -109,7 +109,7 @@ public class NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest extends Te @Shared public static class SharedState { - public static volatile Cluster cluster; + public static volatile ICluster cluster; // Instance.shutdown can only be called once so only the caller knows when its done (isShutdown looks at a field set BEFORE shutting down..) // since the test needs to know when shutdown completes, add this static state so the caller (bytebuddy rewrite) can update it public static final CountDownLatch shutdownComplete = new CountDownLatch(1); @@ -121,7 +121,7 @@ public class NodeCannotJoinAsHibernatingNodeWithoutReplaceAddressTest extends Te { fn.run(); int id = Integer.parseInt(InstanceIDDefiner.getInstanceId().replace("node", "")); - Cluster cluster = Objects.requireNonNull(SharedState.cluster); + ICluster cluster = Objects.requireNonNull(SharedState.cluster); // can't stop here as the stop method and start method share a lock; and block gets called in start... ForkJoinPool.commonPool().execute(() -> { ClusterUtils.stopAbrupt(cluster, cluster.get(id)); diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java index 78e2219..72b7f1a 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeMessageForwardTest.java @@ -28,11 +28,10 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.distributed.UpgradeableCluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.Feature; -import org.apache.cassandra.distributed.shared.Shared; +import org.apache.cassandra.utils.Shared; import static org.apache.cassandra.distributed.shared.AssertUtils.*; -@Shared public class MixedModeMessageForwardTest extends UpgradeTestBase { private static final Logger logger = LoggerFactory.getLogger(MixedModeMessageForwardTest.class); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
