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

Improve Topology Management 

Merge ConfigurationService with TopologyManager to remove cyclic dependency and duplicated work.
Also:
 - Improve visibility of work blocking topology processing
 - Ensure we cannot double-count peers when deciding an epoch's distributed readiness
 - Remove the possibility of distributed stalls, by processing new topologies as soon as a
   contiguous sequence is known locally, regardless of whether anyprior local epoch is ready to
   coordinate or has recorded this fact to peers. The notification of local readiness continues
   to be processed only once all prior epochs are ready, but we can begin using and readying
   later epochs immediately.

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20998
This commit is contained in:
Benedict Elliott Smith 2025-10-28 18:52:06 +00:00
parent 1a96b8803a
commit aa5c3aba1d
40 changed files with 664 additions and 1380 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit a575ab316b4e79f99009801312579bcbf86451ef
Subproject commit a681b1bf7d9e8d1d797af2fd54bdd180a4863406

7
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -56,6 +56,8 @@ import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.RoutingKeys;
import accord.topology.ActiveEpoch;
import accord.topology.ActiveEpochs;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.utils.SortedListMap;
@ -2166,7 +2168,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
public void collect(PartitionsCollector collector)
{
IAccordService service = AccordService.unsafeInstance();
List<Topology> snapshot = service.node().topology().topologySnapshot();
ActiveEpochs snapshot = service.node().topology().active();
Map<TableId, Map<TokenKey, List<ShardAndEpochs>>> tableIdLookup = new HashMap<>();
{
@ -2177,8 +2179,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace
TableId prevTableId = null;
Map<TokenKey, List<ShardAndEpochs>> startLookup = null;
for (Topology topology : snapshot)
for (ActiveEpoch epoch : snapshot)
{
Topology topology = epoch.global();
for (Shard shard : topology.shards())
{
Range range = shard.range;

45
src/java/org/apache/cassandra/db/virtual/AccordVirtualTables.java
View File

@ -32,9 +32,9 @@ import com.google.common.collect.Sets;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.topology.TopologyManager.EpochsSnapshot;
import accord.topology.TopologyManager.EpochsSnapshot.Epoch;
import accord.topology.TopologyManager.EpochsSnapshot.EpochReady;
import accord.topology.ActiveEpoch;
import accord.topology.ActiveEpochs;
import accord.topology.EpochReady;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.marshal.LongType;
@ -46,8 +46,6 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.TokenRange;
import static accord.topology.TopologyManager.EpochsSnapshot.ResultStatus.SUCCESS;
public class AccordVirtualTables
{
public static final String EPOCHS = "accord_epochs";
@ -79,7 +77,7 @@ public class AccordVirtualTables
{
super(parse(keyspace, "CREATE TABLE " + EPOCHS + " (\n" +
" epoch bigint PRIMARY KEY,\n" +
" ready_metadata text,\n" +
" ready_active text,\n" +
" ready_coordinate text,\n" +
" ready_data text,\n" +
" ready_reads text,\n" +
@ -94,16 +92,16 @@ public class AccordVirtualTables
public DataSet data()
{
SimpleDataSet ds = new SimpleDataSet(metadata());
EpochsSnapshot snapshot = epochsSnapshot();
for (Epoch epoch : snapshot)
ActiveEpochs snapshot = AccordService.instance().topology().active();
for (ActiveEpoch epoch : snapshot)
{
ds.row(epoch.epoch);
EpochReady ready = epoch.ready;
ds.column("ready_metadata", ready.metadata.value);
ds.column("ready_coordinate", ready.coordinate.value);
ds.column("ready_data", ready.data.value);
ds.column("ready_reads", ready.reads.value);
ds.column("ready", ready.reads == SUCCESS);
ds.row(epoch.epoch());
EpochReady ready = epoch.epochReady();
ds.column("ready_active", ready.active.toString());
ds.column("ready_coordinate", ready.coordinate.toString());
ds.column("ready_data", ready.data.toString());
ds.column("ready_reads", ready.reads.toString());
ds.column("ready", ready.active.isDone() && ready.coordinate.isDone() && ready.data.isDone() && ready.reads.isDone());
}
return ds;
}
@ -133,21 +131,21 @@ public class AccordVirtualTables
public DataSet data()
{
SimpleDataSet ds = new SimpleDataSet(metadata());
EpochsSnapshot snapshot = epochsSnapshot();
for (Epoch state : snapshot)
ActiveEpochs snapshot = AccordService.instance().topology().active();
for (ActiveEpoch state : snapshot)
{
Map<TableId, List<TokenRange>> addedRanges = groupByTable(state.addedRanges);
Map<TableId, List<TokenRange>> removedRanges = groupByTable(state.removedRanges);
Map<TableId, List<TokenRange>> synced = groupByTable(state.synced);
Map<TableId, List<TokenRange>> closed = groupByTable(state.closed);
Map<TableId, List<TokenRange>> retired = groupByTable(state.retired);
Map<TableId, List<TokenRange>> synced = groupByTable(state.quorumReady());
Map<TableId, List<TokenRange>> closed = groupByTable(state.closed());
Map<TableId, List<TokenRange>> retired = groupByTable(state.retired());
Set<TableId> allTables = union(addedRanges.keySet(), removedRanges.keySet(), synced.keySet(), closed.keySet(), retired.keySet());
for (TableId table : allTables)
{
TableMetadata metadata = Schema.instance.getTableMetadata(table);
if (metadata == null) continue; // table dropped, ignore
ds.row(state.epoch, metadata.keyspace, metadata.name);
ds.row(state.epoch(), metadata.keyspace, metadata.name);
ds.column("added", format(addedRanges.get(table)));
ds.column("removed", format(removedRanges.get(table)));
@ -179,11 +177,6 @@ public class AccordVirtualTables
}
}
private static EpochsSnapshot epochsSnapshot()
{
return AccordService.instance().topology().epochsSnapshot();
}
private static String toStringNoTable(TokenRange tr)
{
// TokenRange extends Range.EndInclusive

2
src/java/org/apache/cassandra/service/Rebuild.java
View File

@ -33,7 +33,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;

654
src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
View File

@ -1,654 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import accord.api.Agent;
import accord.api.TopologySorter;
import accord.coordinate.tracking.NotifyTracker;
import accord.coordinate.tracking.RequestStatus;
import accord.impl.AbstractConfigurationService;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Topologies;
import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.agrona.collections.LongArrayList;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@Simulate(with=MONITORS)
public class AccordConfigurationService extends AbstractConfigurationService<AccordConfigurationService.EpochState, AccordConfigurationService.EpochHistory> implements AccordSyncPropagator.Listener, Shutdownable
{
public static final Logger logger = LoggerFactory.getLogger(AccordConfigurationService.class);
private final AccordSyncPropagator syncPropagator;
private final WatermarkCollector watermarkCollector;
private final AccordEndpointMapper endpointMapper;
private enum State { INITIALIZED, STARTED, SHUTDOWN }
@GuardedBy("this")
private State state = State.INITIALIZED;
public enum SyncStatus { NOT_STARTED, NOTIFYING, COMPLETED }
static class EpochState extends AbstractConfigurationService.AbstractEpochState
{
private NotifyTracker notifyTracker;
private volatile SyncStatus syncStatus = SyncStatus.NOT_STARTED;
protected final AsyncResult.Settable<Void> localSyncNotified = AsyncResults.settable();
public EpochState(long epoch)
{
super(epoch);
}
void ack(Node.Id id)
{
if (notifyTracker != null && RequestStatus.Success == notifyTracker.recordSuccess(id))
doneNotifying();
}
void nack(Node.Id id)
{
if (notifyTracker != null && RequestStatus.Success == notifyTracker.recordFailure(id))
doneNotifying();
}
void startNotifying()
{
syncStatus = SyncStatus.NOTIFYING;
notifyTracker = new NotifyTracker(new Topologies.Single((TopologySorter) null, topology));
}
void doneNotifying()
{
syncStatus = SyncStatus.COMPLETED;
localSyncNotified.trySuccess(null);
notifyTracker = null;
}
AsyncResult<Topology> received()
{
return received;
}
AsyncResult<Void> acknowledged()
{
return acknowledged;
}
@Nullable AsyncResult<Void> reads()
{
return ready == null ? null : ready.reads;
}
AsyncResult.Settable<Void> localSyncNotified()
{
return localSyncNotified;
}
public void addDebugString(StringBuilder sb)
{
sb.append(" syncStatus ")
.append(syncStatus)
.append(" localSyncNotified ")
.append(localSyncNotified);
super.addDebugString(sb);
}
}
static class EpochHistory extends AbstractConfigurationService.AbstractEpochHistory<EpochState>
{
@Override
protected EpochState createEpochState(long epoch)
{
return new EpochState(epoch);
}
}
public AccordConfigurationService(Node.Id node, Agent agent, AccordEndpointMapper endpointMapper, MessageDelivery messagingService, ScheduledExecutorPlus scheduledTasks)
{
super(node, agent);
this.syncPropagator = new AccordSyncPropagator(localId, endpointMapper, messagingService, scheduledTasks, this);
this.watermarkCollector = new WatermarkCollector();
this.endpointMapper = endpointMapper;
listeners.add(watermarkCollector);
}
public AccordConfigurationService(Node.Id node, Agent agent)
{
this(node, agent, new EndpointMapping.Updateable(), MessagingService.instance(), ScheduledExecutors.scheduledTasks);
}
@Override
protected EpochHistory createEpochHistory()
{
return new EpochHistory();
}
/**
* On restart, loads topologies. On bootstrap, discovers existing topologies and initializes the node.
*/
public synchronized void start()
{
Invariants.require(state == State.INITIALIZED, "Expected state to be INITIALIZED but was %s", state);
state = State.STARTED;
endpointMapper.removedNodes().forEach((removed, removedIn) -> {
onNodeRemoved(removedIn, new Node.Id(removed.id));
});
}
@Override
public synchronized boolean isTerminated()
{
return state == State.SHUTDOWN;
}
@Override
public synchronized void shutdown()
{
if (isTerminated())
return;
state = State.SHUTDOWN;
}
@Override
public Object shutdownNow()
{
shutdown();
return null;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
{
return isTerminated();
}
public synchronized void updateMapping(ClusterMetadata metadata)
{
endpointMapper.updateMapping(metadata);
}
private void reportMetadata(ClusterMetadata metadata)
{
Stage.MISC.submit(() -> reportMetadataInternal(metadata));
}
void reportMetadataInternal(ClusterMetadata metadata)
{
Topology topology = AccordTopology.createAccordTopology(metadata);
updateMapping(metadata);
reportTopology(topology);
Set<Node.Id> stillLiveNodes = metadata.directory.states.entrySet()
.stream()
.filter(e -> e.getValue() != NodeState.LEFT && e.getValue() != NodeState.LEAVING)
.map(e -> tcmIdToAccord(e.getKey()))
.collect(Collectors.toSet());
if (epochs.lastAcknowledged() >= topology.epoch()) checkIfNodesRemoved(topology, stillLiveNodes);
else epochs.acknowledgeFuture(topology.epoch()).invokeIfSuccess(() -> checkIfNodesRemoved(topology, stillLiveNodes));
}
private void checkIfNodesRemoved(Topology topology, Set<Node.Id> stillLiveNodes)
{
long minEpoch = epochs.minEpoch();
if (minEpoch == 0 || topology.epoch() <= minEpoch) return;
Topology previous = getTopologyForEpoch(topology.epoch() - 1);
// for all nodes removed, or pending removal, mark them as removed so we don't wait on their replies
Set<Node.Id> removedNodes = Sets.difference(previous.nodes(), topology.nodes());
removedNodes = Sets.filter(removedNodes, id -> !stillLiveNodes.contains(id));
// TODO (desired, efficiency): there should be no need to notify every epoch for every removed node
for (Node.Id removedNode : removedNodes)
{
if (topology.epoch() >= minEpoch)
onNodeRemoved(topology.epoch(), removedNode);
}
}
public void onNodeRemoved(long epoch, Node.Id removed)
{
syncPropagator.onNodesRemoved(removed);
// TODO (desired): this is test only; make integration cleaner
for (long oldEpoch : nonCompletedEpochsBefore(epoch))
receiveRemoteSyncCompletePreListenerNotify(removed, oldEpoch);
listeners.forEach(l -> l.onRemoveNode(epoch, removed));
}
private long[] nonCompletedEpochsBefore(long max)
{
LongArrayList notComplete = new LongArrayList();
synchronized (epochs)
{
for (long epoch = epochs.minEpoch(), maxKnown = epochs.maxEpoch(); epoch <= max && epoch <= maxKnown; epoch++)
{
EpochSnapshot snapshot = getEpochSnapshot(epoch);
if (snapshot.syncStatus != SyncStatus.COMPLETED)
notComplete.add(epoch);
}
}
return notComplete.toLongArray();
}
@VisibleForTesting
void maybeReportMetadata(ClusterMetadata metadata)
{
// don't report metadata until the previous one has been acknowledged
long epoch = metadata.epoch.getEpoch();
synchronized (epochs)
{
// Accord has never been enabled for this cluster.
if (epochs.isEmpty() && !metadata.schema.hasAccordKeyspaces())
return;
// On first boot, we have 2 options:
//
// - we can start listening to TCM _before_ we replay topologies
// - we can start listening to TCM _after_ we replay topologies
//
// If we start listening to TCM _before_ we replay topologies from other nodes,
// we may end up in a situation where TCM reports metadata that would create an
// `epoch - 1` epoch state that is not associated with any topologies, and
// therefore should not be listened upon.
//
// If we start listening to TCM _after_ we replay topologies, we may end up in a
// situation where TCM reports metadata that is 1 (or more) epochs _ahead_ of the
// last known epoch. Previous implementations were using TCM peer catch up, which
// could have resulted in gaps.
//
// Current protocol solves both problems by _first_ replaying topologies form peers,
// then subscribing to TCM _and_, if there are still any gaps, filling them again.
// However, it still has a slight chance of creating an `epoch - 1` epoch state
// not associated with any topologies, which under "right" circumstances could
// have been waited upon with `epochReady`. This check precludes creation of this
// epoch: by the time this code can be called, remote topology replay is already
// done, so TCM listener will only report epochs that are _at least_ min epoch.
if (epochs.maxEpoch() == 0 || epochs.minEpoch() == metadata.epoch.getEpoch())
{
getOrCreateEpochState(epoch); // touch epoch state so subsequent calls see it
reportMetadata(metadata);
return;
}
}
getOrCreateEpochState(epoch - 1).acknowledged().invokeIfSuccess(() -> reportMetadata(metadata));
}
private final Map<Long, Future<Void>> pendingTopologies = new ConcurrentHashMap<>();
@Override
public void fetchTopologyForEpoch(long epoch)
{
long minEpoch = currentEpoch() + 1;
// Find and fetch all epochs in-between
for (long i = minEpoch; i <= epoch; ++i)
fetchTopologyInternal(i);
}
private static final Object SUCCESS = new Object();
protected void fetchTopologyInternal(long epoch)
{
pendingTopologies.computeIfAbsent(epoch, (epoch_) -> {
AsyncPromise<Void> future = new AsyncPromise<>();
fetchTopologyAsync(epoch_,
(success, throwable) -> {
Future<Void> removed = pendingTopologies.remove(epoch_);
Invariants.require(future == removed, "%s should be equal to %s", future, removed);
if (success != null)
future.setSuccess(null);
else
{
future.setFailure(Invariants.nonNull(throwable));
fetchTopologyForEpoch(epoch_);
}
});
return future;
});
}
private void fetchTopologyAsync(long epoch, BiConsumer<Object, ? super Throwable> onResult)
{
// It's not safe for this to block on CMS so for now pick a thread pool to handle it
Stage.ACCORD_MIGRATION.execute(() -> {
try
{
if (ClusterMetadata.current().epoch.getEpoch() < epoch)
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch));
}
catch (Throwable t)
{
onResult.accept(null, t);
return;
}
// In most cases, after fetching log from CMS, we will be caught up to the required epoch.
// This TCM will also notify Accord via reportMetadata, so we do not need to fetch topologies.
// If metadata has reported has skipped one or more epochs, and is _ahead_ of the requested epoch,
// we need to fetch topologies from peers to fill in the gap.
ClusterMetadata metadata = ClusterMetadata.current();
if (metadata.epoch.getEpoch() == epoch)
{
onResult.accept(SUCCESS, null);
return;
}
Set<InetAddressAndPort> peers = new HashSet<>(metadata.directory.allJoinedEndpoints());
peers.remove(FBUtilities.getBroadcastAddressAndPort());
if (peers.isEmpty())
{
onResult.accept(SUCCESS, null);
return;
}
// Fetching only one epoch here since later epochs might have already been requested concurrently
FetchTopologies.fetch(SharedContext.Global.instance, peers, epoch, epoch)
.addCallback((topologyRange, t) -> {
if (t != null)
{
if (currentEpoch() >= epoch)
onResult.accept(SUCCESS, null);
else
onResult.accept(null, t);
}
else
{
topologyRange.forEach(this::reportTopology, epoch, 1);
onResult.accept(SUCCESS, null);
}
});
});
}
@Override
protected Executor executor()
{
return Stage.ACCORD_MIGRATION::execute;
}
@Override
public void reportTopology(Topology topology)
{
long tcmEpoch = ClusterMetadata.current().epoch.getEpoch();
Invariants.require(topology.epoch() <= tcmEpoch,
"Reported topology %s not known to TCM", topology.epoch(), tcmEpoch);
super.reportTopology(topology);
}
@Override
protected void onReadyToCoordinate(Topology topology)
{
long epoch = topology.epoch();
EpochState epochState = getOrCreateEpochState(epoch);
synchronized (this)
{
if (epochState.syncStatus != SyncStatus.NOT_STARTED)
return;
if (topology.nodes().contains(localId)) epochState.startNotifying();
else epochState.doneNotifying();
}
syncPropagator.reportCoordinationReady(epoch, topology.nodes(), localId);
}
@Override
public synchronized void onEndpointAck(Node.Id id, long epoch)
{
if (epochs.wasTruncated(epoch))
return;
EpochState epochState = getOrCreateEpochState(epoch);
synchronized (this)
{
epochState.ack(id);
}
}
@Override
public synchronized void onEndpointNack(Node.Id id, long epoch)
{
if (epochs.wasTruncated(epoch))
return;
EpochState epochState = getOrCreateEpochState(epoch);
synchronized (this)
{
epochState.nack(id);
}
}
@Override
public void reportEpochClosed(Ranges ranges, long epoch)
{
checkStarted();
EpochHistory epochs = this.epochs;
if (epoch < minEpoch() || epochs.wasTruncated(epoch))
return;
Topology topology = getTopologyForEpoch(epoch);
if (topology != null)
syncPropagator.reportClosed(epoch, topology.nodes(), ranges);
}
@VisibleForTesting
public AccordSyncPropagator syncPropagator()
{
return syncPropagator;
}
@Override
public void reportEpochRetired(Ranges ranges, long epoch)
{
if (epochs.wasTruncated(epoch))
return;
checkStarted();
// TODO (expected): ensure we aren't fetching a truncated epoch; otherwise this should be non-null
Topology topology = getTopologyForEpoch(epoch);
if (topology != null)
syncPropagator.reportRetired(epoch, topology.nodes(), ranges);
}
@Override
public void receiveClosed(Ranges ranges, long epoch)
{
super.receiveClosed(ranges, epoch);
}
@Override
public void receiveRetired(Ranges ranges, long epoch)
{
super.receiveRetired(ranges, epoch);
}
@Override
public void reportEpochRemoved(long epoch)
{
logger.info("Epoch removed, truncated epochs until {}", epoch);
epochs.truncateUntil(epoch);
}
private synchronized void checkStarted()
{
State state = this.state;
Invariants.require(state == State.STARTED, "Expected state to be STARTED but was %s", state);
}
public WatermarkCollector watermarkCollector()
{
return watermarkCollector;
}
public AccordEndpointMapper endpointMapper()
{
return endpointMapper;
}
@VisibleForTesting
public static class EpochSnapshot
{
public enum ResultStatus
{
PENDING, SUCCESS, FAILURE;
static ResultStatus of(AsyncResult<?> result)
{
if (result == null || !result.isDone())
return PENDING;
return result.isSuccess() ? SUCCESS : FAILURE;
}
}
public final long epoch;
public final SyncStatus syncStatus;
public final ResultStatus received;
public final ResultStatus acknowledged;
public final ResultStatus reads;
private EpochSnapshot(EpochState state)
{
this.epoch = state.epoch();
this.syncStatus = state.syncStatus;
this.received = ResultStatus.of(state.received());
this.acknowledged = ResultStatus.of(state.acknowledged());
this.reads = ResultStatus.of(state.reads());
}
public EpochSnapshot(long epoch, SyncStatus syncStatus, ResultStatus received, ResultStatus acknowledged, ResultStatus reads)
{
this.epoch = epoch;
this.syncStatus = syncStatus;
this.received = received;
this.acknowledged = acknowledged;
this.reads = reads;
}
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
EpochSnapshot that = (EpochSnapshot) o;
return epoch == that.epoch && syncStatus == that.syncStatus && received == that.received && acknowledged == that.acknowledged && reads == that.reads;
}
public int hashCode()
{
return Objects.hash(epoch, syncStatus, received, acknowledged, reads);
}
public String toString()
{
return "EpochSnapshot{" +
"epoch=" + epoch +
", syncStatus=" + syncStatus +
", received=" + received +
", acknowledged=" + acknowledged +
", reads=" + reads +
'}';
}
public static EpochSnapshot completed(long epoch)
{
return new EpochSnapshot(epoch, SyncStatus.COMPLETED, ResultStatus.SUCCESS, ResultStatus.SUCCESS, ResultStatus.SUCCESS);
}
}
@VisibleForTesting
public EpochSnapshot getEpochSnapshot(long epoch)
{
EpochState state;
// If epoch truncate happens then getting the epoch again will recreate an empty one
synchronized (epochs)
{
if (epoch < epochs.minEpoch() || epoch > epochs.maxEpoch())
return null;
state = getOrCreateEpochState(epoch);
}
return new EpochSnapshot(state);
}
@VisibleForTesting
public long minEpoch()
{
return epochs.minEpoch();
}
@VisibleForTesting
public long maxEpoch()
{
return epochs.maxEpoch();
}
/**
* The callback is resolved while holding the object lock, which can cause the future chain to resolve while also
* holding the lock! This behavior is exposed for tests and is unsafe due to the lock behind held while resolving
* the callback
*/
@VisibleForTesting
public Future<Void> unsafeLocalSyncNotified(long epoch)
{
AsyncPromise<Void> promise = new AsyncPromise<>();
getOrCreateEpochState(epoch).localSyncNotified().invoke((result, failure) -> {
if (failure != null) promise.tryFailure(failure);
else promise.trySuccess(result);
});
return promise;
}
}

29
src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java
View File

@ -18,9 +18,8 @@
package org.apache.cassandra.service.accord;
import accord.api.ConfigurationService;
import accord.api.TopologyListener;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.async.AsyncResult;
@ -50,7 +49,7 @@ import java.util.concurrent.TimeUnit;
/**
* Listens to availability status of peers and updates tcm fast path data accordingly
*/
public abstract class AccordFastPathCoordinator implements ChangeListener, ConfigurationService.Listener
public abstract class AccordFastPathCoordinator implements ChangeListener, TopologyListener
{
private static final AsyncResult<Void> SUCCESS = AsyncResults.success(null);
@ -182,12 +181,11 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
private static class Impl extends AccordFastPathCoordinator implements IEndpointStateChangeSubscriber
{
private final AccordConfigurationService configService;
public Impl(Node.Id localId, AccordConfigurationService configService)
final AccordEndpointMapper endpointMapper;
public Impl(Node.Id localId, AccordEndpointMapper endpointMapper)
{
super(localId);
this.configService = configService;
this.endpointMapper = endpointMapper;
}
@Override
@ -201,7 +199,7 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
{
Gossiper.instance.register(this);
StorageService.instance.addPreShutdownHook(this::onShutdown);
configService.registerListener(this);
AccordService.unsafeInstance().topology().addListener(this);
}
@Override
@ -222,21 +220,21 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
@Override
public void onAlive(InetAddressAndPort endpoint, EndpointState state)
{
Node.Id node = configService.endpointMapper().mappedIdOrNull(endpoint);
Node.Id node = endpointMapper.mappedIdOrNull(endpoint);
if (node != null) onAlive(node);
}
@Override
public void onDead(InetAddressAndPort endpoint, EndpointState state)
{
Node.Id node = configService.endpointMapper().mappedIdOrNull(endpoint);
Node.Id node = endpointMapper.mappedIdOrNull(endpoint);
if (node != null) onDead(node);
}
}
public static AccordFastPathCoordinator create(Node.Id localId, AccordConfigurationService configService)
public static AccordFastPathCoordinator create(Node.Id localId, AccordEndpointMapper endpointMapper)
{
return new Impl(localId, configService);
return new Impl(localId, endpointMapper);
}
synchronized void maybeUpdateFastPath(Node.Id node, Status status)
@ -335,13 +333,8 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
}
@Override
public AsyncResult<Void> onTopologyUpdate(Topology topology)
public void onReceived(Topology topology)
{
updatePeers(topology);
return SUCCESS;
}
@Override public void onRemoteSyncComplete(Node.Id node, long epoch) {}
@Override public void onEpochClosed(Ranges ranges, long epoch) {}
@Override public void onEpochRetired(Ranges ranges, long epoch) {}
}

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

@ -38,7 +38,8 @@ import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.ActiveEpochs;
import accord.topology.EpochReady;
import accord.primitives.Txn;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.metrics.AccordSystemMetrics;
@ -53,7 +54,6 @@ import accord.api.ProtocolModifiers;
import accord.coordinate.CoordinateMaxConflict;
import accord.coordinate.CoordinateTransaction;
import accord.coordinate.KeyBarriers;
import accord.impl.AbstractConfigurationService;
import accord.impl.DefaultLocalListeners;
import accord.impl.DefaultRemoteListeners;
import accord.impl.RequestCallbacks;
@ -77,6 +77,7 @@ import accord.primitives.TxnId;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.topology.TopologyRange;
import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@ -133,7 +134,6 @@ import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Write;
import static accord.topology.TopologyManager.TopologyRange;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordCommandStoreShardCount;
@ -219,7 +219,8 @@ public class AccordService implements IAccordService, Shutdownable
private final Node node;
private final Shutdownable nodeShutdown;
private final AccordMessageSink messageSink;
private final AccordConfigurationService configService;
private final AccordEndpointMapper endpointMapper;
private final AccordTopologyService topologyService;
private final AccordFastPathCoordinator fastPathCoordinator;
private final AccordScheduler scheduler;
private final AccordDataStore dataStore;
@ -258,7 +259,7 @@ public class AccordService implements IAccordService, Shutdownable
{
if (!isSetup()) return ignore -> {};
AccordService i = (AccordService) instance();
return i.configService().watermarkCollector().handler;
return i.topologyService().watermarkCollector().handler;
}
public static IVerbHandler<? extends Request> requestHandlerOrNoop()
@ -300,8 +301,6 @@ public class AccordService implements IAccordService, Shutdownable
as.finishInitialization();
as.configService.start();
as.configService.unsafeMarkTruncated();
as.fastPathCoordinator.start();
ClusterMetadataService.instance().log().addListener(as.fastPathCoordinator);
@ -311,7 +310,6 @@ public class AccordService implements IAccordService, Shutdownable
as.node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
as.state = State.STARTED;
// Only enable durability scheduling _after_ we have fully replayed journal
as.configService.registerListener(as.node.durability());
as.node.durability().start();
instance = as;
@ -320,7 +318,7 @@ public class AccordService implements IAccordService, Shutdownable
AccordSystemMetrics.touch();
AccordViolationHandler.setup();
WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);
WatermarkCollector.fetchAndReportWatermarksAsync(as.topology());
return as;
}
@ -383,12 +381,13 @@ public class AccordService implements IAccordService, Shutdownable
this.scheduler = new AccordScheduler();
this.dataStore = new AccordDataStore();
this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal);
this.configService = new AccordConfigurationService(localId, agent);
this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, configService);
this.messageSink = new AccordMessageSink(agent, configService.endpointMapper(), callbacks);
this.endpointMapper = new EndpointMapping.Updateable();
this.topologyService = new AccordTopologyService(localId, endpointMapper);
this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, endpointMapper);
this.messageSink = new AccordMessageSink(agent, endpointMapper, callbacks);
this.node = new Node(localId,
messageSink,
configService,
topologyService,
time, new AtomicUniqueTimeWithStaleReservation(time),
() -> dataStore,
new KeyspaceSplitter(new EvenSplit<>(getAccordCommandStoreShardCount(), getPartitioner().accordSplitter())),
@ -396,18 +395,18 @@ public class AccordService implements IAccordService, Shutdownable
new DefaultRandom(),
scheduler,
CompositeTopologySorter.create(SizeOfIntersectionSorter.SUPPLIER,
new AccordTopologySorter.Supplier(configService.endpointMapper(), DatabaseDescriptor.getNodeProximity())),
new AccordTopologySorter.Supplier(endpointMapper, DatabaseDescriptor.getNodeProximity())),
DefaultRemoteListeners::new,
ignore -> callbacks,
DefaultProgressLogs::new,
DefaultLocalListeners.Factory::new,
AccordCommandStores.factory(),
new AccordInteropFactory(configService.endpointMapper()),
new AccordInteropFactory(endpointMapper),
journal.durableBeforePersister(),
journal);
this.nodeShutdown = toShutdownable(node);
this.requestHandler = new AccordVerbHandler<>(node, configService.endpointMapper());
this.responseHandler = new AccordResponseVerbHandler<>(callbacks, configService.endpointMapper());
this.requestHandler = new AccordVerbHandler<>(node, endpointMapper);
this.responseHandler = new AccordResponseVerbHandler<>(callbacks, endpointMapper);
}
@Override
@ -419,7 +418,7 @@ public class AccordService implements IAccordService, Shutdownable
node.load();
ClusterMetadata metadata = ClusterMetadata.current();
configService.updateMapping(metadata);
endpointMapper.updateMapping(metadata);
List<TopologyUpdate> images = journal.replayTopologies();
if (!images.isEmpty())
@ -435,7 +434,7 @@ public class AccordService implements IAccordService, Shutdownable
// Replay local epochs
for (TopologyUpdate image : images)
configService.reportTopology(image.global);
node.topology().reportTopology(image.global);
}
}
}
@ -448,17 +447,20 @@ public class AccordService implements IAccordService, Shutdownable
@VisibleForTesting
public void finishInitialization()
{
configService.updateMapping(ClusterMetadata.current());
endpointMapper.updateMapping(ClusterMetadata.current());
TopologyManager topology = node.topology();
long highestKnown = -1;
if (configService.currentTopology() != null)
highestKnown = configService.currentEpoch();
if (!topology.active().isEmpty())
{
highestKnown = topology.active().epoch();
}
try
{
TopologyRange remote = fetchTopologies(highestKnown + 1);
if (remote != null) // TODO (required): if remote.min > highestKnown + 1, should we decide if we need to truncate our local topologies? Probably not until startup has finished.
{
remote.forEach(configService::reportTopology, remote.min, Integer.MAX_VALUE);
remote.forEach(node.topology()::reportTopology, remote.min, Integer.MAX_VALUE);
if (remote.current > highestKnown)
highestKnown = remote.current;
}
@ -470,7 +472,7 @@ public class AccordService implements IAccordService, Shutdownable
public void notifyPostCommit(ClusterMetadata prev, ClusterMetadata next, boolean fromSnapshot)
{
if (state != State.SHUTDOWN)
configService.maybeReportMetadata(next);
maybeReportMetadata(next);
}
});
@ -481,7 +483,7 @@ public class AccordService implements IAccordService, Shutdownable
for (ClusterMetadata item : preinit.getItems())
{
if (item.epoch.getEpoch() > highestKnown)
configService.maybeReportMetadata(item);
maybeReportMetadata(item);
}
}
catch (InterruptedException e)
@ -494,6 +496,45 @@ public class AccordService implements IAccordService, Shutdownable
throw new RuntimeException(e);
}
}
private void maybeReportMetadata(ClusterMetadata metadata)
{
endpointMapper.updateMapping(metadata);
ActiveEpochs epochs = topology().active();
if (!metadata.schema.hasAccordKeyspaces() && epochs.isEmpty())
return;
if (epochs.epoch() >= metadata.epoch.getEpoch())
return;
// On first boot, we have 2 options:
//
// - we can start listening to TCM _before_ we replay topologies
// - we can start listening to TCM _after_ we replay topologies
//
// If we start listening to TCM _before_ we replay topologies from other nodes,
// we may end up in a situation where TCM reports metadata that would create an
// `epoch - 1` epoch state that is not associated with any topologies, and
// therefore should not be listened upon.
//
// If we start listening to TCM _after_ we replay topologies, we may end up in a
// situation where TCM reports metadata that is 1 (or more) epochs _ahead_ of the
// last known epoch. Previous implementations were using TCM peer catch up, which
// could have resulted in gaps.
//
// Current protocol solves both problems by _first_ replaying topologies form peers,
// then subscribing to TCM _and_, if there are still any gaps, filling them again.
// However, it still has a slight chance of creating an `epoch - 1` epoch state
// not associated with any topologies, which under "right" circumstances could
// have been waited upon with `epochReady`. This check precludes creation of this
// epoch: by the time this code can be called, remote topology replay is already
// done, so TCM listener will only report epochs that are _at least_ min epoch.
Topology topology = AccordTopology.createAccordTopology(metadata);
topology().reportTopology(topology);
}
/**
* Queries peers to discover min epoch, and then fetches all topologies between min and current epochs
*/
@ -723,10 +764,9 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public long currentEpoch()
{
return configService.currentEpoch();
return topology().epoch();
}
@Override
public TopologyManager topology()
{
@ -859,7 +899,7 @@ public class AccordService implements IAccordService, Shutdownable
private List<Shutdownable> shutdownableSubsystems()
{
return Arrays.asList(scheduler, nodeShutdown, journal, configService);
return Arrays.asList(scheduler, nodeShutdown, journal, topologyService);
}
@VisibleForTesting
@ -906,7 +946,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public Future<Void> epochReady(Epoch epoch, Function<EpochReady, AsyncResult<Void>> get)
{
return toFuture(configService.epochReady(epoch.getEpoch(), get));
return toFuture(topology().epochReady(epoch.getEpoch(), get));
}
@Override
@ -921,18 +961,18 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public void receive(Message<Notification> message)
{
receive(MessagingService.instance(), configService, message);
receive(MessagingService.instance(), node.topology(), message);
}
@VisibleForTesting
public static void receive(MessageDelivery sink, AbstractConfigurationService<?, ?> configService, Message<Notification> message)
public static void receive(MessageDelivery sink, TopologyManager topologyManager, Message<Notification> message)
{
AccordSyncPropagator.Notification notification = message.payload;
notification.syncComplete.forEach(id -> configService.receiveRemoteSyncComplete(id, notification.epoch));
notification.readyToCoordinate.forEach(id -> topologyManager.onReadyToCoordinate(id, notification.epoch));
if (!notification.closed.isEmpty())
configService.receiveClosed(notification.closed, notification.epoch);
topologyManager.onEpochClosed(notification.closed, notification.epoch);
if (!notification.retired.isEmpty())
configService.receiveRetired(notification.retired, notification.epoch);
topologyManager.onEpochRetired(notification.retired, notification.epoch);
sink.respond(Ok, message);
}
@ -974,9 +1014,15 @@ public class AccordService implements IAccordService, Shutdownable
}
@VisibleForTesting
public AccordConfigurationService configService()
public AccordEndpointMapper endpointMapper()
{
return configService;
return endpointMapper;
}
@Override
public AccordTopologyService topologyService()
{
return topologyService;
}
@Override

