Improve unit tests performance (CASSANDRA-17427 --> trunk)

build.xml

@@ -111,12 +111,18 @@
     <property name="test.distributed.timeout" value="900000" />
     <property name="test.simulation.timeout" value="1800000" />
 
-    <!-- default for cql tests. Can be override by -Dcassandra.test.use_prepared=false -->
+    <!-- default for cql tests. Can be overridden by -Dcassandra.test.use_prepared=false -->
     <property name="cassandra.test.use_prepared" value="true" />
 
+    <!-- The number of active processors seen by JVM -->
+    <property name="cassandra.test.processorCount" value="2"/>
+
     <!-- skip flushing schema tables during tests -->
     <property name="cassandra.test.flush_local_schema_changes" value="false" />
 
+    <!-- fast shutdown of messaging service -->
+    <property name="cassandra.test.messagingService.nonGracefulShutdown" value="true"/>
+
     <!-- https://www.eclemma.org/jacoco/ -->
     <property name="jacoco.version" value="0.8.8"/>
     <property name="jacoco.export.dir" value="${build.dir}/jacoco/" />
@@ -137,6 +143,10 @@
       <available file="${build.src.java}" type="dir" />
     </condition>
 
+    <condition property="cassandra.use_nix_recursive_delete" value="false" else="true">
+        <os family="windows" />
+    </condition>
+
     <!-- Check if all tests are being run or just one. If it's all tests don't spam the console with test output.
          If it's an individual test print the output from the test under the assumption someone is debugging the test
          and wants to know what is going on without having to context switch to the log file that is generated.
@@ -1110,6 +1120,7 @@
              more aggressively rather than waiting. See CASSANDRA-14922 for more details.
         -->
         <jvmarg value="-XX:SoftRefLRUPolicyMSPerMB=0" />
+        <jvmarg value="-XX:ActiveProcessorCount=${cassandra.test.processorCount}" />
         <jvmarg value="-XX:HeapDumpPath=build/test" />
         <jvmarg value="-Dcassandra.test.driver.connection_timeout_ms=${test.driver.connection_timeout_ms}"/>
         <jvmarg value="-Dcassandra.test.driver.read_timeout_ms=${test.driver.read_timeout_ms}"/>
@@ -1122,6 +1133,9 @@
         <jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" />
         <jvmarg value="-Dcassandra.strict.runtime.checks=true" />
         <jvmarg value="-Dcassandra.reads.thresholds.coordinator.defensive_checks_enabled=true" /> <!-- enable defensive checks -->
+        <jvmarg value="-Dcassandra.test.flush_local_schema_changes=${cassandra.test.flush_local_schema_changes}"/>
+        <jvmarg value="-Dcassandra.test.messagingService.nonGracefulShutdown=${cassandra.test.messagingService.nonGracefulShutdown}"/>
+        <jvmarg value="-Dcassandra.use_nix_recursive_delete=${cassandra.use_nix_recursive_delete}"/>
         <jvmarg line="${java11-jvmargs}"/>
         <!-- disable shrinks in quicktheories CASSANDRA-15554 -->
         <jvmarg value="-DQT_SHRINKS=0"/>
@@ -1797,6 +1811,7 @@
         <fileset dir="ide/idea"/>
     </copy>
     <replace file=".idea/workspace.xml" token="trunk" value="${eclipse.project.name}"/>
+    <replace file=".idea/workspace.xml" token="-Dcassandra.use_nix_recursive_delete=true" value="-Dcassandra.use_nix_recursive_delete=${cassandra.use_nix_recursive_delete}"/>
     <copy tofile="${eclipse.project.name}.iml" file="ide/idea-iml-file.xml"/>
     <echo file=".idea/.name">Apache Cassandra ${eclipse.project.name}</echo>
     <echo file=".idea/modules.xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>

ide/idea/workspace.xml

@@ -161,13 +161,19 @@
     <configuration default="true" type="JUnit" factoryName="JUnit">
       <extension name="coverage" enabled="false" merge="false" sample_coverage="true" runner="idea" />
       <module name="" />
+      <extension name="coverage">
+        <pattern>
+          <option name="PATTERN" value="org.apache.cassandra.*" />
+          <option name="ENABLED" value="true" />
+        </pattern>
+      </extension>
       <option name="ALTERNATIVE_JRE_PATH_ENABLED" value="false" />
       <option name="ALTERNATIVE_JRE_PATH" value="" />
       <option name="PACKAGE_NAME" />
       <option name="MAIN_CLASS_NAME" value="" />
       <option name="METHOD_NAME" value="" />
       <option name="TEST_OBJECT" value="class" />
