natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.11' into cassandra-4.0

This commit is contained in:
Jon Meredith 2022-04-27 12:01:42 -06:00
parent 018c8e0d5e 804098e68b
commit 2405d523fa
5 changed files with 58 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -1,5 +1,7 @@
4.0.5
* Optimise BTree build,update and transform operations (CASSANDRA-15510)
Merged from 3.0:
* Schema mutations may not be completed on drain (CASSANDRA-17524)
4.0.4
* Clean up schema migration coordinator and tests (CASSANDRA-17533)

55
src/java/org/apache/cassandra/concurrent/Stage.java
View File

@ -49,28 +49,35 @@ import static java.util.stream.Collectors.toMap;
public enum Stage
{
READ ("ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage),
MUTATION ("MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage),
COUNTER_MUTATION ("CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage),
VIEW_MUTATION ("ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage),
GOSSIP ("GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage),
REQUEST_RESPONSE ("RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage),
ANTI_ENTROPY ("AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage),
MIGRATION ("MigrationStage", "internal", () -> 1, null, Stage::singleThreadedStage),
MISC ("MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage),
TRACING ("TracingStage", "internal", () -> 1, null, Stage::tracingExecutor),
INTERNAL_RESPONSE ("InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage),
IMMEDIATE ("ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor);
READ (false, "ReadStage", "request", DatabaseDescriptor::getConcurrentReaders, DatabaseDescriptor::setConcurrentReaders, Stage::multiThreadedLowSignalStage),
MUTATION (true, "MutationStage", "request", DatabaseDescriptor::getConcurrentWriters, DatabaseDescriptor::setConcurrentWriters, Stage::multiThreadedLowSignalStage),
COUNTER_MUTATION (true, "CounterMutationStage", "request", DatabaseDescriptor::getConcurrentCounterWriters, DatabaseDescriptor::setConcurrentCounterWriters, Stage::multiThreadedLowSignalStage),
VIEW_MUTATION (true, "ViewMutationStage", "request", DatabaseDescriptor::getConcurrentViewWriters, DatabaseDescriptor::setConcurrentViewWriters, Stage::multiThreadedLowSignalStage),
GOSSIP (true, "GossipStage", "internal", () -> 1, null, Stage::singleThreadedStage),
REQUEST_RESPONSE (false, "RequestResponseStage", "request", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedLowSignalStage),
ANTI_ENTROPY (false, "AntiEntropyStage", "internal", () -> 1, null, Stage::singleThreadedStage),
MIGRATION (false, "MigrationStage", "internal", () -> 1, null, Stage::singleThreadedStage),
MISC (false, "MiscStage", "internal", () -> 1, null, Stage::singleThreadedStage),
TRACING (false, "TracingStage", "internal", () -> 1, null, Stage::tracingExecutor),
INTERNAL_RESPONSE (false, "InternalResponseStage", "internal", FBUtilities::getAvailableProcessors, null, Stage::multiThreadedStage),
IMMEDIATE (false, "ImmediateStage", "internal", () -> 0, null, Stage::immediateExecutor);
public static final long KEEP_ALIVE_SECONDS = 60; // seconds to keep "extra" threads alive for when idle
public final String jmxName;
/** Set true if this executor should be gracefully shutdown before stopping
* the commitlog allocator. Tasks on executors that issue mutations may
* block indefinitely waiting for a new commitlog segment, preventing a
* clean drain/shutdown.
*/
public final boolean shutdownBeforeCommitlog;
private final Supplier<LocalAwareExecutorService> initialiser;
private volatile LocalAwareExecutorService executor = null;
Stage(String jmxName, String jmxType, IntSupplier numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize, ExecutorServiceInitialiser initialiser)
Stage(Boolean shutdownBeforeCommitlog, String jmxName, String jmxType, IntSupplier numThreads, LocalAwareExecutorService.MaximumPoolSizeListener onSetMaximumPoolSize, ExecutorServiceInitialiser initialiser)
{
this.shutdownBeforeCommitlog = shutdownBeforeCommitlog;
this.jmxName = jmxName;
this.initialiser = () -> initialiser.init(jmxName, jmxType, numThreads.getAsInt(), onSetMaximumPoolSize);
this.initialiser = () -> initialiser.init(jmxName,jmxType, numThreads.getAsInt(), onSetMaximumPoolSize);
}
private static String normalizeName(String stageName)
@ -150,6 +157,14 @@ public enum Stage
.collect(Collectors.toList());
}
private static List<ExecutorService> mutatingExecutors()
{
return Stream.of(Stage.values())
.filter(stage -> stage.shutdownBeforeCommitlog)
.map(Stage::executor)
.collect(Collectors.toList());
}
/**
* This method shuts down all registered stages.
*/
@ -158,6 +173,18 @@ public enum Stage
ExecutorUtils.shutdownNow(executors());
}
public static void shutdownAndAwaitMutatingExecutors(boolean interrupt, long timeout, TimeUnit units) throws InterruptedException, TimeoutException
{
List<ExecutorService> executors = mutatingExecutors();
ExecutorUtils.shutdown(interrupt, executors);
ExecutorUtils.awaitTermination(timeout, units, executors);
}
public static boolean areMutationExecutorsTerminated()
{
return mutatingExecutors().stream().allMatch(ExecutorService::isTerminated);
}
@VisibleForTesting
public static void shutdownAndWait(long timeout, TimeUnit units) throws InterruptedException, TimeoutException
{

7
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -18,6 +18,8 @@
package org.apache.cassandra.config;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.exceptions.ConfigurationException;
/** A class that extracts system properties for the cassandra node it runs within. */
@ -147,6 +149,11 @@ public enum CassandraRelevantProperties
*/
BOOTSTRAP_SCHEMA_DELAY_MS("cassandra.schema_delay_ms"),
/**
* When draining, how long to wait for mutating executors to shutdown.
*/
DRAIN_EXECUTOR_TIMEOUT_MS("cassandra.drain_executor_timeout_ms", String.valueOf(TimeUnit.MINUTES.toMillis(5))),
/**
* Gossip quarantine delay is used while evaluating membership changes and should only be changed with extreme care.
*/

19
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -125,6 +125,7 @@ import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS;
import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK;
import static org.apache.cassandra.config.CassandraRelevantProperties.DRAIN_EXECUTOR_TIMEOUT_MS;
import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACEMENT_ALLOW_EMPTY;
import static org.apache.cassandra.index.SecondaryIndexManager.getIndexName;
import static org.apache.cassandra.index.SecondaryIndexManager.isIndexColumnFamily;
@ -4924,13 +4925,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
protected synchronized void drain(boolean isFinalShutdown) throws IOException, InterruptedException, ExecutionException
{
ExecutorService counterMutationStage = Stage.COUNTER_MUTATION.executor();
ExecutorService viewMutationStage = Stage.VIEW_MUTATION.executor();
ExecutorService mutationStage = Stage.MUTATION.executor();
if (mutationStage.isTerminated()
&& counterMutationStage.isTerminated()
&& viewMutationStage.isTerminated())
if (Stage.areMutationExecutorsTerminated())
{
if (!isFinalShutdown)
logger.warn("Cannot drain node (did it already happen?)");
@ -4985,14 +4980,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
if (!isFinalShutdown)
setMode(Mode.DRAINING, "clearing mutation stage", false);
viewMutationStage.shutdown();
counterMutationStage.shutdown();
mutationStage.shutdown();
// FIXME? should these *really* take up to one hour?
viewMutationStage.awaitTermination(3600, TimeUnit.SECONDS);
counterMutationStage.awaitTermination(3600, TimeUnit.SECONDS);
mutationStage.awaitTermination(3600, TimeUnit.SECONDS);
Stage.shutdownAndAwaitMutatingExecutors(false,
DRAIN_EXECUTOR_TIMEOUT_MS.getInt(), TimeUnit.MILLISECONDS);
StorageProxy.instance.verifyNoHintsInProgress();

5
test/distributed/org/apache/cassandra/distributed/test/ring/AutoBootstrapTest.java
View File

@ -26,11 +26,14 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.test.ring.BootstrapTest.count;
import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate;
public class AutoBootstrapTest extends BootstrapTest
public class AutoBootstrapTest extends TestBaseImpl
{
// Originally part of BootstrapTest. Broken out into separate test as the in-JVM dtests fail
// if too many instances are created in the same JVM. Bug in the JVM is suspected.