129
src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
View File

@ -19,22 +19,24 @@
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.TopologyListener;
import accord.local.Node;
import accord.messages.SimpleReply;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.utils.Invariants;
import accord.utils.SortedList;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.SortedListSet;
import accord.utils.UnhandledEnum;
import org.agrona.collections.Int2ObjectHashMap;
@ -63,7 +65,7 @@ import org.apache.cassandra.utils.NoSpamLogger;
* Notifies remote replicas that the local replica has synchronised coordination
* information for this epoch.
*/
public class AccordSyncPropagator
public class AccordSyncPropagator implements TopologyListener
{
private static final Logger logger = LoggerFactory.getLogger(AccordSyncPropagator.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES);
@ -74,10 +76,9 @@ public class AccordSyncPropagator
AccordService.instance().receive(message);
};
interface Listener
interface TestListener
{
void onEndpointAck(Node.Id id, long epoch);
void onEndpointNack(Node.Id id, long epoch);
}
private interface ReportPending<T>
@ -136,10 +137,10 @@ public class AccordSyncPropagator
boolean ack(Notification notification)
{
if (!notification.syncComplete.isEmpty())
if (!notification.readyToCoordinate.isEmpty())
{
if (notification.syncComplete.containsAll(syncComplete)) syncComplete = ImmutableSet.of();
else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.syncComplete.contains(v)));
if (notification.readyToCoordinate.containsAll(syncComplete)) syncComplete = ImmutableSet.of();
else syncComplete = ImmutableSet.copyOf(Iterables.filter(syncComplete, v -> !notification.readyToCoordinate.contains(v)));
}
closed = closed.without(notification.closed);
retired = retired.without(notification.retired);
@ -186,21 +187,24 @@ public class AccordSyncPropagator
}
private final PendingNodes pending = new PendingNodes();
private final Node.Id localId;
private final Node.Id self;
private final AccordEndpointMapper endpointMapper;
private final MessageDelivery messagingService;
private final ScheduledExecutorPlus scheduler;
private final Listener listener;
private TestListener listener;
private final ConcurrentHashMap<RetryKey, Notification> retryingNotifications = new ConcurrentHashMap<>();
public AccordSyncPropagator(Node.Id localId, AccordEndpointMapper endpointMapper,
MessageDelivery messagingService, ScheduledExecutorPlus scheduler,
Listener listener)
public AccordSyncPropagator(Node.Id self, AccordEndpointMapper endpointMapper,
MessageDelivery messagingService, ScheduledExecutorPlus scheduler)
{
this.localId = localId;
this.self = self;
this.endpointMapper = endpointMapper;
this.messagingService = messagingService;
this.scheduler = scheduler;
}
void setTestListener(TestListener listener)
{
this.listener = listener;
}
@ -222,52 +226,59 @@ public class AccordSyncPropagator
public String toString()
{
return "AccordSyncPropagator{" +
"localId=" + localId +
"localId=" + self +
", pending=" + pending +
'}';
}
public void onNodesRemoved(Node.Id removed)
public void onNodesRemoved(SortedArrayList<Node.Id> removed)
{
long[] toAck;
synchronized (AccordSyncPropagator.this)
{
PendingEpochs pendingEpochs = pending.remove(removed.id);
if (pendingEpochs == null) return;
toAck = new long[pendingEpochs.size()];
Long2ObjectHashMap<PendingEpoch>.KeyIterator it = pendingEpochs.keySet().iterator();
for (int i = 0; it.hasNext(); i++)
{
long epoch = it.nextLong();
toAck[i] = epoch;
}
Arrays.sort(toAck);
}
for (int i = 0; i < toAck.length; i++)
{
long epoch = toAck[i];
listener.onEndpointAck(removed, epoch);
for (Node.Id id : removed)
pending.remove(id.id);
}
}
public void reportCoordinationReady(long epoch, SortedList<Node.Id> notify, Node.Id syncCompleteId)
@Override
public void onReadyToCoordinate(Topology topology)
{
SortedListSet<Node.Id> remaining = SortedListSet.allOf(notify);
if (remaining.remove(localId))
listener.onEndpointAck(localId, epoch);
report(epoch, remaining, PendingEpoch::syncComplete, syncCompleteId);
onReadyToCoordinate(topology.epoch(), topology.nodes());
}
public void reportClosed(long epoch, Collection<Node.Id> notify, Ranges closed)
@VisibleForTesting
void onReadyToCoordinate(long epoch, SortedArrayList<Node.Id> nodes)
{
report(epoch, notify, PendingEpoch::closed, closed);
SortedListSet<Node.Id> remaining = SortedListSet.allOf(nodes);
if (remaining.remove(self) && listener != null)
listener.onEndpointAck(self, epoch);
report(epoch, remaining, PendingEpoch::syncComplete, self);
}
public void reportRetired(long epoch, Collection<Node.Id> notify, Ranges retired)
@Override
public void onEpochClosed(Ranges ranges, long epoch, Topology topology)
{
report(epoch, notify, PendingEpoch::retired, retired);
if (topology != null)
onEpochClosed(ranges, epoch, topology.nodes());
}
@VisibleForTesting
void onEpochClosed(Ranges ranges, long epoch, Collection<Node.Id> nodes)
{
report(epoch, nodes, PendingEpoch::closed, ranges);
}
@Override
public void onEpochRetired(Ranges ranges, long epoch, Topology topology)
{
if (topology != null)
onEpochRetired(ranges, epoch, topology.nodes());
}
@VisibleForTesting
void onEpochRetired(Ranges ranges, long epoch, Collection<Node.Id> nodes)
{
report(epoch, nodes, PendingEpoch::retired, ranges);
}
private <T> void report(long epoch, Collection<Node.Id> notify, ReportPending<T> report, T param)
@ -291,7 +302,7 @@ public class AccordSyncPropagator
private void scheduleRetry(Node.Id to, Notification notification)
{
Notification retry = new Notification(notification.epoch, notification.syncComplete, notification.closed, notification.retired, notification.attempts + 1);
Notification retry = new Notification(notification.epoch, notification.readyToCoordinate, notification.closed, notification.retired, notification.attempts + 1);
RetryKey key = new RetryKey(to, notification.epoch);
retryingNotifications.compute(key, (k, cur) -> {
if (cur == null)
@ -332,7 +343,6 @@ public class AccordSyncPropagator
case UNKNOWN:
// endpoint is not a member of the latest epoch
pending.ack(to, notification);
listener.onEndpointNack(to, notification.epoch);
return true;
case HEALTHY:
@ -349,7 +359,8 @@ public class AccordSyncPropagator
}
long epoch = notification.epoch;
listener.onEndpointAck(to, epoch);
if (listener != null && notification.readyToCoordinate.contains(self))
listener.onEndpointAck(to, epoch);
}
@Override
@ -377,7 +388,7 @@ public class AccordSyncPropagator
public void serialize(Notification notification, DataOutputPlus out) throws IOException
{
out.writeLong(notification.epoch);
CollectionSerializers.serializeCollection(notification.syncComplete, out, TopologySerializers.nodeId);
CollectionSerializers.serializeCollection(notification.readyToCoordinate, out, TopologySerializers.nodeId);
KeySerializers.ranges.serialize(notification.closed, out);
KeySerializers.ranges.serialize(notification.retired, out);
}
@ -395,26 +406,26 @@ public class AccordSyncPropagator
public long serializedSize(Notification notification)
{
return TypeSizes.LONG_SIZE
+ CollectionSerializers.serializedCollectionSize(notification.syncComplete, TopologySerializers.nodeId)
+ CollectionSerializers.serializedCollectionSize(notification.readyToCoordinate, TopologySerializers.nodeId)
+ KeySerializers.ranges.serializedSize(notification.closed)
+ KeySerializers.ranges.serializedSize(notification.retired);
}
};
final long epoch;
final Collection<Node.Id> syncComplete;
final Collection<Node.Id> readyToCoordinate;
final Ranges closed, retired;
final int attempts;
public Notification(long epoch, Collection<Node.Id> syncComplete, Ranges closed, Ranges retired)
public Notification(long epoch, Collection<Node.Id> readyToCoordinate, Ranges closed, Ranges retired)
{
this(epoch, syncComplete, closed, retired, 0);
this(epoch, readyToCoordinate, closed, retired, 0);
}
public Notification(long epoch, Collection<Node.Id> syncComplete, Ranges closed, Ranges retired, int attempts)
public Notification(long epoch, Collection<Node.Id> readyToCoordinate, Ranges closed, Ranges retired, int attempts)
{
this.epoch = epoch;
this.syncComplete = syncComplete;
this.readyToCoordinate = readyToCoordinate;
this.closed = closed;
this.retired = retired;
this.attempts = attempts;
@ -424,8 +435,8 @@ public class AccordSyncPropagator
{
Invariants.require(add.epoch == this.epoch);
Collection<Node.Id> syncComplete = ImmutableSet.<Node.Id>builder()
.addAll(this.syncComplete)
.addAll(add.syncComplete)
.addAll(this.readyToCoordinate)
.addAll(add.readyToCoordinate)
.build();
return new Notification(epoch, syncComplete, closed.with(add.closed), retired.with(add.retired), Math.max(add.attempts, this.attempts));
}
@ -435,7 +446,7 @@ public class AccordSyncPropagator
{
return "Notification{" +
"epoch=" + epoch +
", syncComplete=" + syncComplete +
", syncComplete=" + readyToCoordinate +
", closed=" + closed +
", retired=" + retired +
'}';
@ -468,5 +479,11 @@ public class AccordSyncPropagator
RetryKey that = (RetryKey) obj;
return that.epoch == this.epoch && that.to.equals(this.to);
}
@Override
public String toString()
{
return epoch + "@" + to;
}
}
}

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