-      <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -XX:HeapDumpPath=build/test -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=true -Dcassandra.reads.thresholds.coordinator.defensive_checks_enabled=true" />
+      <option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.skip_sync=true -ea -XX:MaxMetaspaceSize=1G -XX:SoftRefLRUPolicyMSPerMB=0 -XX:HeapDumpPath=build/test -XX:ActiveProcessorCount=2 -Dcassandra.strict.runtime.checks=true -Dlegacy-sstable-root=$PROJECT_DIR$/test/data/legacy-sstables -Dinvalid-legacy-sstable-root=$PROJECT_DIR$/test/data/invalid-legacy-sstables -Dmigration-sstable-root=$PROJECT_DIR$/test/data/migration-sstables -Dcassandra.ring_delay_ms=1000 -Dcassandra.tolerate_sstable_size=true -Dcassandra.skip_sync=true -Dcassandra.reads.thresholds.coordinator.defensive_checks_enabled=true -Dcassandra.use_nix_recursive_delete=true -Dcassandra.test.messagingService.nonGracefulShutdown=true -Dcassandra.test.flush_local_schema_changes=false " />
       <option name="PARAMETERS" value="" />
       <fork_mode value="class" />
       <option name="WORKING_DIRECTORY" value="" />

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -339,6 +339,23 @@ public enum CassandraRelevantProperties
      * The default used in SimpleSeedProvider is 20.
      */
     SEED_COUNT_WARN_THRESHOLD("cassandra.seed_count_warn_threshold"),
+
+
+    /** When enabled, recursive directory deletion will be executed using a unix command `rm -rf` instead of traversing
+     * and removing individual files. This is now used only tests, but eventually we will make it true by default.*/
+    USE_NIX_RECURSIVE_DELETE("cassandra.use_nix_recursive_delete"),
+
+    /** If set, {@link org.apache.cassandra.net.MessagingService} is shutdown abrtuptly without waiting for anything.
+     * This is an optimization used in unit tests becuase we never restart a node there. The only node is stopoped
+     * when the JVM terminates. Therefore, we can use such optimization and not wait unnecessarily. */
+    NON_GRACEFUL_SHUTDOWN("cassandra.test.messagingService.nonGracefulShutdown"),
+
+    /** Flush changes of {@link org.apache.cassandra.schema.SchemaKeyspace} after each schema modification. In production,
+     * we always do that. However, tests which do not restart nodes may disable this functionality in order to run
+     * faster. Note that this is disabled for unit tests but if an individual test requires schema to be flushed, it
+     * can be also done manually for that particular case: {@code flush(SchemaConstants.SCHEMA_KEYSPACE_NAME);}. */
+    FLUSH_LOCAL_SCHEMA_CHANGES("cassandra.test.flush_local_schema_changes", "true"),
+
     ;
 
     CassandraRelevantProperties(String key, String defaultVal)
@@ -382,6 +399,17 @@ public String getDefaultValue()
         return defaultVal;
     }
 
