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

CEP-15: (C*) Accord message processing should avoid being passed on to a Stage and run directly in the messageing handler 

patch by David Capwell; reviewed by Ariel Weisberg, Benedict Elliott Smith for CASSANDRA-18364
This commit is contained in:
David Capwell 2023-04-27 11:30:50 -07:00
parent 7ad2bf672c
commit 8633a301f7
26 changed files with 273 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.build/build-accord.xml
View File

@ -27,8 +27,10 @@
<arg value="publishToMavenLocal" />
<arg value="-x" />
<arg value="test" />
<arg value="-x" />
<!-- since so much development is done from this hook, by adding checkstyle and rat will avoid issues earlier -->
<arg value="rat" />
<arg value="checkstyleMain" />
<arg value="checkstyleTest" />
<arg value="-Paccord_group=org.apache.cassandra" />
<arg value="-Paccord_artifactId=cassandra-accord" />
<arg value="-Paccord_version=${version}" />

1
.build/git/git-hooks/pre-commit/100-verify-submodules-pushed.sh
View File

@ -84,7 +84,6 @@ _main() {
_log "\t\tgit config --local cassandra.pre-commit.verify-submodules.enabled false"
_log "\tOr"
_log "\t\tgit config --local cassandra.pre-commit.verify-submodule-${file}.enabled false"
set -x
git_sub_dir="${file}/.git"
branch="$(git config -f .gitmodules "submodule.${file}.branch")"
[[ -z "${branch:-}" ]] && error "Submodule ${file} does not define a branch"

2
modules/accord

@ -1 +1 @@
Subproject commit 08aaab6e33d43406e0649146144e4df67648602a
Subproject commit 8226b2d7759319d7a0b0c823ab09b4344c5423f7

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

@ -47,7 +47,6 @@ public enum Stage
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),
ACCORD (true, "AccordStage", "request", DatabaseDescriptor::getConcurrentAccordOps, DatabaseDescriptor::setConcurrentAccordOps, 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),