@ -34,10 +34,10 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import accord.api.ConfigurationService.EpochReady;
import accord.local.Node;
import accord.local.Node.Id;
import accord.primitives.Ranges;
import accord.topology.EpochReady;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.utils.Invariants;

204
src/java/org/apache/cassandra/service/accord/AccordTopologyService.java Normal file
View File

@ -0,0 +1,204 @@
/*
* 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.service.accord;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import accord.api.TopologyListener;
import accord.api.TopologyService;
import accord.local.Node;
import accord.topology.Topology;
import accord.topology.TopologyRetiredException;
import accord.utils.Invariants;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults.SettableByCallback;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Simulate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@Simulate(with=MONITORS)
public class AccordTopologyService implements TopologyService, Shutdownable, TopologyListener
{
public static final Logger logger = LoggerFactory.getLogger(AccordTopologyService.class);
// TODO (expected): move syncPropagator and watermarkCollector out of this class (and merge them)
private final AccordSyncPropagator syncPropagator;
private final WatermarkCollector watermarkCollector;
private SortedArrayList<Node.Id> previouslyRemovedIds = SortedArrayList.ofSorted();
private enum State { INITIALIZED, STARTED, SHUTDOWN }
@GuardedBy("this")
private State state = State.INITIALIZED;
public AccordTopologyService(Node.Id node, AccordEndpointMapper endpointMapper, MessageDelivery messagingService, ScheduledExecutorPlus scheduledTasks)
{
this.syncPropagator = new AccordSyncPropagator(node, endpointMapper, messagingService, scheduledTasks);
this.watermarkCollector = new WatermarkCollector();
}
public AccordTopologyService(Node.Id node, AccordEndpointMapper endpointMapper)
{
this(node, endpointMapper, MessagingService.instance(), ScheduledExecutors.scheduledTasks);
}
/**
* On restart, loads topologies. On bootstrap, discovers existing topologies and initializes the node.
*/
public void onStartup(Node node)
{
SortedArrayList<Node.Id> removed = node.topology().current().removedIds();
synchronized (this)
{
Invariants.require(state == State.INITIALIZED, "Expected state to be INITIALIZED but was %s", state);
state = State.STARTED;
previouslyRemovedIds = removed;
}
node.topology().addListener(watermarkCollector);
node.topology().addListener(syncPropagator);
syncPropagator.onNodesRemoved(removed);
}
@Override
public synchronized boolean isTerminated()
{
return state == State.SHUTDOWN;
}
@Override
public synchronized void shutdown()
{
if (isTerminated())
return;
state = State.SHUTDOWN;
}
@Override
public Object shutdownNow()
{
shutdown();
return null;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
{
return isTerminated();
}
@Override
public void onReceived(Topology topology)
{
SortedArrayList<Node.Id> newlyRemoved;
synchronized (this)
{
newlyRemoved = topology.removedIds().without(previouslyRemovedIds);
previouslyRemovedIds = topology.removedIds().with(previouslyRemovedIds);
}
syncPropagator.onNodesRemoved(newlyRemoved);
}
@Override
public AsyncResult<Topology> fetchTopologyForEpoch(long epoch)
{
SettableByCallback<Topology> result = new SettableByCallback<>();
fetchTopologyAsync(epoch, result);
return result;
}
private void fetchTopologyAsync(long epoch, BiConsumer<? super Topology, ? super Throwable> onResult)
{
// It's not safe for this to block on CMS so for now pick a thread pool to handle it
Stage.ACCORD_MIGRATION.execute(() -> {
ClusterMetadata metadata = ClusterMetadata.current();
try
{
if (metadata.epoch.getEpoch() < epoch)
ClusterMetadataService.instance().fetchLogFromCMS(Epoch.create(epoch));
}
catch (Throwable t)
{
onResult.accept(null, t);
return;
}
// In most cases, after fetching log from CMS, we will be caught up to the required epoch.
// This TCM will also notify Accord via reportMetadata, so we do not need to fetch topologies.
// If metadata has reported has skipped one or more epochs, and is _ahead_ of the requested epoch,
// we need to fetch topologies from peers to fill in the gap.
if (metadata.epoch.getEpoch() == epoch)
{
Topology topology = AccordTopology.createAccordTopology(metadata);
onResult.accept(topology, null);
return;
}
Set<InetAddressAndPort> peers = new HashSet<>(metadata.directory.allJoinedEndpoints());
peers.remove(FBUtilities.getBroadcastAddressAndPort());
if (peers.isEmpty())
{
onResult.accept(null, new TopologyRetiredException("No joined nodes to query; latest epoch was not the one requested", null));
return;
}
// Fetching only one epoch here since later epochs might have already been requested concurrently
FetchTopologies.fetch(SharedContext.Global.instance, peers, epoch, epoch)
.addCallback((success, fail) -> {
if (fail != null) onResult.accept(null, fail);
else if (success.hasEpoch(epoch)) onResult.accept(success.get(epoch), null);
else if (success.min > epoch) onResult.accept(null, new TopologyRetiredException("Could not fetch epoch " + epoch + " from peers; too far ahead", null));
else onResult.accept(null, new RuntimeException("Could not yet retrieve epoch " + epoch));
});
});
}
@VisibleForTesting
public AccordSyncPropagator syncPropagator()
{
return syncPropagator;
}
public WatermarkCollector watermarkCollector()
{
return watermarkCollector;
}
}

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