+    /**
+     * Sets the property to its default value if a default value was specified. Remove the property otherwise.
+     */
+    public void reset()
+    {
+        if (defaultVal != null)
+            System.setProperty(key, defaultVal);
+        else
+            System.getProperties().remove(key);
+    }
+
     /**
      * Gets the value of a system property as a String.
      * @return system property String value if it exists, overrideDefaultValue otherwise.
@@ -584,4 +612,3 @@ public boolean isPresent()
         return System.getProperties().containsKey(key);
     }
 }
-

src/java/org/apache/cassandra/io/util/PathUtils.java

@@ -24,6 +24,7 @@
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.*;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
@@ -44,6 +45,7 @@
 
 import static java.nio.file.StandardOpenOption.*;
 import static java.util.Collections.unmodifiableSet;
+import static org.apache.cassandra.config.CassandraRelevantProperties.USE_NIX_RECURSIVE_DELETE;
 import static org.apache.cassandra.utils.Throwables.merge;
 
 /**
@@ -71,7 +73,7 @@ public final class PathUtils
         if (StorageService.instance.isDaemonSetupCompleted())
             setDeletionListener(ignore -> {});
         else
-            logger.info("Deleting file during startup: {}", path);
+            logger.trace("Deleting file during startup: {}", path);
     };
 
     public static FileChannel newReadChannel(Path path) throws NoSuchFileException
@@ -310,17 +312,72 @@ public static Throwable delete(Path file, Throwable accumulate, @Nullable RateLi
         return accumulate;
     }
 
+    /**
+     * Uses unix `rm -r` to delete a directory recursively.
+     * Note that, it will trigger {@link #onDeletion} listener only for the provided path and will not call it for any
+     * nested path. This method can be much faster than deleting files and directories recursively by traversing them
+     * with Java. Though, we use it only for tests because it provides less information about the problem when something
+     * goes wrong.
+     *
+     * @param path    path to be deleted
+     * @param quietly if quietly, additional `-f` flag is added to the `rm` command so that it will not complain in case
+     *                the provided path is missing
+     */
+    private static void deleteRecursiveUsingNixCommand(Path path, boolean quietly)
+    {
+        String [] cmd = new String[]{ "rm", quietly ? "-rdf" : "-rd", path.toAbsolutePath().toString() };
+        try
+        {
+            if (!quietly && !Files.exists(path))
+                throw new NoSuchFileException(path.toString());
+
+            Process p = Runtime.getRuntime().exec(cmd);
+            int result = p.waitFor();
+
+            String out, err;
+            try (BufferedReader outReader = new BufferedReader(new InputStreamReader(p.getInputStream()));
+                 BufferedReader errReader = new BufferedReader(new InputStreamReader(p.getErrorStream())))
+            {
+                out = outReader.lines().collect(Collectors.joining("\n"));
+                err = errReader.lines().collect(Collectors.joining("\n"));
+            }
+
+            if (result != 0 && Files.exists(path))
+            {
+                logger.error("{} returned:\nstdout:\n{}\n\nstderr:\n{}", Arrays.toString(cmd), out, err);
+                throw new IOException(String.format("%s returned non-zero exit code: %d%nstdout:%n%s%n%nstderr:%n%s", Arrays.toString(cmd), result, out, err));
+            }
+
+            onDeletion.accept(path);
+        }
+        catch (IOException e)
+        {
+            throw propagateUnchecked(e, path, true);
+        }
+        catch (InterruptedException e)
+        {
+            Thread.currentThread().interrupt();
+            throw new FSWriteError(e, path);
+        }
+    }
+
     /**
      * Deletes all files and subdirectories under "path".
      * @param path file to be deleted
      * @throws FSWriteError if any part of the tree cannot be deleted
      */
     public static void deleteRecursive(Path path)
     {
+        if (USE_NIX_RECURSIVE_DELETE.getBoolean() && path.getFileSystem() == FileSystems.getDefault())
+        {
+            deleteRecursiveUsingNixCommand(path, false);
+            return;
+        }
+
         if (isDirectory(path))
             forEach(path, PathUtils::deleteRecursive);
 
-        // The directory is now empty so now it can be smoked
+        // The directory is now empty, so now it can be smoked
         delete(path);
     }
 
@@ -331,6 +388,12 @@ public static void deleteRecursive(Path path)
      */
     public static void deleteRecursive(Path path, RateLimiter rateLimiter)
     {
+        if (USE_NIX_RECURSIVE_DELETE.getBoolean() && path.getFileSystem() == FileSystems.getDefault())
+        {
+            deleteRecursiveUsingNixCommand(path, false);
+            return;
+        }
+
         deleteRecursive(path, rateLimiter, p -> deleteRecursive(p, rateLimiter));
     }
 

src/java/org/apache/cassandra/net/MessagingService.java

@@ -51,6 +51,7 @@
 import static java.util.Collections.synchronizedList;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static org.apache.cassandra.concurrent.Stage.MUTATION;
+import static org.apache.cassandra.config.CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN;
 import static org.apache.cassandra.utils.Clock.Global.nanoTime;
 import static org.apache.cassandra.utils.Throwables.maybeFail;
 