5
src/java/org/apache/cassandra/config/Config.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.service.StartupChecks.StartupCheckType;
import org.apache.cassandra.utils.StorageCompatibilityMode;
import static org.apache.cassandra.config.CassandraRelevantProperties.AUTOCOMPACTION_ON_STARTUP_ENABLED;
import static org.apache.cassandra.config.CassandraRelevantProperties.CASSANDRA_AVAILABLE_PROCESSORS;
import static org.apache.cassandra.config.CassandraRelevantProperties.FILE_CACHE_ENABLED;
import static org.apache.cassandra.config.CassandraRelevantProperties.SKIP_PAXOS_REPAIR_ON_TOPOLOGY_CHANGE;
import static org.apache.cassandra.config.CassandraRelevantProperties.SKIP_PAXOS_REPAIR_ON_TOPOLOGY_CHANGE_KEYSPACES;
@ -190,10 +191,9 @@ public class Config
public int concurrent_reads = 32;
public int concurrent_writes = 32;
public int concurrent_accord_operations = 32;
public int concurrent_counter_writes = 32;
public int concurrent_materialized_view_writes = 32;
public int available_processors = -1;
public OptionaldPositiveInt available_processors = new OptionaldPositiveInt(CASSANDRA_AVAILABLE_PROCESSORS.getInt(OptionaldPositiveInt.UNDEFINED_VALUE));
public int memtable_flush_writers = 0;
@Replaces(oldName = "memtable_heap_space_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
@ -624,6 +624,7 @@ public class Config
public volatile boolean use_statements_enabled = true;
public boolean accord_transactions_enabled = false;
public OptionaldPositiveInt accord_shard_count = OptionaldPositiveInt.UNDEFINED;
/**
* Optionally disable asynchronous UDF execution.

30
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -643,9 +643,6 @@ public class DatabaseDescriptor
if (conf.concurrent_counter_writes < 2)
throw new ConfigurationException("concurrent_counter_writes must be at least 2, but was " + conf.concurrent_counter_writes, false);
if (conf.concurrent_accord_operations < 1)
throw new ConfigurationException("concurrent_accord_operations must be at least 1, but was " + conf.concurrent_accord_operations, false);
if (conf.networking_cache_size == null)
conf.networking_cache_size = new DataStorageSpec.IntMebibytesBound(Math.min(128, (int) (Runtime.getRuntime().maxMemory() / (16 * 1048576))));
@ -2670,20 +2667,6 @@ public class DatabaseDescriptor
conf.concurrent_materialized_view_writes = concurrent_materialized_view_writes;
}
public static int getConcurrentAccordOps()
{
return conf.concurrent_accord_operations;
}
public static void setConcurrentAccordOps(int concurrent_operations)
{
if (concurrent_operations < 0)
{
throw new IllegalArgumentException("Concurrent accord operations must be non-negative");
}
conf.concurrent_accord_operations = concurrent_operations;
}
public static int getFlushWriters()
{
return conf.memtable_flush_writers;
@ -2691,7 +2674,13 @@ public class DatabaseDescriptor
public static int getAvailableProcessors()
{
return conf == null ? -1 : conf.available_processors;
OptionaldPositiveInt ap = conf == null ? OptionaldPositiveInt.UNDEFINED : conf.available_processors;
return ap.or(Runtime.getRuntime()::availableProcessors);
}
public static void setAvailableProcessors(int value)
{
conf.available_processors = new OptionaldPositiveInt(value);
}
public static int getConcurrentCompactors()
@ -5198,6 +5187,11 @@ public class DatabaseDescriptor
conf.accord_transactions_enabled = b;
}
public static int getAccordShardCount()
{
return conf.accord_shard_count.or(DatabaseDescriptor::getAvailableProcessors);
}
public static boolean getForceNewPreparedStatementBehaviour()
{
return conf.force_new_prepared_statement_behaviour;

73
src/java/org/apache/cassandra/config/OptionaldPositiveInt.java Normal file
View File

@ -0,0 +1,73 @@
/*
* 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.config;
import java.util.Objects;
import java.util.function.IntSupplier;
public class OptionaldPositiveInt
{
public static final int UNDEFINED_VALUE = -1;
public static final OptionaldPositiveInt UNDEFINED = new OptionaldPositiveInt(UNDEFINED_VALUE);
private final int value;
public OptionaldPositiveInt(int value)
{
if (!(value == -1 || value >= 1))
throw new IllegalArgumentException(String.format("Only -1 (undefined) and positive values are allowed; given %d", value));
this.value = value;
}
public boolean isDefined()
{
return value != UNDEFINED_VALUE;
}
public int or(int defaultValue)
{
return value == UNDEFINED_VALUE ? defaultValue : value;
}
public int or(IntSupplier defaultValue)
{
return value == UNDEFINED_VALUE ? defaultValue.getAsInt() : value;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
OptionaldPositiveInt that = (OptionaldPositiveInt) o;
return value == that.value;
}
@Override
public int hashCode()
{
return Objects.hash(value);
}
@Override
public String toString()
{
return !isDefined() ? "null" : Integer.toString(value);
}
}

26
src/java/org/apache/cassandra/config/YamlConfigurationLoader.java
View File

@ -135,16 +135,7 @@ public class YamlConfigurationLoader implements ConfigurationLoader
throw new AssertionError(e);
}
SafeConstructor constructor = new CustomConstructor(Config.class, Yaml.class.getClassLoader());
Map<Class<?>, Map<String, Replacement>> replacements = getNameReplacements(Config.class);
verifyReplacements(replacements, configBytes);
PropertiesChecker propertiesChecker = new PropertiesChecker(replacements);
constructor.setPropertyUtils(propertiesChecker);
Yaml yaml = new Yaml(constructor);
Config result = loadConfig(yaml, configBytes);
propertiesChecker.check();
maybeAddSystemProperties(result);
return result;
return loadConfig(configBytes);
}
catch (YAMLException e)
{
@ -152,6 +143,21 @@ public class YamlConfigurationLoader implements ConfigurationLoader
}
}
@VisibleForTesting
static Config loadConfig(byte[] configBytes)
{
SafeConstructor constructor = new CustomConstructor(Config.class, Yaml.class.getClassLoader());
Map<Class<?>, Map<String, Replacement>> replacements = getNameReplacements(Config.class);
verifyReplacements(replacements, configBytes);
PropertiesChecker propertiesChecker = new PropertiesChecker(replacements);
constructor.setPropertyUtils(propertiesChecker);
Yaml yaml = new Yaml(constructor);
Config result = loadConfig(yaml, configBytes);
propertiesChecker.check();
maybeAddSystemProperties(result);
return result;
}
private static void maybeAddSystemProperties(Object obj)
{
if (CassandraRelevantProperties.CONFIG_ALLOW_SYSTEM_PROPERTIES.getBoolean())

30
src/java/org/apache/cassandra/net/Verb.java
View File

@ -268,29 +268,29 @@ public enum Verb
// accord
ACCORD_SIMPLE_RSP (119, P2, writeTimeout, REQUEST_RESPONSE, () -> EnumSerializer.simpleReply, RESPONSE_HANDLER ),
ACCORD_PREACCEPT_RSP (121, P2, writeTimeout, REQUEST_RESPONSE, () -> PreacceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_PREACCEPT_REQ (120, P2, writeTimeout, ACCORD, () -> PreacceptSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_PREACCEPT_RSP ),
ACCORD_PREACCEPT_REQ (120, P2, writeTimeout, IMMEDIATE, () -> PreacceptSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_PREACCEPT_RSP ),
ACCORD_ACCEPT_RSP (124, P2, writeTimeout, REQUEST_RESPONSE, () -> AcceptSerializers.reply, RESPONSE_HANDLER ),
ACCORD_ACCEPT_REQ (122, P2, writeTimeout, ACCORD, () -> AcceptSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_ACCEPT_RSP ),
ACCORD_ACCEPT_INVALIDATE_REQ (123, P2, writeTimeout, ACCORD, () -> AcceptSerializers.invalidate, () -> AccordService.instance().verbHandler(), ACCORD_ACCEPT_RSP ),
ACCORD_ACCEPT_REQ (122, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_ACCEPT_RSP ),
ACCORD_ACCEPT_INVALIDATE_REQ (123, P2, writeTimeout, IMMEDIATE, () -> AcceptSerializers.invalidate, () -> AccordService.instance().verbHandler(), ACCORD_ACCEPT_RSP ),
ACCORD_READ_RSP (128, P2, writeTimeout, REQUEST_RESPONSE, () -> ReadDataSerializers.reply, RESPONSE_HANDLER ),
ACCORD_READ_REQ (127, P2, writeTimeout, ACCORD, () -> ReadDataSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_READ_RSP ),
ACCORD_COMMIT_REQ (125, P2, writeTimeout, ACCORD, () -> CommitSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_READ_RSP ),
ACCORD_COMMIT_INVALIDATE_REQ (126, P2, writeTimeout, ACCORD, () -> CommitSerializers.invalidate, () -> AccordService.instance().verbHandler() ),
ACCORD_READ_REQ (127, P2, writeTimeout, IMMEDIATE, () -> ReadDataSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_READ_RSP ),
ACCORD_COMMIT_REQ (125, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_READ_RSP ),
ACCORD_COMMIT_INVALIDATE_REQ (126, P2, writeTimeout, IMMEDIATE, () -> CommitSerializers.invalidate, () -> AccordService.instance().verbHandler() ),
ACCORD_APPLY_RSP (130, P2, writeTimeout, REQUEST_RESPONSE, () -> ApplySerializers.reply, RESPONSE_HANDLER ),
ACCORD_APPLY_REQ (129, P2, writeTimeout, ACCORD, () -> ApplySerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_APPLY_RSP ),
ACCORD_APPLY_REQ (129, P2, writeTimeout, IMMEDIATE, () -> ApplySerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_APPLY_RSP ),
ACCORD_RECOVER_RSP (132, P2, writeTimeout, REQUEST_RESPONSE, () -> RecoverySerializers.reply, RESPONSE_HANDLER ),
ACCORD_RECOVER_REQ (131, P2, writeTimeout, ACCORD, () -> RecoverySerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_RECOVER_RSP ),
ACCORD_RECOVER_REQ (131, P2, writeTimeout, IMMEDIATE, () -> RecoverySerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_RECOVER_RSP ),
ACCORD_BEGIN_INVALIDATE_RSP (134, P2, writeTimeout, REQUEST_RESPONSE, () -> BeginInvalidationSerializers.reply, RESPONSE_HANDLER ),
ACCORD_BEGIN_INVALIDATE_REQ (133, P2, writeTimeout, ACCORD, () -> BeginInvalidationSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_BEGIN_INVALIDATE_RSP ),
ACCORD_BEGIN_INVALIDATE_REQ (133, P2, writeTimeout, IMMEDIATE, () -> BeginInvalidationSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_BEGIN_INVALIDATE_RSP ),
ACCORD_WAIT_COMMIT_RSP (136, P2, writeTimeout, REQUEST_RESPONSE, () -> WaitOnCommitSerializer.reply, RESPONSE_HANDLER ),
ACCORD_WAIT_COMMIT_REQ (135, P2, writeTimeout, ACCORD, () -> WaitOnCommitSerializer.request, () -> AccordService.instance().verbHandler(), ACCORD_WAIT_COMMIT_RSP ),
ACCORD_INFORM_OF_TXNID_REQ (137, P2, writeTimeout, ACCORD, () -> InformOfTxnIdSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_HOME_DURABLE_REQ (138, P2, writeTimeout, ACCORD, () -> InformHomeDurableSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_DURABLE_REQ (139, P2, writeTimeout, ACCORD, () -> InformDurableSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_WAIT_COMMIT_REQ (135, P2, writeTimeout, IMMEDIATE, () -> WaitOnCommitSerializer.request, () -> AccordService.instance().verbHandler(), ACCORD_WAIT_COMMIT_RSP ),
ACCORD_INFORM_OF_TXNID_REQ (137, P2, writeTimeout, IMMEDIATE, () -> InformOfTxnIdSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_HOME_DURABLE_REQ (138, P2, writeTimeout, IMMEDIATE, () -> InformHomeDurableSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_DURABLE_REQ (139, P2, writeTimeout, IMMEDIATE, () -> InformDurableSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_SIMPLE_RSP ),
ACCORD_CHECK_STATUS_RSP (141, P2, writeTimeout, REQUEST_RESPONSE, () -> CheckStatusSerializers.reply, RESPONSE_HANDLER ),
ACCORD_CHECK_STATUS_REQ (140, P2, writeTimeout, ACCORD, () -> CheckStatusSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_CHECK_STATUS_RSP ),
ACCORD_CHECK_STATUS_REQ (140, P2, writeTimeout, IMMEDIATE, () -> CheckStatusSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_CHECK_STATUS_RSP ),
ACCORD_GET_DEPS_RSP (143, P2, writeTimeout, REQUEST_RESPONSE, () -> GetDepsSerializers.reply, RESPONSE_HANDLER ),
ACCORD_GET_DEPS_REQ (142, P2, writeTimeout, ACCORD, () -> GetDepsSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_GET_DEPS_RSP ),
ACCORD_GET_DEPS_REQ (142, P2, writeTimeout, IMMEDIATE, () -> GetDepsSerializers.request, () -> AccordService.instance().verbHandler(), ACCORD_GET_DEPS_RSP ),
// generic failure response

21
src/java/org/apache/cassandra/service/accord/AccordCallback.java
View File

@ -22,28 +22,29 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.coordinate.Timeout;
import accord.local.AgentExecutor;
import accord.messages.Callback;
import accord.messages.SafeCallback;
import accord.messages.Reply;
import org.apache.cassandra.exceptions.RequestFailureReason;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.RequestCallback;
class AccordCallback<T extends Reply> implements RequestCallback<T>
class AccordCallback<T extends Reply> extends SafeCallback<T> implements RequestCallback<T>
{
private static final Logger logger = LoggerFactory.getLogger(AccordCallback.class);
private final Callback<T> callback;
public AccordCallback(Callback<T> callback)
public AccordCallback(AgentExecutor executor, Callback<T> callback)
{
this.callback = callback;
super(executor, callback);
}
@Override
public void onResponse(Message<T> msg)
{
logger.debug("Received response {} from {}", msg.payload, msg.from());
callback.onSuccess(EndpointMapping.endpointToId(msg.from()), msg.payload);
success(EndpointMapping.endpointToId(msg.from()), msg.payload);
}
private static Throwable convertReason(RequestFailureReason reason)
@ -56,9 +57,15 @@ class AccordCallback<T extends Reply> implements RequestCallback<T>
@Override
public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason)
{
logger.debug("Received failure {} from {} for {}", failureReason, from, callback);
logger.debug("Received failure {} from {} for {}", failureReason, from, this);
// TODO (now): we should distinguish timeout failures with some placeholder Exception
callback.onFailure(EndpointMapping.endpointToId(from), convertReason(failureReason));
failure(EndpointMapping.endpointToId(from), convertReason(failureReason));
}
@Override
public boolean trackLatencyForSnitch()
{
return true;
}
@Override

33
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -50,7 +50,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
public class AccordCommandStore implements CommandStore
public class AccordCommandStore extends CommandStore
{
private static long getThreadId(ExecutorService executor)
{
@ -68,7 +68,6 @@ public class AccordCommandStore implements CommandStore
}
}
private final int id;
private final long threadId;
public final String loggingId;
private final ExecutorService executor;
@ -79,12 +78,6 @@ public class AccordCommandStore implements CommandStore
private AccordSafeCommandStore current = null;
private long lastSystemTimestampMicros = Long.MIN_VALUE;
private final NodeTimeService time;
private final Agent agent;
private final DataStore dataStore;
private final ProgressLog progressLog;
private final RangesForEpochHolder rangesForEpochHolder;
public AccordCommandStore(int id,
NodeTimeService time,
Agent agent,
@ -92,24 +85,20 @@ public class AccordCommandStore implements CommandStore
ProgressLog.Factory progressLogFactory,
RangesForEpochHolder rangesForEpoch)
{
this.id = id;
this.time = time;
this.agent = agent;
this.dataStore = dataStore;
this.progressLog = progressLogFactory.create(this);
this.rangesForEpochHolder = rangesForEpoch;
super(id, time, agent, dataStore, progressLogFactory, rangesForEpoch);
this.loggingId = String.format("[%s]", id);
this.executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']');
this.threadId = getThreadId(this.executor);
this.stateCache = new AccordStateCache(8<<20);
this.commandCache = stateCache.instance(TxnId.class, accord.local.Command.class, AccordSafeCommand::new, AccordObjectSizes::command);
this.commandsForKeyCache = stateCache.instance(RoutableKey.class, CommandsForKey.class, AccordSafeCommandsForKey::new, AccordObjectSizes::commandsForKey);
executor.execute(() -> CommandStore.register(this));
}
@Override
public int id()
public boolean inStore()
{
return id;
return Thread.currentThread().getId() == threadId;
}
public void setCacheSize(long bytes)
@ -125,12 +114,12 @@ public class AccordCommandStore implements CommandStore
public void checkInStoreThread()
{
Invariants.checkState(Thread.currentThread().getId() == threadId);
Invariants.checkState(inStore());
}
public void checkNotInStoreThread()
{
Invariants.checkState(Thread.currentThread().getId() != threadId);
Invariants.checkState(!inStore());
}
public ExecutorService executor()
@ -197,13 +186,7 @@ public class AccordCommandStore implements CommandStore
public DataStore dataStore()
{
return dataStore;
}
@Override
public Agent agent()
{
return agent;
return store;
}
NodeTimeService time()

5
src/java/org/apache/cassandra/service/accord/AccordCommandStores.java
View File

@ -25,14 +25,15 @@ import accord.local.CommandStores;
import accord.local.NodeTimeService;
import accord.local.ShardDistributor;
import accord.topology.Topology;
import accord.utils.RandomSource;
public class AccordCommandStores extends CommandStores<AccordCommandStore>
{
private long cacheSize;
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store,
AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random,
ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory)
{
super(time, agent, store, shardDistributor, progressLogFactory, AccordCommandStore::new);
super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore::new);
setCacheSize(maxCacheSize());
}

23
src/java/org/apache/cassandra/service/accord/AccordMessageSink.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord;
import java.util.EnumMap;
import java.util.Map;
import java.util.Objects;
import com.google.common.base.Preconditions;
@ -28,7 +27,9 @@ import org.apache.cassandra.net.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.Agent;
import accord.api.MessageSink;
import accord.local.AgentExecutor;
import accord.local.Node;
import accord.messages.Callback;
import accord.messages.MessageType;
@ -90,23 +91,25 @@ public class AccordMessageSink implements MessageSink
return VerbMapping.instance.mapping.get(type);
}
private final Agent agent;
private final MessageDelivery messaging;
public AccordMessageSink(MessageDelivery messaging)
public AccordMessageSink(Agent agent, MessageDelivery messaging)
{
this.agent = agent;
this.messaging = messaging;
}
public AccordMessageSink()
public AccordMessageSink(Agent agent)
{
this(MessagingService.instance());
this(agent, MessagingService.instance());
}
@Override
public void send(Node.Id to, Request request)
{
Verb verb = getVerb(request.type());
Objects.requireNonNull(verb, "verb");
Preconditions.checkNotNull(verb, "Verb is null for type %s", request.type());
Message<Request> message = Message.out(verb, request);
InetAddressAndPort endpoint = getEndpoint(to);
logger.debug("Sending {} {} to {}", verb, message.payload, endpoint);
@ -114,14 +117,14 @@ public class AccordMessageSink implements MessageSink
}
@Override
public void send(Node.Id to, Request request, Callback callback)
public void send(Node.Id to, Request request, AgentExecutor executor, Callback callback)
{
Verb verb = getVerb(request.type());
Preconditions.checkArgument(verb != null);
Preconditions.checkNotNull(verb, "Verb is null for type %s", request.type());
Message<Request> message = Message.out(verb, request);
InetAddressAndPort endpoint = getEndpoint(to);
logger.debug("Sending {} {} to {}", verb, message.payload, endpoint);
messaging.sendWithCallback(message, endpoint, new AccordCallback<>((Callback<Reply>) callback));
messaging.sendWithCallback(message, endpoint, new AccordCallback<>(executor, (Callback<Reply>) callback));
}
@Override
@ -129,7 +132,9 @@ public class AccordMessageSink implements MessageSink
{
Message<?> replyTo = (Message<?>) replyContext;
Message<?> replyMsg = replyTo.responseWith(reply);
Preconditions.checkArgument(replyMsg.verb() == getVerb(reply.type()));
Verb verb = getVerb(reply.type());
Preconditions.checkNotNull(verb, "Verb is null for type %s", reply.type());
Preconditions.checkArgument(replyMsg.verb() == verb, "Expected reply message with verb %s but got %s; reply type was %s", replyMsg.verb(), verb, reply.type());
InetAddressAndPort endpoint = getEndpoint(replyingToNode);
logger.debug("Replying {} {} to {}", replyMsg.verb(), replyMsg.payload, endpoint);
messaging.send(replyMsg, endpoint);

8
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -61,7 +61,6 @@ import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.config.DatabaseDescriptor.getConcurrentAccordOps;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -134,7 +133,8 @@ public class AccordService implements IAccordService, Shutdownable
{
Node.Id localId = EndpointMapping.endpointToId(FBUtilities.getBroadcastAddressAndPort());
logger.info("Starting accord with nodeId {}", localId);
this.messageSink = new AccordMessageSink();
AccordAgent agent = new AccordAgent();
this.messageSink = new AccordMessageSink(agent);
this.configService = new AccordConfigurationService(localId);
this.scheduler = new AccordScheduler();
this.node = new Node(localId,
@ -142,8 +142,8 @@ public class AccordService implements IAccordService, Shutdownable
configService,
AccordService::uniqueNow,
() -> null,
new KeyspaceSplitter(new EvenSplit<>(getConcurrentAccordOps(), getPartitioner().accordSplitter())),
new AccordAgent(),
new KeyspaceSplitter(new EvenSplit<>(DatabaseDescriptor.getAccordShardCount(), getPartitioner().accordSplitter())),
agent,
new DefaultRandom(),
scheduler,
SizeOfIntersectionSorter.SUPPLIER,

2
src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
View File

@ -302,7 +302,7 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
{
Invariants.checkArgument(this.callback == null);
this.callback = callback;
commandStore.executor().submit(this);
commandStore.executor().execute(this);
}
private static Iterable<RoutableKey> toRoutableKeys(Seekables<?, ?> keys)

6
src/java/org/apache/cassandra/service/accord/serializers/ReadDataSerializers.java
View File

@ -81,7 +81,7 @@ public class ReadDataSerializers
out.writeByte(0);
ReadOk readOk = (ReadOk) reply;
TxnData.serializer.serialize((TxnData) readOk.data, out, version);
TxnData.nullableSerializer.serialize((TxnData) readOk.data, out, version);
}
@Override
@ -91,7 +91,7 @@ public class ReadDataSerializers
if (id != 0)
return nacks[id - 1];
return new ReadOk(TxnData.serializer.deserialize(in, version));
return new ReadOk(TxnData.nullableSerializer.deserialize(in, version));
}
@Override
@ -101,7 +101,7 @@ public class ReadDataSerializers
return TypeSizes.BYTE_SIZE;
ReadOk readOk = (ReadOk) reply;
return TypeSizes.BYTE_SIZE + TxnData.serializer.serializedSize((TxnData) readOk.data, version);
return TypeSizes.BYTE_SIZE + TxnData.nullableSerializer.serializedSize((TxnData) readOk.data, version);
}
};
}

3
src/java/org/apache/cassandra/service/accord/txn/TxnData.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.NullableSerializer;
import org.apache.cassandra.utils.ObjectSizes;
public class TxnData implements Data, Result, Iterable<FilteredPartition>
@ -196,4 +197,6 @@ public class TxnData implements Data, Result, Iterable<FilteredPartition>
return size;
}
};
public static final IVersionedSerializer<TxnData> nullableSerializer = NullableSerializer.wrap(serializer);
}

7
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -137,7 +137,7 @@ public class FBUtilities
public static void setAvailableProcessors(int value)
{
availableProcessors = value;
DatabaseDescriptor.setAvailableProcessors(value);
}
@VisibleForTesting
@ -148,10 +148,7 @@ public class FBUtilities
public static int getAvailableProcessors()
{
if (availableProcessors > 0)
return availableProcessors;
else
return Runtime.getRuntime().availableProcessors();
return DatabaseDescriptor.getAvailableProcessors();
}
public static final int MAX_UNSIGNED_SHORT = 0xFFFF;

2
test/data/config/version=5.1-alpha1.yml
View File

@ -408,7 +408,7 @@ max_concurrent_automatic_sstable_upgrades: "java.lang.Integer"
maximum_replication_factor_warn_threshold: "java.lang.Integer"
denylist_reads_enabled: "java.lang.Boolean"
permissions_cache_active_update: "java.lang.Boolean"
available_processors: "java.lang.Integer"
available_processors: "org.apache.cassandra.config.OptionaldPositiveInt"
file_cache_round_up: "java.lang.Boolean"
secondary_indexes_per_table_warn_threshold: "java.lang.Integer"
tables_warn_threshold: "java.lang.Integer"

7
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
View File

@ -91,7 +91,6 @@ import org.apache.cassandra.simulator.utils.KindOfSequence;
import org.apache.cassandra.simulator.utils.LongRange;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.BufferPool;
@ -580,7 +579,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
.set("concurrent_counter_writes", take(1, 4))
.set("concurrent_materialized_view_writes", take(1, 4))
.set("concurrent_reads", take(1, 4))
.forceSet("available_processors", take(3, 4));
.set("available_processors", take(3, 4));
}
// begin allocating for a new node
@ -627,7 +626,7 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
if (remaining * min <= allocationPool)
return min;
if (times == remaining)
return allocationPool / remaining;
return Math.max(allocationPool / remaining, min);
if (times + 1 == remaining)
return random.uniform(Math.max(min, (allocationPool - max) / times), Math.min(max, (allocationPool - min) / times));
@ -638,7 +637,6 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
}
}
public final RandomSource random;
public final SimulatedSystems simulated;
public final Cluster cluster;
@ -772,7 +770,6 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
@Override
public void beforeStartup(IInstance i)
{
((IInvokableInstance) i).unsafeAcceptOnThisThread(FBUtilities::setAvailableProcessors, i.config().getInt("available_processors"));
((IInvokableInstance) i).unsafeAcceptOnThisThread(IfInterceptibleThread::setThreadLocalRandomCheck, (LongConsumer) threadLocalRandomCheck);
int num = i.config().num();

4
test/simulator/main/org/apache/cassandra/simulator/paxos/AbstractPairOfSequencesPaxosSimulation.java
View File

@ -209,6 +209,8 @@ abstract class AbstractPairOfSequencesPaxosSimulation extends PaxosSimulation
public Action get()
{
int[] primaryKeyIndex = consume(simulated.random, available);
if (primaryKeyIndex == null)
return Actions.empty("All primary keys are taken, try again later");
long untilNanos = simulated.time.nanoTime() + SECONDS.toNanos(simulateKeyForSeconds.select(simulated.random));
int concurrency = withinKeyConcurrency.select(simulated.random);
Supplier<Action> supplier = factory.apply(simulated, primaryKeyIndex);
@ -249,7 +251,7 @@ abstract class AbstractPairOfSequencesPaxosSimulation extends PaxosSimulation
private int[] consume(RandomSource random, List<Integer> available)
{
if (available.isEmpty())
throw new AssertionError("available partitions are empty!");
return null;
int numPartitions = available.size() == 1 || !allowMultiplePartitions() ? 1 : random.uniform(1, available.size());
int[] partitions = new int[numPartitions];
for (int counter = 0; counter < numPartitions; counter++)

32
test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulation.java
View File

@ -18,7 +18,9 @@
package org.apache.cassandra.simulator.paxos;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@ -286,19 +288,39 @@ public abstract class PaxosSimulation implements Simulation, ClusterActionListen
private RuntimeException logAndThrow()
{
Integer causedByPrimaryKey = null;
Throwable causedByThrowable = null;
class Violation
{
final int primaryKey;
final Throwable cause;
Violation(int primaryKey, Throwable cause)
{
this.primaryKey = primaryKey;
this.cause = cause;
}
}
List<Violation> violations = new ArrayList<>();
for (Throwable t : simulated.failures.get())
{
Integer causedByPrimaryKey;
if (null != (causedByPrimaryKey = causedBy(t)))
{
causedByThrowable = t;
violations.add(new Violation(causedByPrimaryKey, t));
break;
}
}
log(causedByPrimaryKey);
Throwable t = (causedByPrimaryKey != null) ? causedByThrowable : simulated.failures.get().get(0);
if (!violations.isEmpty())
{
AssertionError error = new AssertionError("History violations detected");
violations.forEach(v -> {
log(v.primaryKey);
error.addSuppressed(v.cause);
});
throw error;
}
Throwable t = simulated.failures.get().get(0);
Throwables.throwIfUnchecked(t);
throw new RuntimeException(t);
}

8
test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedAction.java
View File

@ -385,11 +385,11 @@ public abstract class SimulatedAction extends Action implements InterceptorOfCon
notify = from;
}
boolean isTimeout = deliver != FAILURE;
Executor callbackExecutor = notify.executorFor(verb.id);
if (callbackExecutor instanceof ImmediateExecutor)
callbackExecutor = to.executor();
Executor notifierExecutor = notify.executorFor(verb.id);
if (notifierExecutor instanceof ImmediateExecutor)
notifierExecutor = notify.executor();
InterceptedExecution.InterceptedTaskExecution failTask = new InterceptedRunnableExecution(
(InterceptingExecutor) callbackExecutor,
(InterceptingExecutor) notifierExecutor,
() -> notify.unsafeApplyOnThisThread((socketAddress, id, innerIsTimeout) -> {
InetAddressAndPort address = InetAddressAndPort.getByAddress(socketAddress);
RequestCallbacks.CallbackInfo callback = instance().callbacks.remove(id, address);

1
test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java
View File

@ -147,6 +147,7 @@ public class DatabaseDescriptorRefTest
"org.apache.cassandra.config.GuardrailsOptions$ConsistencyLevels",
"org.apache.cassandra.config.GuardrailsOptions$TableProperties",
"org.apache.cassandra.config.JMXServerOptions",
"org.apache.cassandra.config.OptionaldPositiveInt",
"org.apache.cassandra.config.ParameterizedClass",
"org.apache.cassandra.config.RepairConfig",
"org.apache.cassandra.config.RepairRetrySpec",

51
test/unit/org/apache/cassandra/config/YamlConfigurationLoaderTest.java
View File

@ -33,6 +33,9 @@ import java.util.function.Predicate;
import com.google.common.collect.ImmutableMap;
import org.junit.Test;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import org.apache.cassandra.distributed.shared.WithProperties;
import org.apache.cassandra.io.util.File;
import org.yaml.snakeyaml.error.YAMLException;
@ -426,6 +429,36 @@ public class YamlConfigurationLoaderTest
assertThat(from("compaction_tombstone_warning_threshold", "0").partition_tombstones_warn_threshold).isEqualTo(0);
}
@Test
public void process()
{
for (Type type : Type.values())
{
Config c = fromType(type, "available_processors", 4);
assertThat(c.available_processors).isEqualTo(new OptionaldPositiveInt(4));
assertThat(c.accord_shard_count).isEqualTo(OptionaldPositiveInt.UNDEFINED);
c = fromType(type, "available_processors", 3, "accord_shard_count", 1);
assertThat(c.available_processors).isEqualTo(new OptionaldPositiveInt(3));
assertThat(c.accord_shard_count).isEqualTo(new OptionaldPositiveInt(1));
}
}
private enum Type { MAP, YAML }
private static Config fromType(Type type, Object... values)
{
switch (type)
{
case MAP:
return from(values);
case YAML:
return fromYaml(values);
default:
throw new AssertionError("Unexpected type: " + type);
}
}
private static Config from(Object... values)
{
assert values.length % 2 == 0 : "Map can only be created with an even number of inputs: given " + values.length;
@ -469,6 +502,24 @@ public class YamlConfigurationLoaderTest
assertTrue(config.authenticator.parameters.isEmpty());
}
private static Config fromYaml(Object... values)
{
assert values.length % 2 == 0 : "Map can only be created with an even number of inputs: given " + values.length;
ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
for (int i = 0; i < values.length; i += 2)
builder.put((String) values[i], values[i + 1]);
ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); // checkstyle: permit this instantiation
try
{
byte[] bytes = mapper.writeValueAsBytes(builder.build());
return YamlConfigurationLoader.loadConfig(bytes);
}
catch (JsonProcessingException e)
{
throw new AssertionError("Unable to convert map to YAML", e);
}
}
public static Config load(String path)
{
URL url = YamlConfigurationLoaderTest.class.getClassLoader().getResource(path);

12
test/unit/org/apache/cassandra/service/accord/AccordMessageSinkTest.java
View File

@ -18,17 +18,21 @@
package org.apache.cassandra.service.accord;
import org.apache.cassandra.net.MessageDelivery;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
import accord.api.Agent;
import accord.local.Node;
import accord.messages.InformOfTxnId;
import accord.messages.SimpleReply;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.Verb;
import org.mockito.Mockito;
import org.apache.cassandra.tcm.ClusterMetadataService;
public class AccordMessageSinkTest
{
@ -36,6 +40,8 @@ public class AccordMessageSinkTest
public static void setup()
{
DatabaseDescriptor.clientInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ClusterMetadataService.initializeForClients();
}
@Test
@ -48,7 +54,7 @@ public class AccordMessageSinkTest
SimpleReply reply = SimpleReply.Ok;
MessageDelivery messaging = Mockito.mock(MessageDelivery.class);
AccordMessageSink sink = new AccordMessageSink(messaging);
AccordMessageSink sink = new AccordMessageSink(Mockito.mock(Agent.class), messaging);
sink.reply(new Node.Id(1), req, reply);
Mockito.verify(messaging).send(Mockito.any(), Mockito.any());