@ -68,7 +68,7 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
}
long waitForEpoch = request.waitForEpoch();
if (node.topology().hasAtLeastEpoch(waitForEpoch))
if (node.topology().active().hasAtLeastEpoch(waitForEpoch))
{
request.process(node, fromNodeId, message.header);
}

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

@ -28,6 +28,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.topology.Topology;
import accord.topology.TopologyRange;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.io.UnversionedSerializer;
@ -43,7 +45,6 @@ import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.utils.concurrent.Future;
import static accord.topology.TopologyManager.TopologyRange;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryFetchTopology;
/**
@ -153,7 +154,7 @@ public class FetchTopologies
return;
}
TopologyRange topologies = AccordService.instance().topology().between(message.payload.minEpoch, message.payload.maxEpoch);
TopologyRange topologies = AccordService.instance().topology().active().between(message.payload.minEpoch, message.payload.maxEpoch);
logger.debug("Responding with {} failure to {}", topologies, message.payload);
MessagingService.instance().respond(topologies, message);
};

26
src/java/org/apache/cassandra/service/accord/IAccordService.java
View File

@ -28,7 +28,6 @@ import java.util.function.Function;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import accord.api.ConfigurationService.EpochReady;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.slf4j.Logger;
@ -48,6 +47,7 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.topology.EpochReady;
import accord.topology.TopologyManager;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@ -180,7 +180,9 @@ public interface IAccordService
void awaitDone(TableId id, long epoch);
AccordConfigurationService configService();
AccordEndpointMapper endpointMapper();
AccordTopologyService topologyService();
Params journalConfiguration();
@ -350,9 +352,15 @@ public interface IAccordService
}
@Override
public AccordConfigurationService configService()
public AccordEndpointMapper endpointMapper()
{
return null;
throw new UnsupportedOperationException();
}
@Override
public AccordTopologyService topologyService()
{
throw new UnsupportedOperationException();
}
@Override
@ -420,9 +428,15 @@ public interface IAccordService
}
@Override
public AccordConfigurationService configService()
public AccordEndpointMapper endpointMapper()
{
return delegate.configService();
return delegate.endpointMapper();
}
@Override
public AccordTopologyService topologyService()
{
return delegate.topologyService();
}
@Nonnull

32
src/java/org/apache/cassandra/service/accord/WatermarkCollector.java
View File

@ -28,19 +28,21 @@ import java.util.Objects;
import java.util.Set;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterators;
import com.google.common.primitives.Ints;
import accord.api.ConfigurationService;
import accord.api.TopologyListener;
import accord.local.Node;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.utils.Invariants;
import accord.utils.ReducingRangeMap;
import accord.utils.async.AsyncResult;
import org.agrona.collections.Long2LongHashMap;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.UnversionedSerializer;
@ -62,7 +64,7 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retry
/**
* Collects watermarks of closed and retired epochs per range, and synced epochs per node.
*/
public class WatermarkCollector implements ConfigurationService.Listener
public class WatermarkCollector implements TopologyListener
{
private static final Comparator<Map.Entry<Range, Long>> sortByEpochThenRange = (a, b) -> {
int c = Long.compareUnsigned(a.getValue(), b.getValue());
@ -81,30 +83,25 @@ public class WatermarkCollector implements ConfigurationService.Listener
synced = new Long2LongHashMap(-1);
}
@Override public AsyncResult<Void> onTopologyUpdate(Topology topology)
{
return null;
}
@Override
public synchronized void onRemoteSyncComplete(Node.Id node, long epoch)
public synchronized void onRemoteReadyToCoordinate(Node.Id node, long epoch)
{
synced.compute(node.id, (k, prev) -> prev == -1 ? epoch : Long.max(prev, epoch));
}
@Override
public synchronized void onEpochClosed(Ranges ranges, long epoch)
public synchronized void onEpochClosed(Ranges ranges, long epoch, @Nullable Topology topology)
{
closed = ReducingRangeMap.merge(closed, ReducingRangeMap.create(ranges, epoch), Long::max);
}
@Override
public synchronized void onEpochRetired(Ranges ranges, long epoch)
public synchronized void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology)
{
retired = ReducingRangeMap.merge(retired, ReducingRangeMap.create(ranges, epoch), Long::max);
}
public final IVerbHandler<Void> handler = new IVerbHandler<Void>()
public final IVerbHandler<Void> handler = new IVerbHandler<>()
{
public void doVerb(Message<Void> message)
{
@ -123,7 +120,7 @@ public class WatermarkCollector implements ConfigurationService.Listener
};
@VisibleForTesting
static void fetchAndReportWatermarksAsync(AccordConfigurationService configService)
static void fetchAndReportWatermarksAsync(TopologyManager topologyManager)
{
SharedContext context = SharedContext.Global.instance;
Set<InetAddressAndPort> peers = new HashSet<>();
@ -142,14 +139,13 @@ public class WatermarkCollector implements ConfigurationService.Listener
return;
Snapshot snapshot = m.payload;
long minEpoch = configService.minEpoch();
forEachEpoch(configService::receiveClosed, snapshot.closed);
forEachEpoch(configService::receiveRetired, snapshot.retired);
long minEpoch = topologyManager.minEpoch();
forEachEpoch(topologyManager::onEpochClosed, snapshot.closed);
forEachEpoch(topologyManager::onEpochRetired, snapshot.retired);
for (Map.Entry<Long, Long> e : snapshot.synced.entrySet())
{
Node.Id node = new Node.Id(Ints.saturatedCast(e.getKey()));
for (long epoch = minEpoch; epoch <= e.getValue(); epoch++)
configService.receiveRemoteSyncComplete(node, epoch);
topologyManager.onReadyToCoordinate(node, e.getValue());
}
});
}