@@ -523,15 +524,20 @@ public Future<Void> maybeReconnectWithNewIp(InetAddressAndPort address, InetAddr
     }
 
     /**
-     * Wait for callbacks and don't allow any more to be created (since they could require writing hints)
+     * Wait for callbacks and don't allow anymore to be created (since they could require writing hints)
      */
     public void shutdown()
     {
-        shutdown(1L, MINUTES, true, true);
+        if (NON_GRACEFUL_SHUTDOWN.getBoolean())
+            // this branch is used in unit-tests when we really never restart a node and shutting down means the end of test
+            shutdownAbrubtly();
+        else
+            shutdown(1L, MINUTES, true, true);
     }
 
     public void shutdown(long timeout, TimeUnit units, boolean shutdownGracefully, boolean shutdownExecutors)
     {
+        logger.debug("Shutting down: timeout={}s, gracefully={}, shutdownExecutors={}", units.toSeconds(timeout), shutdownGracefully, shutdownExecutors);
         if (isShuttingDown)
         {
             logger.info("Shutdown was already called");
@@ -555,7 +561,7 @@ public void shutdown(long timeout, TimeUnit units, boolean shutdownGracefully, b
                       () -> {
                           List<ExecutorService> inboundExecutors = new ArrayList<>();
                           inboundSockets.close(synchronizedList(inboundExecutors)::add).get();
-                          ExecutorUtils.awaitTermination(1L, TimeUnit.MINUTES, inboundExecutors);
+                          ExecutorUtils.awaitTermination(timeout, units, inboundExecutors);
                       },
                       () -> {
                           if (shutdownExecutors)
@@ -587,6 +593,30 @@ public void shutdown(long timeout, TimeUnit units, boolean shutdownGracefully, b
         }
     }
 
+    public void shutdownAbrubtly()
+    {
+        logger.debug("Shutting down abruptly");
+        if (isShuttingDown)
+        {
+            logger.info("Shutdown was already called");
+            return;
+        }
+
+        isShuttingDown = true;
+        logger.info("Waiting for messaging service to quiesce");
+        // We may need to schedule hints on the mutation stage, so it's erroneous to shut down the mutation stage first
+        assert !MUTATION.executor().isShutdown();
+
+        callbacks.shutdownNow(false);
+        inboundSockets.close();
+        for (OutboundConnections pool : channelManagers.values())
+            pool.close(false);
+
+        maybeFail(socketFactory::shutdownNow,
+                  inboundSink::clear,
+                  outboundSink::clear);
+    }
+
     private void shutdownExecutors(long deadlineNanos) throws TimeoutException, InterruptedException
     {
         socketFactory.shutdownNow();

src/java/org/apache/cassandra/schema/SchemaKeyspace.java

@@ -76,7 +76,7 @@ private SchemaKeyspace()
 
     private static final Logger logger = LoggerFactory.getLogger(SchemaKeyspace.class);
 
-    private static final boolean FLUSH_SCHEMA_TABLES = Boolean.parseBoolean(System.getProperty("cassandra.test.flush_local_schema_changes", "true"));
+    private static final boolean FLUSH_SCHEMA_TABLES = CassandraRelevantProperties.FLUSH_LOCAL_SCHEMA_CHANGES.getBoolean();
     private static final boolean IGNORE_CORRUPTED_SCHEMA_TABLES = Boolean.parseBoolean(System.getProperty("cassandra.ignore_corrupted_schema_tables", "false"));
 
     /**

src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java

@@ -30,10 +30,12 @@
 import java.util.Collections;
 import java.util.Enumeration;
 
-import io.netty.util.concurrent.FastThreadLocal;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.utils.logging.LoggingSupportFactory;
+import io.netty.util.concurrent.FastThreadLocal;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.utils.logging.LoggingSupportFactory;
 
 /**
  * Custom {@link SecurityManager} and {@link Policy} implementation that only performs access checks
@@ -46,6 +48,8 @@
  */
 public final class ThreadAwareSecurityManager extends SecurityManager
 {
+    private static final Logger logger = LoggerFactory.getLogger(ThreadAwareSecurityManager.class);
+
     public static final PermissionCollection noPermissions = new PermissionCollection()
     {
         public void add(Permission permission)
@@ -81,6 +85,14 @@ public static void install()
     {
         if (installed)
             return;
+
+        // this line is needed - we need to make sure AccessControlException is loaded before we install this SM
+        // otherwise we may get into stackoverflow when javax.security is not allowed package, and ACE is tried to be
+        // loaded when it is going to be thrown from SM (class loader triggers SM to verify javax.security,
+        // it recognizes it as not allowed and attempts to throw it...)
+        //noinspection PlaceholderCountMatchesArgumentCount
+        logger.trace("Initialized thread aware security manager", AccessControlException.class.getName());
+
         System.setSecurityManager(new ThreadAwareSecurityManager());
         LoggingSupportFactory.getLoggingSupport().onStartup();
         installed = true;

test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java

@@ -60,6 +60,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.dht.IPartitioner;
@@ -184,6 +185,12 @@ public static abstract class AbstractBuilder<I extends IInstance, C extends IClu
         private INodeProvisionStrategy.Strategy nodeProvisionStrategy = INodeProvisionStrategy.Strategy.MultipleNetworkInterfaces;
         private ShutdownExecutor shutdownExecutor = DEFAULT_SHUTDOWN_EXECUTOR;
 
+        {
+            // those properties may be set for unit-test optimizations; those should not be used when running dtests
+            CassandraRelevantProperties.FLUSH_LOCAL_SCHEMA_CHANGES.reset();
+            CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
+        }
+
         public AbstractBuilder(Factory<I, C, B> factory)
         {
             super(factory);
@@ -1319,4 +1326,3 @@ private enum Utils implements NameHelper
         INSTANCE;
     }
 }
-