9
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -90,6 +90,7 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.fetch
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.recover;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryBootstrap;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryDurability;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryFetchTopology;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retryJoinBootstrap;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.retrySyncPoint;
import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.slowTxnPreaccept;
@ -303,7 +304,7 @@ public class AccordAgent implements Agent, OwnershipEventListener
}
RoutingKey homeKey = command.route().homeKey();
Shard shard = node.topology().forEpochIfKnown(homeKey, command.txnId().epoch());
Shard shard = node.topology().active().forEpochIfKnown(homeKey, command.txnId().epoch());
startTime = nonClashingStartTime(startTime, shard == null ? null : shard.nodes, node.id(), ONE_SECOND, random);
long delayMicros = Math.max(1, startTime - nowMicros);
@ -383,6 +384,12 @@ public class AccordAgent implements Agent, OwnershipEventListener
return retrySyncPoint.computeWait(attempt, units);
}
@Override
public long retryTopologyDelay(Node node, int attempt, TimeUnit units)
{
return retryFetchTopology.computeWait(attempt, units);
}
@Override
public long retryDurabilityDelay(Node node, int attempt, TimeUnit units)
{

6
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
View File

@ -46,6 +46,7 @@ import accord.primitives.Timestamp;
import accord.primitives.TimestampWithUniqueHlc;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.ActiveEpochs;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
@ -156,9 +157,10 @@ public class AccordInteropExecution implements ReadCoordinator
this.endpointMapper = endpointMapper;
// TODO (required): compare this to latest logic in Accord, make sure it makes sense
this.executes = node.topology().forEpoch(route, executeAt.epoch(), SHARE);
ActiveEpochs epochs = node.topology().active();
this.executes = epochs.forEpoch(route, executeAt.epoch(), SHARE);
this.allTopologies = txnId.epoch() != executeAt.epoch()
? node.topology().preciseEpochs(route, txnId.epoch(), executeAt.epoch(), SHARE)
? epochs.preciseEpochs(route, txnId.epoch(), executeAt.epoch(), SHARE)
: executes;
this.executeTopology = executes.getEpoch(executeAt.epoch());
this.coordinateTopology = allTopologies.getEpoch(txnId.epoch());

19
src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java
View File

@ -32,7 +32,6 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.UnversionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.AccordConfigurationService;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers;
import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
@ -77,7 +76,6 @@ public interface AccordTopologyUpdate
epochs[i] = in.readLong();
ranges[i] = KeySerializers.ranges.deserialize(in);
}
Invariants.require(ranges.length == epochs.length);
return new CommandStores.RangesForEpoch(epochs, ranges);
}
@ -162,10 +160,7 @@ public interface AccordTopologyUpdate
out.writeBoolean(image.update != null);
if (image.update != null)
TopologyUpdateSerializer.instance.serialize(image.update, out);
if (image.syncStatus == null)
out.writeByte(Byte.MAX_VALUE);
else
out.writeByte(image.syncStatus.ordinal());
out.writeByte(0); // defunct enum byte
KeySerializers.ranges.serialize(image.closed, out);
KeySerializers.ranges.serialize(image.retired, out);
@ -192,9 +187,7 @@ public interface AccordTopologyUpdate
update = TopologyUpdateSerializer.instance.deserialize(in);
TopologyImage image = new TopologyImage(epoch, kind, update);
byte syncStateByte = in.readByte();
if (syncStateByte != Byte.MAX_VALUE)
image.syncStatus = AccordConfigurationService.SyncStatus.values()[syncStateByte];
in.readByte(); // defunct enum byte
image.closed = KeySerializers.ranges.deserialize(in);
image.retired = KeySerializers.ranges.deserialize(in);
@ -252,7 +245,6 @@ public interface AccordTopologyUpdate
private final long epoch;
private final Kind kind;
private Journal.TopologyUpdate update;
private AccordConfigurationService.SyncStatus syncStatus = null;
private Ranges closed = Ranges.EMPTY;
private Ranges retired = Ranges.EMPTY;
@ -317,9 +309,6 @@ public interface AccordTopologyUpdate
Invariants.require(accumulator.update == null || accumulator.update.equals(update));
accumulator.update = update;
// We're iterating in _reverse_ order
if (accumulator.syncStatus == null)
accumulator.syncStatus = syncStatus;
accumulator.closed = accumulator.closed.with(closed);
accumulator.retired = accumulator.retired.with(retired);
}
@ -330,13 +319,13 @@ public interface AccordTopologyUpdate
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
TopologyImage that = (TopologyImage) o;
return epoch == that.epoch && Objects.equals(update, that.update) && syncStatus == that.syncStatus && closed.equals(that.closed) && retired.equals(that.retired);
return epoch == that.epoch && Objects.equals(update, that.update) && closed.equals(that.closed) && retired.equals(that.retired);
}
@Override
public int hashCode()
{
return Objects.hash(update, syncStatus, closed, retired, epoch);
return Objects.hash(update, closed, retired, epoch);
}
}

2
src/java/org/apache/cassandra/service/accord/repair/AccordRepair.java
View File

@ -90,7 +90,7 @@ public class AccordRepair
if (endpoints != null)
{
including = new ArrayList<>(endpoints.size());
AccordEndpointMapper mapper = AccordService.instance().configService().endpointMapper();
AccordEndpointMapper mapper = AccordService.instance().endpointMapper();
for (InetAddressAndPort ep : endpoints)
{
Node.Id id = mapper.mappedIdOrNull(ep);

1
src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java
View File

@ -42,7 +42,6 @@ import static accord.primitives.KeyDeps.SerializerSupport.txnIdsToKeys;
import static accord.primitives.RangeDeps.SerializerSupport.ranges;
import static accord.primitives.RangeDeps.SerializerSupport.rangesToTxnIds;
import static accord.primitives.RangeDeps.SerializerSupport.txnIdsToRanges;
import static org.apache.cassandra.service.accord.serializers.KeySerializers.keys;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.deserializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedIntsSize;

4
src/java/org/apache/cassandra/tcm/sequences/BootstrapAndJoin.java
View File

@ -30,7 +30,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import accord.local.Node;
import accord.utils.async.AsyncResult;
import com.googlecode.concurrenttrees.common.Iterables;
@ -371,7 +371,7 @@ public class BootstrapAndJoin extends MultiStepOperation<Epoch>
{
IAccordService service = AccordService.instance();
{
Future<?> ready = AccordService.toFuture(service.epochReadyFor(metadata, EpochReady::metadata));
Future<?> ready = AccordService.toFuture(service.epochReadyFor(metadata, EpochReady::active));
logger.info("Waiting for Accord metadata to be ready");
ready.syncThrowUncheckedOnInterrupt();
ready.rethrowIfFailed();

2
src/java/org/apache/cassandra/tcm/sequences/DropAccordTable.java
View File

@ -26,7 +26,7 @@ import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.Keyspaces;

2
src/java/org/apache/cassandra/tcm/sequences/Move.java
View File

@ -31,7 +31,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.db.TypeSizes;

5
src/java/org/apache/cassandra/tcm/transformations/PrepareJoin.java
View File

@ -24,6 +24,7 @@ import java.util.Set;
import com.google.common.collect.ImmutableSet;
import accord.utils.Invariants;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
@ -105,8 +106,8 @@ public class PrepareJoin implements Transformation
boolean joinTokenRing,
boolean streamData)
{
this.nodeId = nodeId;
this.tokens = tokens;
this.nodeId = Invariants.nonNull(nodeId);
this.tokens = Invariants.nonNull(tokens);
this.placementProvider = placementProvider;
this.joinTokenRing = joinTokenRing;
this.streamData = streamData;

2
test/distributed/org/apache/cassandra/distributed/shared/ClusterUtils.java
View File

@ -49,7 +49,7 @@ import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import org.agrona.collections.IntArrayList;
import org.apache.cassandra.tcm.compatibility.TokenRingUtils;
import org.apache.cassandra.utils.FBUtilities;

9
test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTest.java
View File

@ -22,7 +22,9 @@ import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import org.junit.Assert;
@ -30,6 +32,7 @@ import org.junit.Test;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import accord.topology.EpochReady;
import accord.topology.TopologyManager;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
@ -70,7 +73,8 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase
IInvokableInstance newInstance = cluster.bootstrap(config);
newInstance.startup(cluster);
spinUntilTrue(() -> cluster.stream().anyMatch(instance -> instance.callOnInstance(() -> StreamListener.listener.hasFailedStream)));
newInstance.shutdown(false);
try { newInstance.shutdown(false).get(5L, TimeUnit.MINUTES); }
catch (InterruptedException | ExecutionException | TimeoutException e) { throw new RuntimeException(e); }
cluster.get(1, 2).forEach(instance -> instance.runOnInstance(() -> StreamListener.listener.failStream = false));
newInstance.startup(cluster);
// todo: re-add once we fix write survey/join ring = false mode
@ -174,6 +178,7 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase
.with(NETWORK, GOSSIP))
.start())
{
cluster.setUncaughtExceptionsFilter(throwable -> throwable.getClass().getSimpleName().equals("UncheckedInterruptedException"));
cluster.schemaChange("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor':2}");
cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key(k, c)) WITH transactional_mode='full'");
@ -228,7 +233,7 @@ public class AccordBootstrapTest extends AccordBootstrapTestBase
for (long epoch = topologyManager.minEpoch() ; epoch <= topologyManager.epoch() ; ++epoch)
{
CountDownLatch latch = new CountDownLatch(1);
topologyManager.epochReady(epoch).data.invokeIfSuccess(latch::countDown);
topologyManager.epochReady(epoch, EpochReady::data).invokeIfSuccess(latch::countDown);
while (true)
{
try

34
test/distributed/org/apache/cassandra/distributed/test/accord/AccordBootstrapTestBase.java
View File

@ -23,7 +23,8 @@ import java.util.function.Function;
import org.junit.Assert;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import accord.topology.TopologyManager;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
@ -35,7 +36,6 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableFuncti
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.accord.AccordConfigurationService;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.tcm.ClusterMetadata;
@ -45,8 +45,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.assertj.core.api.Assertions;
import static org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot.ResultStatus.SUCCESS;
import static org.apache.cassandra.service.accord.AccordConfigurationService.SyncStatus.COMPLETED;
import static org.apache.cassandra.service.accord.AccordService.toFuture;
public class AccordBootstrapTestBase extends TestBaseImpl
{
@ -89,9 +88,8 @@ public class AccordBootstrapTestBase extends TestBaseImpl
{
boolean completed = service().epochReady(Epoch.create(epoch), await).await(60, TimeUnit.SECONDS);
Assertions.assertThat(completed)
.describedAs("Epoch %s did not become ready within timeout on %s -> %s",
epoch, FBUtilities.getBroadcastAddressAndPort(),
service().configService().getEpochSnapshot(epoch))
.describedAs("Epoch %s did not become ready within timeout on %s",
epoch, FBUtilities.getBroadcastAddressAndPort())
.isTrue();
}
catch (InterruptedException e)
@ -104,10 +102,10 @@ public class AccordBootstrapTestBase extends TestBaseImpl
{
try
{
AccordConfigurationService configService = service().configService();
boolean completed = configService.unsafeLocalSyncNotified(epoch).await(60, TimeUnit.SECONDS);
Assert.assertTrue(String.format("Local sync notification for epoch %s did not become ready within timeout on %s\n%s",
epoch, FBUtilities.getBroadcastAddressAndPort(), service().configService().getDebugStr()), completed);
TopologyManager tm = service().topology();
boolean completed = toFuture(tm.epochReady(epoch, EpochReady::coordinate)).await(60, TimeUnit.SECONDS);
Assert.assertTrue(String.format("Local sync notification for epoch %s did not become ready within timeout on %s",
epoch, FBUtilities.getBroadcastAddressAndPort()), completed);
}
catch (InterruptedException e)
{
@ -131,7 +129,7 @@ public class AccordBootstrapTestBase extends TestBaseImpl
static long awaitMaxEpochMetadataReady(Cluster cluster)
{
return awaitMaxEpoch(cluster, EpochReady::metadata, false);
return awaitMaxEpoch(cluster, EpochReady::active, false);
}
static long awaitMaxEpoch(Cluster cluster, SerializableFunction<EpochReady, AsyncResult<Void>> await, boolean expectReadyToRead)
@ -147,16 +145,16 @@ public class AccordBootstrapTestBase extends TestBaseImpl
Assert.assertEquals(maxEpoch, ClusterMetadata.current().epoch.getEpoch());
AccordService service = (AccordService) AccordService.instance();
awaitEpoch(maxEpoch, aw);
AccordConfigurationService configService = service.configService();
TopologyManager tm = service.topology();
awaitLocalSyncNotification(maxEpoch);
for (long epoch = configService.minEpoch(); epoch <= maxEpoch; epoch++)
for (long epoch = tm.minEpoch(); epoch <= maxEpoch; epoch++)
{
Assert.assertEquals(COMPLETED, configService.getEpochSnapshot(maxEpoch).syncStatus);
Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).acknowledged);
Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).received);
EpochReady epochReady = tm.active().epochReady(epoch);
Assert.assertTrue(epochReady.active().isDone());
Assert.assertTrue(epochReady.coordinate.isDone());
if (expectReadyToRead)
Assert.assertEquals(SUCCESS, configService.getEpochSnapshot(maxEpoch).reads);
Assert.assertTrue(epochReady.reads.isDone());
}
}, node.transfer(await));
}

2
test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java
View File

@ -709,7 +709,7 @@ public abstract class AccordCQLTestBase extends AccordTestBase
Unseekables<?> routables = AccordTestUtils.createTxn(sb.toString()).keys().toParticipants();
long epoch = AccordService.instance().topology().epoch();
Topologies topology = AccordService.instance().topology().withUnsyncedEpochs(routables, epoch, epoch);
Topologies topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch);
// we don't detect out-of-bounds read/write yet, so use this to validate we reach different shards
Assertions.assertThat(topology.totalShards()).isEqualTo(2);
});

1
test/distributed/org/apache/cassandra/distributed/test/accord/AccordMoveTest.java
View File

@ -63,7 +63,6 @@ public class AccordMoveTest extends AccordBootstrapTestBase
cluster.schemaChange("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor':2}");
cluster.schemaChange("CREATE TABLE ks.tbl (k int, c int, v int, primary key(k, c)) WITH transactional_mode='full'");
long initialMax = maxEpoch(cluster);
long[] tokens = new long[3];
for (int i=0; i<3; i++)
{

5
test/distributed/org/apache/cassandra/distributed/test/accord/AccordRecoverFromAvailabilityLossTest.java
View File

@ -31,6 +31,7 @@ import org.junit.Test;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import accord.topology.EpochReady;
import accord.topology.TopologyManager;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
@ -106,7 +107,7 @@ public class AccordRecoverFromAvailabilityLossTest extends AccordBootstrapTestBa
FBUtilities.waitOnFuture(cluster.get(removeIdx).shutdown(false));
{
List<java.util.concurrent.Future<?>> results = new ArrayList<>();
List<Future<?>> results = new ArrayList<>();
for (int key = 50; key < 100; key++)
{
String query = "BEGIN TRANSACTION\n" +
@ -155,7 +156,7 @@ public class AccordRecoverFromAvailabilityLossTest extends AccordBootstrapTestBa
for (long epoch = topologyManager.minEpoch() ; epoch <= topologyManager.epoch() ; ++epoch)
{
CountDownLatch latch = new CountDownLatch(1);
topologyManager.epochReady(epoch).data.invokeIfSuccess(latch::countDown);
topologyManager.epochReady(epoch, EpochReady::data).invokeIfSuccess(latch::countDown);
while (true)
{
try

6
test/distributed/org/apache/cassandra/distributed/test/accord/AccordSimpleFastPathTest.java
View File

@ -23,6 +23,7 @@ import java.util.Comparator;
import java.util.HashSet;
import java.util.Set;
import accord.topology.TopologyManager;
import org.apache.cassandra.service.accord.AccordFastPath;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
@ -40,7 +41,6 @@ import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.accord.AccordConfigurationService;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.slf4j.Logger;
@ -109,8 +109,8 @@ public class AccordSimpleFastPathTest extends TestBaseImpl
Assert.assertEquals(idSet(), accordFastPath.unavailableIds());
long epoch = cm.epoch.getEpoch();
AccordConfigurationService configService = ((AccordService) AccordService.instance()).configService();
Topology topology = configService.getTopologyForEpoch(epoch);
TopologyManager tm = AccordService.instance().topology();
Topology topology = tm.active().globalForEpoch(epoch);
Assert.assertFalse(topology.shards().isEmpty());
topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath)));
return cm.epoch.getEpoch();

3
test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
View File

@ -177,8 +177,7 @@ public abstract class AccordTestBase extends TestBaseImpl
return true;
AccordService accord = ((AccordService) AccordService.instance());
return metadata.epoch.getEpoch() == accord.configService().currentEpoch() &&
metadata.epoch.getEpoch() == accord.topology().current().epoch();
return metadata.epoch.getEpoch() == accord.topology().current().epoch();
}, 60));
}

2
test/distributed/org/apache/cassandra/distributed/test/accord/journal/AccordJournalReplayTest.java
View File

@ -137,7 +137,7 @@ public class AccordJournalReplayTest extends TestBaseImpl
return txnId.toString();
});
cluster.get(1).shutdown(false);
cluster.get(1).shutdown(false).get();
cluster.get(1).startup();
cluster.get(1).runOnInstance(() -> {
AccordService service = (AccordService) AccordService.instance();

32
test/simulator/test/org/apache/cassandra/simulator/test/EpochStressTest.java
View File

@ -33,12 +33,11 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.EpochReady;
import accord.topology.TopologyManager;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.service.accord.AccordConfigurationService;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.simulator.Action;
@ -195,7 +194,6 @@ public class EpochStressTest extends SimulationTestBase
AccordService accord = (AccordService) AccordService.instance();
Node node = accord.node();
AccordConfigurationService configService = (AccordConfigurationService) node.configService();
TopologyManager tm = node.topology();
long minEpoch = tm.minEpoch();
@ -212,26 +210,9 @@ public class EpochStressTest extends SimulationTestBase
for (long epoch = minEpoch; epoch <= maxEpoch; epoch++)
{
long finalEpoch = epoch;
ConfigurationService.EpochReady ready = tm.epochReady(epoch);
EpochReady ready = tm.active().epochReady(epoch);
while (!isDone(ready))
sleep.accept(() -> "Epoch " + finalEpoch + "'s EpochReady is not done; " + ready);
AccordConfigurationService.EpochSnapshot snapshot = configService.getEpochSnapshot(epoch);
while (!isDone(snapshot))
{
AccordConfigurationService.SyncStatus status = snapshot.syncStatus;
sleep.accept(() -> "Epoch " + finalEpoch + "'s SyncStatus is not done; " + status);
snapshot = configService.getEpochSnapshot(epoch);
}
Ranges expected = tm.globalForEpoch(epoch).ranges().mergeTouching();
Ranges synced = tm.syncComplete(epoch).mergeTouching();
while (!isDone(synced, expected))
{
Ranges finalSynced = synced;
sleep.accept(() -> "Epoch " + finalEpoch + "'s syncComplete is not done; missing " + expected.without(finalSynced));
synced = tm.syncComplete(epoch).mergeTouching();
}
}
logger.info("All epochs completed in {}", Duration.ofNanos(Clock.Global.nanoTime() - startNano));
}
@ -241,14 +222,9 @@ public class EpochStressTest extends SimulationTestBase
return synced.equals(expected);
}
private static boolean isDone(AccordConfigurationService.EpochSnapshot snapshot)
private static boolean isDone(EpochReady ready)
{
return snapshot.syncStatus == AccordConfigurationService.SyncStatus.COMPLETED;
}
private static boolean isDone(ConfigurationService.EpochReady ready)
{
return ready.metadata.isDone()
return ready.active.isDone()
&& ready.coordinate.isDone()
&& ready.data.isDone()
&& ready.reads.isDone();

45
test/unit/org/apache/cassandra/db/virtual/AccordVirtualTablesTest.java
View File

@ -26,10 +26,15 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.api.ConfigurationService;
import accord.Utils;
import accord.api.MessageSink;
import accord.api.TopologySorter;
import accord.impl.mock.MockCluster;
import accord.local.Node;
import accord.primitives.Ranges;
import accord.topology.ActiveEpoch;
import accord.topology.ActiveEpochs;
import accord.topology.EpochReady;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
@ -54,8 +59,8 @@ public class AccordVirtualTablesTest extends CQLTester
public static final Node.Id N1 = new Node.Id(1);
public static final SortedArrays.SortedArrayList<Node.Id> ALL = SortedArrays.SortedArrayList.ofSorted(N1);
public static final Set<Node.Id> FP = Collections.singleton(N1);
public static final String SUCCESS = "success";
public static final String PENDING = "pending";
public static final String SUCCESS = "Success";
public static final String PENDING = "SettableResult{status=pending}";
public static final List<String> FULL_RANGE = List.of("(-Inf, +Inf]");
public static TableId T1;
@ -92,12 +97,14 @@ public class AccordVirtualTablesTest extends CQLTester
{
TopologyManager tm = empty();
long e1 = 1;
tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1), e -> {});
ActiveEpoch a1 = active(topology(e1, T1), EpochReady.done(e1), Topology.EMPTY);
tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a1 }, -1));
assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS),
row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS));
long e2 = 2;
tm.onTopologyUpdate(topology(e2, T1), () -> pendingReady(e1), e -> {});
ActiveEpoch a2 = active(topology(e2, T1), pendingReady(e2), Topology.EMPTY);
tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a2, a1 }, -1));
assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.EPOCHS),
row(e2, false, PENDING, PENDING, PENDING, PENDING),
row(e1, true, SUCCESS, SUCCESS, SUCCESS, SUCCESS));
@ -108,7 +115,8 @@ public class AccordVirtualTablesTest extends CQLTester
{
TopologyManager tm = empty();
long e1 = 1;
tm.onTopologyUpdate(topology(e1, T1), () -> ConfigurationService.EpochReady.done(e1), e -> {});
ActiveEpoch a1 = active(topology(e1, T1), EpochReady.done(e1), Topology.EMPTY);
tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a1 }, -1));
// the range was added in the first epoch, so its fully synced
assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
@ -116,12 +124,13 @@ public class AccordVirtualTablesTest extends CQLTester
// range is no longer "added" so doesn't show up as synced!
long e2 = 2;
tm.onTopologyUpdate(topology(e2, T1), () -> ConfigurationService.EpochReady.done(e2), e -> {});
ActiveEpoch a2 = active(topology(e2, T1), EpochReady.done(e2), a1.global());
tm.unsafeSetActive(ActiveEpochs.unsafeNew(tm, new ActiveEpoch[] { a2, a1 }, -1));
assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE));
// sync the range
tm.onEpochSyncComplete(N1, e2);
tm.onReadyToCoordinate(N1, e2);
assertRows(execute("SELECT * FROM " + VIRTUAL_VIEWS + "." + AccordVirtualTables.TABLE_EPOCHS),
row(e2, T1_META.keyspace, T1_META.name, List.of(), List.of(), List.of(), List.of(), FULL_RANGE),
row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, List.of(), List.of(), List.of(), FULL_RANGE));
@ -139,9 +148,9 @@ public class AccordVirtualTablesTest extends CQLTester
row(e1, T1_META.keyspace, T1_META.name, FULL_RANGE, FULL_RANGE, List.of(), FULL_RANGE, FULL_RANGE));
}
private static ConfigurationService.EpochReady pendingReady(long epoch)
private static EpochReady pendingReady(long epoch)
{
return new ConfigurationService.EpochReady(epoch, AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable());
return new EpochReady(epoch, AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable(), AsyncResults.settable());
}
private static Topology topology(long epoch, TableId tableId)
@ -152,26 +161,34 @@ public class AccordVirtualTablesTest extends CQLTester
private static TopologyManager empty()
{
TopologySorter sorter = (TopologySorter.StaticSorter) (node1, node2, shards) -> 0;
TopologySorter.Supplier supplier = new TopologySorter.Supplier()
{
@Override
public TopologySorter get(Topology topologies)
{
return sorter;
return SORTER;
}
@Override
public TopologySorter get(Topologies topologies)
{
return sorter;
return SORTER;
}
};
TopologyManager tm = new TopologyManager(supplier, null, N1, null, null);
Node node = Utils.createNode(N1, Topology.EMPTY, new MessageSink.NoOpSink(), new MockCluster.Clock(0));
TopologyManager tm = new TopologyManager(supplier, node, ignore -> { throw new UnsupportedOperationException(); }, null, null);
var mock = Mockito.mock(IAccordService.class);
Mockito.when(mock.topology()).thenReturn(tm);
AccordService.unsafeSetNewAccordService(mock);
return tm;
}
private static final TopologySorter SORTER = (TopologySorter.StaticSorter) (node1, node2, shards) -> 0;
private static ActiveEpoch active(Topology topology, EpochReady ready, Topology prev)
{
return ActiveEpoch.unsafeNew(N1, topology, ready, SORTER, prev.ranges());
}
}

2
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
View File

@ -31,7 +31,7 @@ import java.util.function.Supplier;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import accord.api.ConfigurationService.EpochReady;
import accord.topology.EpochReady;
import accord.api.Journal;
import accord.api.RoutingKey;
import accord.local.CommandStores;

295
test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java
View File

@ -1,295 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.net.UnknownHostException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import accord.api.Journal.TopologyUpdate;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.impl.AbstractConfigurationServiceTest;
import accord.local.Node.Id;
import accord.topology.Topology;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.async.AsyncResult;
import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.TestParams;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.net.ConnectionType;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.service.accord.EndpointMapping.Updateable;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
import org.apache.cassandra.tcm.membership.Location;
import org.apache.cassandra.tcm.membership.NodeAddresses;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.concurrent.Future;
import static accord.impl.AbstractConfigurationServiceTest.TestListener;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
public class AccordConfigurationServiceTest
{
private static final Id ID1 = new Id(1);
private static final Id ID2 = new Id(2);
private static final Id ID3 = new Id(3);
private static final SortedArrayList<Id> ID_LIST = new SortedArrayList<>(new Id[] { ID1, ID2, ID3 });
private static final String KEYSPACE_NAME = "test_ks";
private static final TableId TBL_ID = TableId.fromUUID(new UUID(0, 1));
private static EndpointMapping mappingForEpoch(long epoch)
{
try
{
EndpointMapping.Builder builder = EndpointMapping.builder(epoch);
builder.add(InetAddressAndPort.getByName("127.0.0.1"), ID1);
builder.add(InetAddressAndPort.getByName("127.0.0.2"), ID2);
builder.add(InetAddressAndPort.getByName("127.0.0.3"), ID3);
return builder.build();
}
catch (UnknownHostException e)
{
throw new RuntimeException(e);
}
}
private static class Messaging implements MessageDelivery
{
static class Request
{
final Message<?> message;
final InetAddressAndPort to;
final RequestCallback<?> callback;
public Request(Message<?> message, InetAddressAndPort to, RequestCallback<?> callback)
{
this.message = message;
this.to = to;
this.callback = callback;
}
}
final List<Request> requests = new ArrayList<>();
@Override
public <REQ> void send(Message<REQ> message, InetAddressAndPort to)
{
requests.add(new Request(message, to, null));
}
@Override
public <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb)
{
requests.add(new Request(message, to, cb));
}
@Override
public <REQ, RSP> void sendWithCallback(Message<REQ> message, InetAddressAndPort to, RequestCallback<RSP> cb, ConnectionType specifyConnection)
{
throw new UnsupportedOperationException();
}
@Override
public <REQ, RSP> Future<Message<RSP>> sendWithResult(Message<REQ> message, InetAddressAndPort to)
{
throw new UnsupportedOperationException();
}
@Override
public <V> void respond(V response, Message<?> message)
{
throw new UnsupportedOperationException();
}
}
@BeforeClass
public static void beforeClass() throws Throwable
{
ServerTestUtils.daemonInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
SchemaLoader.prepareServer();
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1),
parse("CREATE TABLE tbl (k int, c int, v int, primary key (k, c)) WITH transactional_mode='full'", "ks"));
}
@Test
public void loadTest() throws Throwable
{
ValidatingClusterMetadataService cms = ValidatingClusterMetadataService.createAndRegister(Version.MIN_ACCORD_VERSION);
AccordJournal journal = null;
try
{
journal = initJournal();
AccordConfigurationService service = new AccordConfigurationService(ID1, new AccordAgent(), new Updateable(), new Messaging(), ScheduledExecutors.scheduledTasks);
AccordJournal journal_ = journal;
TestListener listener = new TestListener(service, true)
{
@Override
public AsyncResult<Void> onTopologyUpdate(Topology topology)
{
// Fake journal save
journal_.saveTopology(new TopologyUpdate(new Int2ObjectHashMap<>(), topology), () -> {});
return super.onTopologyUpdate(topology);
}
};
service.registerListener(listener);
service.start();
Topology topology1 = createTopology(cms);
((Updateable)service.endpointMapper()).updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
service.reportTopology(topology1);
service.receiveRemoteSyncComplete(ID1, 1);
service.receiveRemoteSyncComplete(ID2, 1);
service.receiveRemoteSyncComplete(ID3, 1);
Topology topology2 = createTopology(cms);
service.reportTopology(topology2);
service.receiveRemoteSyncComplete(ID1, 2);
Topology topology3 = createTopology(cms);
service.reportTopology(topology3);
AccordConfigurationService loaded = new AccordConfigurationService(ID1, new AccordAgent(), new Updateable(), new Messaging(), ScheduledExecutors.scheduledTasks);
((EndpointMapping.Updateable)loaded.endpointMapper()).updateMapping(mappingForEpoch(cms.metadata().epoch.getEpoch() + 1));
listener = new AbstractConfigurationServiceTest.TestListener(loaded, true);
loaded.registerListener(listener);
journal_.closeCurrentSegmentForTestingIfNonEmpty();
List<TopologyUpdate> list = journal.replayTopologies();
// Simulate journal replay
for (TopologyUpdate update : list)
loaded.reportTopology(update.global);
loaded.start();
listener.assertTopologiesFor(1L, 2L, 3L);
listener.assertTopologyForEpoch(1, topology1);
listener.assertTopologyForEpoch(2, topology2);
listener.assertTopologyForEpoch(3, topology3);
}
finally
{
if (journal != null)
journal.shutdown();
}
}
private static AccordJournal initJournal() throws Throwable
{
File directory = new File(Files.createTempDirectory("config_service_test"));
directory.deleteRecursiveOnExit();
Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord");
ColumnFamilyStore cfs = ks.getColumnFamilyStore(AccordKeyspace.JOURNAL);
AccordJournal journal = new AccordJournal(new TestParams(), directory, cfs);
journal.start(null);
journal.unsafeSetStarted();
return journal;
}
private static Topology createTopology(ValidatingClusterMetadataService cms)
{
ClusterMetadata previous = cms.metadata();
ClusterMetadata.Transformer next = previous.transformer();
maybeCreateTable(previous, next);
ClusterMetadata metadata = next.build().metadata;
cms.setMetadata(metadata);
return AccordTopology.createAccordTopology(metadata);
}
private static void maybeCreateTable(ClusterMetadata previous, ClusterMetadata.Transformer next)
{
Optional<KeyspaceMetadata> ks = previous.schema.getKeyspaces().get(KEYSPACE_NAME);
if (ks.isPresent()) return;
// lets create it
TableMetadata table = TableMetadata.builder(KEYSPACE_NAME, "tbl")
.id(TBL_ID)
.kind(TableMetadata.Kind.REGULAR)
.partitioner(Murmur3Partitioner.instance)
.addPartitionKeyColumn("pk", Int32Type.instance)
.build();
KeyspaceMetadata keyspace = KeyspaceMetadata.create(KEYSPACE_NAME, KeyspaceParams.simple(ID_LIST.size()))
.withSwapped(Tables.builder().add(table).build());
next.with(new DistributedSchema(previous.schema.getKeyspaces().with(keyspace)));
for (Id node : ID_LIST)
{
// not forcing the cms node id to match as they do when this logic was first added...
next.register(new NodeAddresses(getAddress(node)),
new Location("dc1", "rack1"),
NodeVersion.CURRENT);
next.proposeToken(new NodeId(node.id), Collections.singleton(new Murmur3Partitioner.LongToken(node.id)));
}
DataPlacement.Builder replication = DataPlacement.builder();
Range<Token> fullRange = new Range<>(Murmur3Partitioner.MINIMUM, Murmur3Partitioner.MINIMUM);
for (int i = 0; i < ID_LIST.size(); i++)
{
InetAddressAndPort address = getAddress(ID_LIST.get(i));
Replica replica = new Replica(address, fullRange, true);
replication.withReadReplica(next.epoch(), replica).withWriteReplica(next.epoch(), replica);
}
next.with(previous.placements.unbuild().with(keyspace.params.replication, replication.build()).build());
}
private static InetAddressAndPort getAddress(Id node)
{
try
{
return InetAddressAndPort.getByAddress(new byte[]{127, 0, 0, (byte) node.id});
}
catch (UnknownHostException e)
{
throw new RuntimeException(e);
}
}
}

150
test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
View File

@ -41,24 +41,28 @@ import com.google.common.collect.Sets;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.Utils;
import accord.api.Agent;
import accord.impl.AbstractTestConfigurationService;
import accord.api.MessageSink;
import accord.api.TopologyListener;
import accord.api.TopologySorter;
import accord.impl.TestAgent;
import accord.impl.basic.Pending;
import accord.impl.basic.PendingQueue;
import accord.impl.basic.MonitoredPendingQueue;
import accord.impl.basic.RandomDelayQueue;
import accord.impl.basic.SimulatedDelayedExecutorService;
import accord.impl.mock.MockCluster;
import accord.local.Node;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.RandomSource;
import accord.utils.SortedArrays.SortedArrayList;
import accord.utils.SortedList;
import accord.utils.SortedListSet;
import org.apache.cassandra.concurrent.AdaptingScheduledExecutorPlus;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
@ -72,7 +76,6 @@ import org.apache.cassandra.net.ConnectionType;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.tcm.ValidatingClusterMetadataService;
import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.AccordGenerators;
@ -82,6 +85,7 @@ import org.assertj.core.api.Assertions;
import static accord.utils.Property.qt;
import static org.apache.cassandra.simulator.RandomSource.Choices.choose;
import static org.apache.cassandra.utils.AccordGenerators.partitioner;
public class AccordSyncPropagatorTest
{
@ -95,14 +99,14 @@ public class AccordSyncPropagatorTest
@Test
public void burnTest()
{
Gen<Ranges> rangesGen = AccordGenerators.ranges().filter(r -> !r.isEmpty());
Gen<Gen<Ranges>> rangesGenGen = partitioner().map(p -> AccordGenerators.ranges(p).filter(r -> !r.isEmpty()));
Gen<List<Node.Id>> nodesGen = Gens.lists(AccordGens.nodes()).unique().ofSizeBetween(1, 40);
qt().withExamples(100).check(rs -> {
qt().withExamples(10).check(rs -> {
// when gossip and cluster metadata don't know an endpoint, retries are avoided (node removed)
// so when instances are created here they are added to gossip to trick the membership check...
Gossiper.instance.clearUnsafe();
SortedList<Node.Id> nodes = SortedArrayList.copyUnsorted(nodesGen.next(rs), Node.Id[]::new);
SortedArrayList<Node.Id> nodes = SortedArrayList.copyUnsorted(nodesGen.next(rs), Node.Id[]::new);
SortedListSet<Node.Id> nodesAsSet = SortedListSet.allOf(nodes);
List<Throwable> failures = new ArrayList<>();
@ -112,22 +116,31 @@ public class AccordSyncPropagatorTest
SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, agent, null);
ScheduledExecutorPlus scheduler = new AdaptingScheduledExecutorPlus(globalExecutor);
Cluster cluster = new Cluster(nodes, rs, scheduler);
long epochOffset = rs.nextLong(1, 1024);
int numEpochs = rs.nextInt(1, 10);
final long epochOffset = rs.nextLong(1, 1024);
Map<Long, Ranges> allRanges = new HashMap<>();
Cluster cluster = new Cluster(nodes, epochOffset, allRanges, rs, scheduler);
Gen<Ranges> rangesGen = rangesGenGen.next(rs);
int numEpochs = rs.nextInt(1, 10);
Pending.Global.setNoActiveOrigin();
Set<Object> prefixes = new HashSet<>();
for (int i = 0; i < numEpochs; i++)
{
long epoch = epochOffset + i;
Ranges ranges = rangesGen.next(rs);
allRanges.put(epoch, ranges);
Ranges ranges;
{
Ranges tmp = rangesGen.next(rs);
while (tmp.stream().anyMatch(r -> prefixes.contains(r.prefix())))
tmp = rangesGen.next(rs);
ranges = tmp;
}
ranges.stream().forEach(r -> prefixes.add(r.prefix()));
allRanges.put(epoch, ranges.mergeTouching());
scheduler.schedule(() -> {
for (Node.Id nodeId : nodes)
{
cluster.node(nodeId).configurationService.receiveRemoteSyncComplete(nodeId, epoch);
cluster.node(nodeId).propagator.reportCoordinationReady(epoch, nodes, nodeId);
cluster.node(nodeId).topology.onReadyToCoordinate(nodeId, epoch);
cluster.node(nodeId).propagator.onReadyToCoordinate(epoch, nodes);
}
for (int j = 0, attempts = rs.nextInt(1, 4); j < attempts; j++)
@ -137,8 +150,12 @@ public class AccordSyncPropagatorTest
Cluster.Instance inst = cluster.node(choose(rs, nodesAsSet));
scheduler.schedule(() -> {
Ranges subrange = Ranges.of(range);
inst.propagator.reportClosed(epoch, nodes, subrange);
scheduler.schedule(() -> inst.propagator.reportRetired(epoch, nodes, subrange), 1, TimeUnit.MINUTES);
inst.topology.onEpochClosed(subrange, epoch);
inst.propagator.onEpochClosed(subrange, epoch, nodes);
scheduler.schedule(() -> {
inst.topology.onEpochRetired(subrange, epoch);
inst.propagator.onEpochRetired(subrange, epoch, nodes);
}, 1, TimeUnit.MINUTES);
}, rs.nextInt(30, 300), TimeUnit.SECONDS);
}
}
@ -166,12 +183,16 @@ public class AccordSyncPropagatorTest
for (Cluster.Instance inst : cluster.instances.values())
{
Cluster.ConfigService cs = inst.configurationService;
Cluster.Listener cs = inst.listener;
assertSetsEqual(cs.completedEpochs, allRanges.keySet(), "completedEpochs %s", inst.id);
assertSetsEqual(cs.syncCompletes.keySet(), allRanges.keySet(), "syncCompletes %s", inst.id);
for (Map.Entry<Long, Set<Node.Id>> e : cs.syncCompletes.entrySet())
assertSetsEqual(e.getValue(), nodesAsSet, "syncCompletes values on %s", inst.id);
for (Map.Entry<?, Ranges> e : cs.closed.entrySet())
e.setValue(e.getValue().mergeTouching());
for (Map.Entry<?, Ranges> e : cs.redundant.entrySet())
e.setValue(e.getValue().mergeTouching());
assertMapEquals(cs.closed, allRanges, "Unexpected state for closed on %s", inst.id);
assertMapEquals(cs.redundant, allRanges, "Unexpected state for redundant on %s", inst.id);
}
@ -195,7 +216,7 @@ public class AccordSyncPropagatorTest
V value = e.getValue();
V other = expected.get(e.getKey());
if (!Objects.equals(value, other))
errors.add(String.format("Missmatch at key %s: expected %s but given %s", e.getKey(), other, value));
errors.add(String.format("Mismatch at key %s: expected %s but given %s", e.getKey(), other, value));
}
if (!errors.isEmpty())
throw new AssertionError(String.join("\n", errors));
@ -214,6 +235,8 @@ public class AccordSyncPropagatorTest
private final ScheduledExecutorPlus scheduler;
private Cluster(List<Node.Id> nodes,
long minEpoch,
Map<Long, Ranges> allRanges,
RandomSource rs,
ScheduledExecutorPlus scheduler)
{
@ -225,10 +248,16 @@ public class AccordSyncPropagatorTest
{
InetAddressAndPort address = addressFromInt(id.id);
nodeToAddress.put(id, address);
ConfigService cs = new ConfigService(id, nodes);
Node node = Utils.createNode(id, Topology.EMPTY, new MessageSink.NoOpSink(), new MockCluster.Clock(0));
TopologyManager topology = new TopologyManager((TopologySorter.StaticSorter)(a,b,c)->0, node, ignore -> { throw new UnsupportedOperationException(); }, null, null);
Listener cs = new Listener(id, minEpoch, nodes, allRanges);
Sink sink = new Sink(id);
FailureWrapper fw = new FailureWrapper(Cluster.this, id);
instances.put(id, new Instance(id, cs, sink, new AccordSyncPropagator(id, fw, sink, scheduler, cs)));
AccordSyncPropagator propagator = new AccordSyncPropagator(id, fw, sink, scheduler);
topology.addListener(propagator);
topology.addListener(cs);
propagator.setTestListener(cs);
instances.put(id, new Instance(id, topology, cs, sink, propagator));
Gossiper.instance.endpointStateMap.put(address, new EndpointState(HeartBeatState.empty()));
}
this.nodeToAddress = nodeToAddress.build();
@ -326,7 +355,7 @@ public class AccordSyncPropagatorTest
throw new IllegalStateException("Unknown action: " + action);
}
callbacks.put(message.id(), cb);
scheduler.schedule(() -> AccordService.receive(this, node(to).configurationService, (Message<AccordSyncPropagator.Notification>) message.withFrom(mappedEndpointOrNull(from))), 500, TimeUnit.MILLISECONDS);
scheduler.schedule(() -> AccordService.receive(this, node(to).topology, (Message<AccordSyncPropagator.Notification>) message.withFrom(mappedEndpointOrNull(from))), 500, TimeUnit.MILLISECONDS);
scheduler.schedule(() -> {
RequestCallback<?> removed = callbacks.remove(message.id());
if (removed != null)
@ -413,56 +442,58 @@ public class AccordSyncPropagatorTest
}
}
private class ConfigService extends AbstractTestConfigurationService implements AccordSyncPropagator.Listener
private static class Listener implements AccordSyncPropagator.TestListener, TopologyListener
{
private final Node.Id self;
private final long minEpoch;
private final List<Node.Id> nodes;
private final Map<Long, Set<Node.Id>> syncCompletes = new HashMap<>();
private final Map<Long, Set<Node.Id>> endpointAcks = new HashMap<>();
private final NavigableSet<Long> completedEpochs = Collections.synchronizedNavigableSet(new TreeSet<>());
private final Map<Long, Ranges> allRanges;
private final Map<Long, Ranges> closed = new HashMap<>();
private final Map<Long, Ranges> redundant = new HashMap<>();
private ConfigService(Node.Id node, List<Node.Id> nodes)
private Listener(Node.Id node, long minEpoch, List<Node.Id> nodes, Map<Long, Ranges> allRanges)
{
super(node, new AccordAgent());
this.self = node;
this.minEpoch = minEpoch;
this.nodes = nodes;
this.allRanges = allRanges;
}
@Override
protected void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch)
public void onRemoteReadyToCoordinate(Node.Id node, long epoch)
{
syncCompletes.computeIfAbsent(epoch, ignore -> new HashSet<>()).add(node);
}
@Override
public void fetchTopologyForEpoch(long epoch)
public void onEpochClosed(Ranges ranges, long epoch, @Nullable Topology topology)
{
// TODO
merge(closed, ranges, epoch);
}
@Override
protected void onReadyToCoordinate(Topology topology)
public void onEpochRetired(Ranges ranges, long epoch, @Nullable Topology topology)
{
instances.get(localId).propagator.reportCoordinationReady(topology.epoch(), topology.nodes(), localId);
merge(redundant, ranges, epoch);
}
@Override
public void reportEpochClosed(Ranges ranges, long epoch)
{
Topology topology = getTopologyForEpoch(epoch);
instances.get(localId).propagator.reportClosed(epoch, topology.nodes(), ranges);
}
@Override
public void reportEpochRetired(Ranges ranges, long epoch)
{
Topology topology = getTopologyForEpoch(epoch);
instances.get(localId).propagator.reportRetired(epoch, topology.nodes(), ranges);
}
@Override
public void reportEpochRemoved(long epoch)
private void merge(Map<Long, Ranges> map, Ranges ranges, long epoch)
{
while (epoch >= minEpoch)
{
Ranges cur = map.get(epoch);
Ranges upd = cur == null ? ranges : cur.with(ranges);
if (upd == cur)
break;
if (cur != null)
ranges = ranges.without(cur);
ranges = ranges.without(allRanges.get(epoch));
map.put(epoch, upd);
--epoch;
}
}
@Override
@ -472,42 +503,25 @@ public class AccordSyncPropagatorTest
if (acks.add(id) && acks.containsAll(nodes))
completedEpochs.add(epoch);
}
@Override
public void onEndpointNack(Node.Id id, long epoch)
{
onEndpointAck(id, epoch);
}
@Override
public synchronized void receiveClosed(Ranges ranges, long epoch)
{
super.receiveClosed(ranges, epoch);
closed.merge(epoch, ranges, Ranges::with);
}
@Override
public synchronized void receiveRetired(Ranges ranges, long epoch)
{
super.receiveRetired(ranges, epoch);
redundant.merge(epoch, ranges, Ranges::with);
}
}
public class Instance
{
private final Node.Id id;
private final ConfigService configurationService;
private final TopologyManager topology;
private final Listener listener;
private final Sink messagingService;
private final AccordSyncPropagator propagator;
private Instance(Node.Id id,
ConfigService configurationService,
TopologyManager topology,
Listener listener,
Sink messagingService,
AccordSyncPropagator propagator)
{
this.id = id;
this.configurationService = configurationService;
this.topology = topology;
this.listener = listener;
this.messagingService = messagingService;
this.propagator = propagator;
}

140
test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
View File

@ -39,8 +39,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import java.util.function.Supplier;
import javax.annotation.Nullable;
@ -49,14 +48,17 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ConfigurationService;
import accord.api.ConfigurationService.EpochReady;
import accord.Utils;
import accord.api.MessageSink;
import accord.impl.TestAgent;
import accord.impl.mock.MockCluster;
import accord.local.ShardDistributor;
import accord.impl.DefaultTimeouts;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TestAgent;
import accord.local.Node;
import accord.local.TimeService;
import accord.primitives.Ranges;
import accord.topology.EpochReady;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.utils.Gen;
@ -66,6 +68,7 @@ import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import accord.utils.async.Cancellable;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.SimulatedExecutorFactory;
@ -86,8 +89,7 @@ import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.schema.TableParams;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.service.accord.AccordConfigurationService.EpochSnapshot;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
@ -112,6 +114,8 @@ import org.assertj.core.description.Description;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
import static org.apache.cassandra.config.DatabaseDescriptor.getAccordCommandStoreShardCount;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
public class EpochSyncTest
{
@ -128,7 +132,7 @@ public class EpochSyncTest
@Test
public void test()
{
stateful().withSeed(1).withExamples(50).withSteps(500).check(commands(() -> Cluster::new)
stateful().withSeed(152472217520379L).withExamples(50).withSteps(500).check(commands(() -> Cluster::new)
.destroyState(cluster -> {
finishPendingWork(cluster);
cluster.processAll();
@ -195,7 +199,7 @@ public class EpochSyncTest
private static class Cluster
{
private static final int rf = 2;
private static final int rf = 3;
private static final ReplicationParams replication_params = ReplicationParams.simple(rf);
private final RandomSource rs;
@ -290,7 +294,8 @@ public class EpochSyncTest
private boolean hasPendingWork()
{
return !status(s -> s == Cluster.Status.Registered).isEmpty()
|| !cms.metadata().inProgressSequences.isEmpty();
|| !cms.metadata().inProgressSequences.isEmpty()
|| globalExecutor.hasWork();
}
private boolean hasNoPendingWork()
@ -334,7 +339,7 @@ public class EpochSyncTest
return metadata.placements.get(replication_params).reads.byEndpoint().keySet().contains(address.broadcastAddress);
}
public enum EpochTracker { topologyManager, accordSyncPropagator, configurationService}
public enum EpochTracker { topologyManager, accordSyncPropagator }
Set<EpochTracker> globalSynced(long epoch)
{
@ -376,38 +381,34 @@ public class EpochSyncTest
{
Instance inst = instances.get(id);
if (removed.contains(id)) continue; // ignore removed nodes
AccordConfigurationService conf = inst.config;
TopologyManager tm = inst.topology;
for (long epoch = Math.max(tm.firstNonEmpty(), inst.epoch.getEpoch()); epoch <= cms.metadata().epoch.getEpoch(); epoch++)
{
// validate config
EpochSnapshot snapshot = conf.getEpochSnapshot(epoch);
if (isDone)
{
Assertions.assertThat(snapshot).describedAs("node%s does not have epoch %d", id, epoch).isNotNull();
Assertions.assertThat(snapshot.syncStatus).isEqualTo(AccordConfigurationService.SyncStatus.COMPLETED);
Assertions.assertThat(inst.topologyService.syncPropagator().hasPending(epoch))
.describedAs("node%s has pending notifications in AccordSyncPropagator for epoch %d", id, epoch)
.isFalse();
// validate topology manager
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Ranges ranges = tm.globalForEpoch(epoch).ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
Ranges ranges = tm.active().globalForEpoch(epoch).ranges().mergeTouching();
Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching();
Assertions.assertThat(actual)
.describedAs("node%s does not have all expected sync ranges for epoch %d; missing %s", id, epoch, ranges.without(actual))
.isEqualTo(ranges);
}
else
{
if (snapshot == null || snapshot.syncStatus != AccordConfigurationService.SyncStatus.COMPLETED) continue;
if (!allSynced(epoch))
continue;
Assertions.assertThat(tm.hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Topology topology = tm.globalForEpoch(epoch);
Assertions.assertThat(tm.active().hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Topology topology = tm.active().globalForEpoch(epoch);
Ranges ranges = topology.ranges().mergeTouching();
Ranges actual = tm.syncComplete(epoch).mergeTouching();
Ranges actual = tm.unsafeQuorumReady(epoch).mergeTouching();
// TopologyManager defines syncComplete for an epoch as (epoch - 1).syncComplete. This means that an epoch has reached quorum, but will still miss ranges as previous epochs have not
if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.syncComplete(epoch - 1).mergeTouching()))
if (!ranges.equals(actual) && tm.minEpoch() != epoch && !ranges.equals(tm.unsafeQuorumReady(epoch - 1).mergeTouching()))
continue;
long epoch_ = epoch;
Assertions.assertThat(actual)
@ -415,11 +416,8 @@ public class EpochSyncTest
{
public String value()
{
return String.format("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s; previous epochs %s",
id, epoch_, ranges.without(actual), topology.nodes(),
LongStream.range(inst.epoch.getEpoch(), epoch_ + 1)
.mapToObj(e -> String.format("%d -> %s(synced=%s): %s", e, conf.getEpochSnapshot(e).syncStatus, globalSynced(e), tm.syncComplete(e)))
.collect(Collectors.joining("\n")));
return String.format("node%s does not have all expected sync ranges for epoch %d; missing %s; peers=%s",
id, epoch_, ranges.without(actual), topology.nodes());
}
})
@ -548,7 +546,7 @@ public class EpochSyncTest
return Action.DELIVER;
},
SimulatedMessageDelivery.randomDelay(rs.fork()),
(to, msg) -> instances.get(nodeId(to)).reciver.recieve(msg),
(to, msg) -> instances.get(nodeId(to)).receiver.recieve(msg),
(action, to, msg) -> logger.trace("{} message {}", action, msg),
scheduler::schedule,
failures::add);
@ -594,6 +592,7 @@ public class EpochSyncTest
Instance inst = Objects.requireNonNull(instances.get(pick), "Unknown id " + pick);
Invariants.require(!removed.contains(pick), "Can not remove node twice; node " + pick);
removed.add(pick);
Assertions.assertThat(inst.topologyService.syncPropagator().hasPending()).isFalse();
inst.status = Status.Leaving;
PrepareLeave prepareLeave = new PrepareLeave(new NodeId(pick.id), false, new UniformRangePlacement(), LeaveStreams.Kind.REMOVENODE);
notify(process(prepareLeave).metadata);
@ -615,7 +614,8 @@ public class EpochSyncTest
{
Instance inst = instances.get(id);
inst.maybeTransition(current, t);
inst.config.maybeReportMetadata(current);
Topology topology = AccordTopology.createAccordTopology(current);
inst.topology.reportTopology(topology);
}
}
@ -650,72 +650,44 @@ public class EpochSyncTest
{
private final Node.Id id;
private final long token;
private final AccordConfigurationService config;
private final AccordTopologyService topologyService;
private final SimulatedMessageDelivery messaging;
private final SimulatedMessageDelivery.SimulatedMessageReceiver reciver;
private final SimulatedMessageDelivery.SimulatedMessageReceiver receiver;
private final TopologyManager topology;
private final Epoch epoch;
private Status status = Status.Init;
Instance(Node.Id node, long token, Epoch epoch, SimulatedMessageDelivery messagingService, AccordEndpointMapper mapper)
Instance(Node.Id id, long token, Epoch epoch, SimulatedMessageDelivery messagingService, AccordEndpointMapper mapper)
{
this.id = node;
this.id = id;
this.token = token;
this.epoch = epoch;
MockCluster.Clock clock = new MockCluster.Clock(0);
Node node = Utils.createNode(id, ignore -> AsyncResults.settable(), new MessageSink.NoOpSink(), clock, new TestAgent(clock), new TokenKey.KeyspaceSplitter(new ShardDistributor.EvenSplit<>(getAccordCommandStoreShardCount(), getPartitioner().accordSplitter())));
// TODO (review): Should there be a real scheduler here? Is it possible to adapt the Scheduler interface to scheduler used in this test?
TimeService time = TimeService.ofNonMonotonic(globalExecutor::currentTimeMillis, TimeUnit.MILLISECONDS);
this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, new TestAgent.RethrowAgent(), id, time, new DefaultTimeouts(time));
config = new AccordConfigurationService(node, new AccordAgent(), mapper, messagingService, scheduler);
config.registerListener(new ConfigurationService.Listener()
this.topologyService = new AccordTopologyService(id, mapper, messagingService, scheduler);
this.topology = new TopologyManager(SizeOfIntersectionSorter.SUPPLIER, node, topologyService, time, new DefaultTimeouts(time))
{
@Override
public AsyncResult<Void> onTopologyUpdate(Topology topology)
protected EpochReady bootstrap(Supplier<EpochReady> bootstrap)
{
bootstrap.get();
AsyncResult<Void> metadata = schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult();
AsyncResult<Void> coordination = metadata.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult());
AsyncResult<Void> data = coordination.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult());
AsyncResult<Void> reads = data.flatMap(ignore -> schedule(rs.nextInt(1, 10), TimeUnit.SECONDS, (Callable<Void>) () -> null).beginAsResult());
EpochReady ready = new EpochReady(topology.epoch(), metadata, coordination, data, reads);
topology().onTopologyUpdate(topology, () -> ready, e -> {});
ready.coordinate.invokeIfSuccess(() -> topology().onEpochSyncComplete(id, topology.epoch()));
if (topology().minEpoch() == topology.epoch() && topology().epoch() != topology.epoch())
return ready.coordinate;
config.acknowledgeEpoch(ready);
return ready.coordinate;
return new EpochReady(topology.epoch(), metadata, coordination, data, reads);
}
@Override
public void onRemoteSyncComplete(Node.Id node, long epoch)
{
topology.onEpochSyncComplete(node, epoch);
}
@Override
public void onRemoveNode(long epoch, Node.Id removed)
{
// TODO
//topology.onRemoveNode(epoch, removed);
}
@Override
public void onEpochClosed(Ranges ranges, long epoch)
{
topology.onEpochClosed(ranges, epoch);
}
@Override
public void onEpochRetired(Ranges ranges, long epoch)
{
topology.onEpochRetired(ranges, epoch);
}
});
};
this.topology.addListener(topologyService.syncPropagator());
this.topology.addListener(topologyService.watermarkCollector());
Map<Verb, IVerbHandler<?>> handlers = new EnumMap<>(Verb.class);
//noinspection unchecked
handlers.put(Verb.ACCORD_SYNC_NOTIFY_REQ, msg -> AccordService.receive(messagingService, config, (Message<AccordSyncPropagator.Notification>) (Message<?>) msg));
handlers.put(Verb.ACCORD_SYNC_NOTIFY_REQ, msg -> AccordService.receive(messagingService, topology, (Message<AccordSyncPropagator.Notification>) (Message<?>) msg));
this.messaging = messagingService;
this.reciver = messagingService.receiver(new SimulatedMessageDelivery.SimpleVerbHandler(handlers));
this.receiver = messagingService.receiver(new SimulatedMessageDelivery.SimpleVerbHandler(handlers));
}
@Override
@ -736,7 +708,6 @@ public class EpochSyncTest
case Init:
Invariants.require(!t.nodes().contains(id), "Node was in Init state but present in the Topology!");
Invariants.require(current.directory.peerId(address(id)) != null, "Node exists but not in TCM");
start();
status = Status.Registered;
break;
case Registered:
@ -762,26 +733,13 @@ public class EpochSyncTest
}
}
private void start()
{
config.start();
}
TopologyManager topology()
{
return topology;
}
Set<EpochTracker> synced(long epoch)
{
if (epoch < this.epoch.getEpoch()) throw new IllegalArgumentException("Asked for epoch before this instance existed");
EnumSet<EpochTracker> done = EnumSet.noneOf(EpochTracker.class);
EpochSnapshot snapshot = config.getEpochSnapshot(epoch);
if (snapshot != null && snapshot.syncStatus == AccordConfigurationService.SyncStatus.COMPLETED)
done.add(EpochTracker.configurationService);
if (topology.hasReachedQuorum(epoch))
if (topology.unsafeIsQuorumReady(epoch))
done.add(EpochTracker.topologyManager);
if (!config.syncPropagator().hasPending(epoch))
if (!topologyService.syncPropagator().hasPending(epoch))
done.add(EpochTracker.accordSyncPropagator);
return done;
}

1
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -674,6 +674,7 @@ public class CommandsForKeySerializerTest
@Override public long slowReplicaDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil blockedUntil, TimeUnit units) { return 0; }
@Override public long slowAwaitDelay(Node node, SafeCommandStore safeStore, TxnId txnId, int retryCount, ProgressLog.BlockedUntil retrying, TimeUnit units) { return 0; }
@Override public long retrySyncPointDelay(Node node, int attempt, TimeUnit units) { return 0; }
@Override public long retryTopologyDelay(Node node, int attempt, TimeUnit units) { return 0; }
@Override public long retryDurabilityDelay(Node node, int attempt, TimeUnit units) { return 0; }
@Override public long expireEpochWait(TimeUnit units) { return 0; }
@Override public long expiresAt(ReplyContext replyContext, TimeUnit unit) { return 0; }

8
test/unit/org/apache/cassandra/utils/AccordGenerators.java
View File

@ -63,7 +63,7 @@ import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.topology.Shard;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.topology.TopologyRange;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
@ -807,7 +807,7 @@ public class AccordGenerators
};
}
public static Gen<TopologyManager.TopologyRange> topologyRangeGen()
public static Gen<TopologyRange> topologyRangeGen()
{
Gen.LongGen epochGen = AccordGens.epochs();
return rs -> {
@ -823,7 +823,7 @@ public class AccordGenerators
if (minEpoch == Timestamp.MAX_EPOCH)
{
// not possible to have a list of values, so to simplfiy just return empty
return new TopologyManager.TopologyRange(Timestamp.MAX_EPOCH, Timestamp.MAX_EPOCH, -1, Collections.emptyList());
return new TopologyRange(Timestamp.MAX_EPOCH, Timestamp.MAX_EPOCH, -1, Collections.emptyList());
}
long epochsRemaining = Timestamp.MAX_EPOCH - minEpoch;
int size = rs.nextInt(1, Math.toIntExact(Math.min(100, epochsRemaining)));
@ -841,7 +841,7 @@ public class AccordGenerators
firstNonEmpty = t.epoch();
topologies.add(t);
}
return new TopologyManager.TopologyRange(topologies.get(0).epoch(), topologies.get(topologies.size() - 1).epoch(), firstNonEmpty, topologies);
return new TopologyRange(topologies.get(0).epoch(), topologies.get(topologies.size() - 1).epoch(), firstNonEmpty, topologies);
};
}