Make TopologyException a checked exception to ensure they are handled carefully, as they may occur at surprising times

Also Fix:
 - Restore MaxDecidedRX on replay
 - When catchup_on_start_exit_on_failure == false, should startup on any kind of failure, not only timeout
 - lazy vtable LIMIT clause regression
 - DurabilityService.onEpochRetired
 - Command.validate when uniqueHlc differs
 - Avoid unsafe publication of AccordExecutor to scheduledFastTasks
 - AccordCache hitRate metric names
 - use long for return type of DurationSpec.toNanoseconds
 - Repair without all replicas should not request all Accord replicas participate
 - ExecuteAtSerializer
 - SyncPoints should be coordinated in an epoch that contains the ranges
Also Improve:
 - Split Accord startup into local+distributed, ensure we
 - Add logging to FetchDurableBefore on startup
 - Add randomised testing of AbstractLazyVirtualTable
 - Add validation of lazy virtual table key ordering
 - Don't send requests to faulty replicas
 - shrinkOrEvict large objects without holding lock
 - Accord dtest shutdown

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21042
This commit is contained in:
Benedict Elliott Smith 2025-11-13 12:07:51 +00:00
parent dbfcd52fd5
commit 4defbb9b1c
90 changed files with 1146 additions and 432 deletions

@ -1 +1 @@
Subproject commit 938ba19adaf70bf9901fb2dc177dce313d7ee15c
Subproject commit 973335bfe7e930c6646016a4a48cff084f49b660

View File

@ -228,6 +228,7 @@ public enum CassandraRelevantProperties
DTEST_ACCORD_ENABLED("jvm_dtest.accord.enabled", "true"),
DTEST_ACCORD_JOURNAL_SANITY_CHECK_ENABLED("jvm_dtest.accord.journal_sanity_check_enabled", "false"),
DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
DTEST_IGNORE_SHUTDOWN_THREADCOUNT("jvm_dtests.ignore_shutdown_threadcount"),
/** This property indicates if the code is running under the in-jvm dtest framework */
DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
/** In_JVM dtest property indicating that the test should use "latest" configuration */

View File

@ -587,13 +587,11 @@ public abstract class DurationSpec
}
/**
* Returns this duration in the number of nanoseconds as an {@code int}
*
* @return this duration in number of nanoseconds or {@code Integer.MAX_VALUE} if the number of nanoseconds is too large.
* Returns this duration in the number of nanoseconds as a {@code long}
*/
public int toNanoseconds()
public long toNanoseconds()
{
return Ints.saturatedCast(unit().toNanos(quantity()));
return unit().toNanos(quantity());
}
/**

View File

@ -181,7 +181,14 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
TopPartitionTracker.Collector topPartitionCollector)
{
this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector,
AccordService.isSetup() ? AccordService.instance() : null);
accord(controller));
}
private static IAccordService accord(AbstractCompactionController controller)
{
IAccordService accord = AccordService.tryGetUnsafe();
Invariants.require(accord != null || (!isAccordJournal(controller.cfs) && !isAccordCommandsForKey(controller.cfs)));
return accord;
}
public CompactionIterator(OperationType type,
@ -194,7 +201,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
IAccordService accord)
{
this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector,
() -> accord.getCompactionInfo(),
() -> Invariants.nonNull(accord).getCompactionInfo(),
() -> Version.fromVersion(accord.journalConfiguration().userVersion()));
}

View File

@ -24,6 +24,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;
import java.util.NoSuchElementException;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
@ -223,6 +224,8 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (!dataRange.contains(partitionKey))
return dropCks -> {};
return partitions.computeIfAbsent(partitionKey, SimplePartition::new);
}
@ -232,22 +235,23 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
final Iterator<SimplePartition> partitions = this.partitions.values().iterator();
return new UnfilteredPartitionIterator()
{
private UnfilteredRowIterator next;
@Override public TableMetadata metadata() { return metadata; }
@Override public void close() {}
@Override
public boolean hasNext()
{
return partitions.hasNext();
}
@Override
public UnfilteredRowIterator next()
while (next == null && partitions.hasNext())
{
SimplePartition partition = partitions.next();
Iterator<Row> rows = partition.rows();
return new UnfilteredRowIterator()
if (!rows.hasNext())
continue;
next = new UnfilteredRowIterator()
{
@Override public TableMetadata metadata() { return metadata; }
@Override public boolean isReverseOrder() { return dataRange.isReversed(); }
@ -263,6 +267,19 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
@Override public EncodingStats stats() { return EncodingStats.NO_STATS; }
};
}
return next != null;
}
@Override
public UnfilteredRowIterator next()
{
if (!hasNext())
throw new NoSuchElementException();
UnfilteredRowIterator result = next;
next = null;
return result;
}
};
}
@ -312,9 +329,23 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (!dataRange.contains(partitionKey) || !dataRange.clusteringIndexFilter(partitionKey).selects(clustering))
return drop -> {};
if (isSortedByPartitionKey)
checkCorrectlySorted(partitionKey);
return partitions.computeIfAbsent(partitionKey, SimplePartition::new).row(clustering);
}
private void checkCorrectlySorted(DecoratedKey newPartitionKey)
{
if (partitions.isEmpty())
return;
DecoratedKey prevKey = partitions.lastKey();
int c = metadata.partitionKeyType.compare(prevKey.getKey(), newPartitionKey.getKey());
if (dataRange.isReversed() ? c < 0 : c > 0)
throw new IllegalArgumentException(Arrays.toString(composePartitionKeys(prevKey, metadata)) + (dataRange.isReversed() ? " < " : " > ") + Arrays.toString(composePartitionKeys(newPartitionKey, metadata)));
}
private final class SimplePartition implements PartitionCollector, RowsCollector
{
private final DecoratedKey key;
@ -323,6 +354,7 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
private int rowCount;
private SimpleRow staticRow;
private boolean dropRows;
private boolean isSortedAndFiltered = true;
private SimplePartition(DecoratedKey key)
{
@ -346,6 +378,17 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
return row(decomposeClusterings(metadata, clusteringKeys));
}
private void checkCorrectlySorted(Clustering<?> newClustering)
{
if (rowCount == 0)
return;
Clustering<?> prevClustering = rows[rowCount - 1].clustering;
int c = metadata.comparator.compare(prevClustering, newClustering);
if (dataRange.isReversed() ? c <= 0 : c >= 0)
throw new IllegalArgumentException(Arrays.toString(composeClusterings(prevClustering, metadata)) + (dataRange.isReversed() ? " <= " : " >= ") + Arrays.toString(composeClusterings(newClustering, metadata)));
}
RowCollector row(Clustering<?> clustering)
{
if (nanoTime() > deadlineNanos)
@ -354,6 +397,9 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (dropRows || !dataRange.clusteringIndexFilter(key).selects(clustering))
return drop -> {};
if (isSorted)
checkCorrectlySorted(clustering);
if (totalRowCount >= limits.count())
{
boolean filter;
@ -370,9 +416,13 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (filter)
{
// first filter within each partition
for (SimplePartition partition : partitions.values())
{
// first filter within each partition
partition.filterAndSort();
// and truncate if there are per-partition limits
partition.truncate(limits.perPartitionCount());
}
// then drop any partitions that completely fall outside our limit
Iterator<SimplePartition> iter = partitions.descendingMap().values().iterator();
@ -418,12 +468,16 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
if (rowCount == rows.length)
rows = Arrays.copyOf(rows, Math.max(8, rowCount * 2));
rows[rowCount++] = result;
isSortedAndFiltered = false;
}
return result;
}
void filterAndSort()
{
if (isSortedAndFiltered)
return;
int newCount = 0;
for (int i = 0 ; i < rowCount; ++i)
{
@ -441,6 +495,7 @@ public abstract class AbstractLazyVirtualTable implements VirtualTable
rowCount = newCount;
}
Arrays.sort(rows, 0, newCount, rowComparator());
isSortedAndFiltered = true;
}
int truncate(int newCount)

View File

@ -946,7 +946,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
collector.partition(commandStore.id()).collect(collect -> {
while (view.advance())
{
// TODO (required): view should return an immutable per-row view so that we can call lazyAdd
// TODO (desired): view should return an immutable per-row view so that we can call lazyAdd
collect.add(view.txnId().toString())
.eagerCollect(columns -> {
columns.add("table_id", tableIdStr)
@ -1516,7 +1516,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
AbstractJournalTable(TableMetadata metadata)
{
super(metadata, FAIL, ASC);
super(metadata, FAIL, UNSORTED, ASC);
}
@Override

View File

@ -466,7 +466,7 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
// TODO (required): Find a better way to test unwritten allocations and/or corruption
// TODO (expected): Find a better way to test unwritten allocations and/or corruption
@VisibleForTesting
void consumeBufferUnsafe(Consumer<ByteBuffer> fn)
{

View File

@ -24,6 +24,9 @@ import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
import org.apache.cassandra.concurrent.Shutdownable;
@ -31,6 +34,8 @@ import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFac
public final class Compactor<K, V> implements Runnable, Shutdownable
{
private static final Logger logger = LoggerFactory.getLogger(Compactor.class);
private final Journal<K, V> journal;
private final SegmentCompactor<K, V> segmentCompactor;
private final ScheduledExecutorPlus executor;
@ -99,6 +104,7 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
@Override
public void shutdown()
{
logger.debug("Shutting down " + executor);
executor.shutdown();
}

View File

@ -102,10 +102,12 @@ final class Flusher<K, V>
void shutdown() throws InterruptedException
{
logger.debug("Shutting down " + flushExecutor + " and awaiting termination");
flushExecutor.shutdown();
flushExecutor.awaitTermination(1, MINUTES);
if (fsyncExecutor != null)
{
logger.debug("Shutting down " + fsyncExecutor + " and awaiting termination");
fsyncExecutor.shutdownNow(); // `now` to interrupt potentially parked runnable
fsyncExecutor.awaitTermination(1, MINUTES);
}

View File

@ -114,7 +114,7 @@ public class Journal<K, V> implements Shutdownable
final AtomicReference<State> state = new AtomicReference<>(State.UNINITIALIZED);
// TODO (required): we do not need wait queues here, we can just wait on a signal on a segment while its byte buffer is being allocated
// TODO (expected): we do not need wait queues here, we can just wait on a signal on a segment while its byte buffer is being allocated
private final WaitQueue segmentPrepared = newWaitQueue();
private final WaitQueue allocatorThreadWaitQueue = newWaitQueue();
private final BooleanSupplier allocatorThreadWaitCondition = () -> (availableSegment == null);
@ -256,6 +256,7 @@ public class Journal<K, V> implements Shutdownable
{
Invariants.require(state.compareAndSet(State.NORMAL, State.SHUTDOWN),
"Unexpected journal state while trying to shut down", state);
logger.debug("Shutting down " + allocator + " and awaiting termination");
allocator.shutdown();
wakeAllocator(); // Wake allocator to force it into shutdown
// TODO (expected): why are we awaitingTermination here when we have a separate method for it?
@ -265,6 +266,7 @@ public class Journal<K, V> implements Shutdownable
compactor.awaitTermination(1, TimeUnit.MINUTES);
flusher.shutdown();
closeAllSegments();
logger.debug("Shutting down " + releaser + " and " + closer + " and awaiting termination");
releaser.shutdown();
closer.shutdown();
closer.awaitTermination(1, TimeUnit.MINUTES);

View File

@ -94,13 +94,13 @@ public class AccordCacheMetrics
this.hits = Metrics.gauge(factory.createMetricName("Hits"), hitRate::totalHits);
this.misses = Metrics.gauge(factory.createMetricName("Misses"), hitRate::totalMisses);
this.requests = Metrics.gauge(factory.createMetricName("Requests"), hitRate::totalRequests);
this.requestRate1m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(1));
this.requestRate5m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(5));
this.requestRate1m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.ONE_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(1));
this.requestRate5m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIVE_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(5));
this.requestRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(15));
this.hitRate1m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.ONE_MINUTE + "HitRate"), () -> hitRate.hitRate(1));
this.hitRate5m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIVE_MINUTE + "HitRate"), () -> hitRate.hitRate(5));
this.hitRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "HitRate"), () -> hitRate.hitRate(15));
this.hitRateAllTime = Metrics.gauge(factory.createMetricName("Misses"), hitRate::hitRateAllTime);
this.hitRateAllTime = Metrics.gauge(factory.createMetricName("HitRate"), hitRate::hitRateAllTime);
this.subTypeName = subTypeName;
}

View File

@ -45,9 +45,9 @@ public class AccordMetricUtils
static <I, T> Gauge<T> fromAccordService(Function<IAccordService, I> initialStep, Function<I, T> finalStep, T ifNotSetup)
{
if (AccordService.isSetup())
if (AccordService.isSetupOrStarting())
{
IAccordService service = AccordService.instance();
IAccordService service = AccordService.unsafeInstance();
if (!service.isEnabled())
return () -> ifNotSetup;
I intermediate = initialStep.apply(service);

View File

@ -138,7 +138,7 @@ public class AccordSystemMetrics
private AccordSystemMetrics()
{
Invariants.expect(AccordService.isSetup());
Invariants.expect(AccordService.isSetupOrStarting());
DefaultNameFactory factory = new DefaultNameFactory(ACCORD_SYSTEM);
minEpoch = Metrics.gauge(factory.createMetricName(MIN_EPOCH), fromTopologyManager(TopologyManager::minEpoch));
maxEpoch = Metrics.gauge(factory.createMetricName(MAX_EPOCH), fromTopologyManager(TopologyManager::epoch));
@ -164,13 +164,13 @@ public class AccordSystemMetrics
private synchronized void refreshHistograms()
{
if (!AccordService.isSetup())
if (!AccordService.isSetupOrStarting())
{
snapshot = new Snapshot();
return;
}
IAccordService service = AccordService.instance();
IAccordService service = AccordService.unsafeInstance();
if (!service.isEnabled())
{
snapshot = new Snapshot();

View File

@ -174,7 +174,7 @@ public interface MessageDelivery
public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
long retryDelay = backoff.computeWait(attempt + 1, NANOSECONDS);
// TODO (required): we already have a separate retry predicate, retries should not be taken into consideration when retrying
// TODO (expected): we already have a separate retry predicate, retries should not be taken into consideration when retrying
if (retryDelay < 0)
{
onResult.result(attempt, null, new GivingUpException(attempt, errorMessage.apply(attempt, ResponseFailureReason.GiveUp, from, failure)));

View File

@ -355,10 +355,10 @@ public enum Verb
ACCORD_GET_LATEST_DEPS_REQ (151, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(LatestDepsSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_LATEST_DEPS_RSP),
ACCORD_GET_MAX_CONFLICT_RSP (152, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.reply), AccordService::responseHandlerOrNoop ),
ACCORD_GET_MAX_CONFLICT_REQ (153, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP),
ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ),
ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ),
ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ),
ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ),
ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, IMMEDIATE, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_SYNC_NOTIFY_RSP (158, P2, writeTimeout, MISC, () -> accordEmbedded(EnumSerializer.simpleReply), RESPONSE_HANDLER),
ACCORD_SYNC_NOTIFY_REQ (159, P2, writeTimeout, MISC, () -> accordEmbedded(Notification.serializer), () -> AccordSyncPropagator.verbHandler, ACCORD_SYNC_NOTIFY_RSP ),

View File

@ -31,6 +31,7 @@ import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.RepairCoordinator.NeighborsAndRanges;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Future;
@ -44,13 +45,15 @@ public abstract class AbstractRepairTask implements RepairTask
protected final InetAddressAndPort broadcastAddressAndPort;
protected final RepairOption options;
protected final String keyspace;
protected final NeighborsAndRanges neighborsAndRanges;
protected AbstractRepairTask(RepairCoordinator coordinator)
protected AbstractRepairTask(RepairCoordinator coordinator, NeighborsAndRanges neighborsAndRanges)
{
this.coordinator = Objects.requireNonNull(coordinator);
this.broadcastAddressAndPort = coordinator.ctx.broadcastAddressAndPort();
this.options = Objects.requireNonNull(coordinator.state.options);
this.keyspace = Objects.requireNonNull(coordinator.state.keyspace);
this.neighborsAndRanges = neighborsAndRanges;
}
private List<RepairSession> submitRepairSessions(TimeUUID parentSession,
@ -71,6 +74,7 @@ public abstract class AbstractRepairTask implements RepairTask
excludedDeadNodes,
keyspace,
options.getParallelism(),
neighborsAndRanges.includesAllReplicas,
isIncremental,
options.isPullRepair(),
options.getPreviewKind(),
@ -79,6 +83,7 @@ public abstract class AbstractRepairTask implements RepairTask
options.repairPaxos(),
options.dontPurgeTombstones(),
options.repairAccord(),
options.permitNoQuorum(),
executor,
validationScheduler,
cfnames);

View File

@ -31,7 +31,6 @@ import org.apache.cassandra.utils.concurrent.Future;
public class IncrementalRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
private final RepairCoordinator.NeighborsAndRanges neighborsAndRanges;
private final String[] cfnames;
protected IncrementalRepairTask(RepairCoordinator coordinator,
@ -39,9 +38,8 @@ public class IncrementalRepairTask extends AbstractRepairTask
RepairCoordinator.NeighborsAndRanges neighborsAndRanges,
String[] cfnames)
{
super(coordinator);
super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
this.neighborsAndRanges = neighborsAndRanges;
this.cfnames = cfnames;
}
@ -60,7 +58,7 @@ public class IncrementalRepairTask extends AbstractRepairTask
.add(broadcastAddressAndPort)
.build();
// Not necessary to include self for filtering. The common ranges only contains neighbhor node endpoints.
List<CommonRange> allRanges = neighborsAndRanges.filterCommonRanges(keyspace, cfnames);
List<CommonRange> allRanges = neighborsAndRanges.filterCommonRanges(keyspace, cfnames).commonRanges;
CoordinatorSession coordinatorSession = coordinator.ctx.repair().consistent.coordinated.registerSession(parentSession, allParticipants, neighborsAndRanges.shouldExcludeDeadParticipants);

View File

@ -17,8 +17,6 @@
*/
package org.apache.cassandra.repair;
import java.util.List;
import org.apache.cassandra.concurrent.ExecutorPlus;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.Future;
@ -26,20 +24,15 @@ import org.apache.cassandra.utils.concurrent.Future;
public class NormalRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
private final List<CommonRange> commonRanges;
private final boolean excludedDeadNodes;
private final String[] cfnames;
protected NormalRepairTask(RepairCoordinator coordinator,
TimeUUID parentSession,
List<CommonRange> commonRanges,
boolean excludedDeadNodes,
RepairCoordinator.NeighborsAndRanges neighborsAndRanges,
String[] cfnames)
{
super(coordinator);
super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
this.commonRanges = commonRanges;
this.excludedDeadNodes = excludedDeadNodes;
this.cfnames = cfnames;
}
@ -52,6 +45,6 @@ public class NormalRepairTask extends AbstractRepairTask
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
return runRepair(parentSession, false, executor, validationScheduler, commonRanges, excludedDeadNodes, cfnames);
return runRepair(parentSession, false, executor, validationScheduler, neighborsAndRanges.commonRanges, neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
}
}

View File

@ -42,17 +42,13 @@ import org.apache.cassandra.utils.concurrent.Future;
public class PreviewRepairTask extends AbstractRepairTask
{
private final TimeUUID parentSession;
private final List<CommonRange> commonRanges;
private final boolean excludedDeadNodes;
private final String[] cfnames;
private volatile String successMessage = name() + " completed successfully";
protected PreviewRepairTask(RepairCoordinator coordinator, TimeUUID parentSession, List<CommonRange> commonRanges, boolean excludedDeadNodes, String[] cfnames)
protected PreviewRepairTask(RepairCoordinator coordinator, TimeUUID parentSession, RepairCoordinator.NeighborsAndRanges neighborsAndRanges, String[] cfnames)
{
super(coordinator);
super(coordinator, neighborsAndRanges);
this.parentSession = parentSession;
this.commonRanges = commonRanges;
this.excludedDeadNodes = excludedDeadNodes;
this.cfnames = cfnames;
}
@ -71,7 +67,7 @@ public class PreviewRepairTask extends AbstractRepairTask
@Override
public Future<CoordinatedRepairResult> performUnsafe(ExecutorPlus executor, Scheduler validationScheduler)
{
Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, validationScheduler, commonRanges, excludedDeadNodes, cfnames);
Future<CoordinatedRepairResult> f = runRepair(parentSession, false, executor, validationScheduler, neighborsAndRanges.commonRanges, neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
return f.map(result -> {
if (result.hasFailed())
return result;

View File

@ -394,6 +394,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
private NeighborsAndRanges getNeighborsAndRanges() throws RepairException
{
Set<InetAddressAndPort> allNeighbors = new HashSet<>();
Set<InetAddressAndPort> includeNeighbors = new HashSet<>();
List<CommonRange> commonRanges = new ArrayList<>();
//pre-calculate output of getLocalReplicas and pass it to getNeighbors to increase performance and prevent
@ -403,10 +404,14 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
boolean isCMS = ClusterMetadata.current().isCMSMember(FBUtilities.getBroadcastAddressAndPort());
for (Range<Token> range : state.options.getRanges())
{
EndpointsForRange neighbors = ctx.repair().getNeighbors(state.keyspace, keyspaceLocalRanges, range,
EndpointsForRange allForRange = ctx.repair().getNeighbors(state.keyspace, keyspaceLocalRanges, range);
allNeighbors.addAll(allForRange.endpoints());
EndpointsForRange includeForRange = ctx.repair().filterNeighbors(allForRange, range,
state.options.getDataCenters(),
state.options.getHosts());
if (neighbors.isEmpty())
if (includeForRange.isEmpty())
{
if (state.options.ignoreUnreplicatedKeyspaces())
{
@ -423,11 +428,11 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
throw RepairException.warn(String.format("Nothing to repair for %s in %s - aborting", range, state.keyspace));
}
}
addRangeToNeighbors(commonRanges, range, neighbors);
allNeighbors.addAll(neighbors.endpoints());
addRangeToNeighbors(commonRanges, range, includeForRange);
includeNeighbors.addAll(includeForRange.endpoints());
}
if (allNeighbors.isEmpty())
if (includeNeighbors.isEmpty())
{
if (state.options.ignoreUnreplicatedKeyspaces())
{
@ -447,11 +452,12 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
if (shouldExcludeDeadParticipants)
{
Set<InetAddressAndPort> actualNeighbors = Sets.newHashSet(Iterables.filter(allNeighbors, ctx.failureDetector()::isAlive));
shouldExcludeDeadParticipants = !allNeighbors.equals(actualNeighbors);
allNeighbors = actualNeighbors;
Set<InetAddressAndPort> actualNeighbors = Sets.newHashSet(Iterables.filter(includeNeighbors, ctx.failureDetector()::isAlive));
shouldExcludeDeadParticipants = !includeNeighbors.equals(actualNeighbors);
if (shouldExcludeDeadParticipants) includeNeighbors = actualNeighbors;
else logger.info("{} all replicas {} considered up and healthy; clearing force flag for this job", state.id, includeNeighbors);
}
return new NeighborsAndRanges(shouldExcludeDeadParticipants, allNeighbors, commonRanges);
return new NeighborsAndRanges(shouldExcludeDeadParticipants, includeNeighbors.containsAll(allNeighbors), includeNeighbors, commonRanges);
}
private void maybeStoreParentRepairStart(String[] cfnames)
@ -496,7 +502,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
RepairTask task;
if (state.options.isPreview())
{
task = new PreviewRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
task = new PreviewRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), cfnames);
}
else if (state.options.isIncremental())
{
@ -504,7 +510,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
}
else
{
task = new NormalRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), neighborsAndRanges.shouldExcludeDeadParticipants, cfnames);
task = new NormalRepairTask(this, state.id, neighborsAndRanges.filterCommonRanges(state.keyspace, cfnames), cfnames);
}
ExecutorPlus executor = createExecutor();
@ -635,12 +641,14 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
public static final class NeighborsAndRanges
{
final boolean shouldExcludeDeadParticipants;
public final boolean includesAllReplicas;
public final Set<InetAddressAndPort> participants;
public final List<CommonRange> commonRanges;
public NeighborsAndRanges(boolean shouldExcludeDeadParticipants, Set<InetAddressAndPort> participants, List<CommonRange> commonRanges)
public NeighborsAndRanges(boolean shouldExcludeDeadParticipants, boolean includesAllReplicas, Set<InetAddressAndPort> participants, List<CommonRange> commonRanges)
{
this.shouldExcludeDeadParticipants = shouldExcludeDeadParticipants;
this.includesAllReplicas = includesAllReplicas;
this.participants = participants;
this.commonRanges = commonRanges;
}
@ -650,11 +658,11 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
* and exludes ranges left without any participants
* When not in the force mode, no-op.
*/
public List<CommonRange> filterCommonRanges(String keyspace, String[] tableNames)
public NeighborsAndRanges filterCommonRanges(String keyspace, String[] tableNames)
{
if (!shouldExcludeDeadParticipants)
{
return commonRanges;
return this;
}
else
{
@ -682,7 +690,7 @@ public class RepairCoordinator implements Runnable, ProgressEventNotifier, Repai
}
}
Preconditions.checkState(!filtered.isEmpty(), "Not enough live endpoints for a repair");
return filtered;
return new NeighborsAndRanges(shouldExcludeDeadParticipants, includesAllReplicas, participants, filtered);
}
}
}

View File

@ -39,6 +39,7 @@ import com.google.common.util.concurrent.FutureCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.durability.DurabilityService.SyncRemote;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
@ -70,6 +71,9 @@ import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.local.durability.DurabilityService.SyncRemote.NoRemote;
import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static com.google.common.util.concurrent.Futures.getUnchecked;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosRepairEnabled;
import static org.apache.cassandra.schema.SchemaConstants.METADATA_KEYSPACE_NAME;
@ -179,20 +183,18 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
if (doAccordRepair)
{
accordRepair = paxosRepair.flatMap(unused -> {
boolean requireAllEndpoints;
// If the session excluded dead nodes it's not eligible for migration and is not supposed to occur at ALL anyways
if (session.excludedDeadNodes)
requireAllEndpoints = false;
SyncRemote sync;
// If the session is doing a data repair (which flushes sstables if not incremental) we can do the barriers at QUORUM
if (session.excludedDeadNodes || !session.allReplicas || (session.repairData && !session.isIncremental))
{
sync = session.permitNoQuorum ? NoRemote : Quorum;
}
else
{
// If the session is doing a data repair (which flushes sstables if not incremental) we can do the barriers at QUORUM
if (session.repairData && !session.isIncremental)
requireAllEndpoints = false;
else
requireAllEndpoints = true;
sync = All;
}
logger.info("{} {}.{} starting accord repair, require all endpoints {}", session.previewKind.logPrefix(session.getId()), desc.keyspace, desc.columnFamily, requireAllEndpoints);
AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, requireAllEndpoints, allEndpoints);
logger.info("{} {}.{} starting accord repair, sync {}", session.previewKind.logPrefix(session.getId()), desc.keyspace, desc.columnFamily, sync);
AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, sync, sync == All ? null : allEndpoints);
return repair.repair(taskExecutor).flatMap(accordRepairResult -> {
// Propagate the HLC discovered during Accord repair to Paxos so Paxos doesn't use ballots < Accord has already used
if (accordRepairResult.maxHlc != IAccordService.NO_HLC)

View File

@ -119,12 +119,14 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
/** Range to repair */
public final boolean isIncremental;
public final boolean allReplicas;
public final PreviewKind previewKind;
public final boolean repairData;
public final boolean repairPaxos; // TODO (now): rename to repairPaxosIfSupported
public final boolean repairPaxos;
public final boolean repairAccord;
public final boolean dontPurgeTombstones;
public final boolean excludedDeadNodes;
public final boolean permitNoQuorum;
private final AtomicBoolean isFailed = new AtomicBoolean(false);
@ -161,6 +163,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
boolean excludedDeadNodes,
String keyspace,
RepairParallelism parallelismDegree,
boolean allReplicas,
boolean isIncremental,
boolean pullRepair,
PreviewKind previewKind,
@ -168,7 +171,7 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
boolean repairData,
boolean repairPaxos,
boolean dontPurgeTombstones,
boolean repairAccord,
boolean repairAccord, boolean permitNoQuorum,
String... cfnames)
{
this.ctx = ctx;
@ -176,12 +179,14 @@ public class RepairSession extends AsyncFuture<RepairSessionResult> implements I
this.repairData = repairData;
this.repairPaxos = repairPaxos;
this.repairAccord = repairAccord;
this.permitNoQuorum = permitNoQuorum;
assert cfnames.length > 0 : "Repairing no column families seems pointless, doesn't it";
this.state = new SessionState(ctx, parentRepairSession, keyspace, cfnames, commonRange);
this.parallelismDegree = parallelismDegree;
this.isIncremental = isIncremental;
this.previewKind = previewKind;
this.pullRepair = pullRepair;
this.allReplicas = allReplicas && !commonRange.hasSkippedReplicas;
this.optimiseStreams = optimiseStreams;
this.dontPurgeTombstones = dontPurgeTombstones;
this.taskExecutor = new SafeExecutor(createExecutor(ctx));

View File

@ -52,7 +52,7 @@ import org.apache.cassandra.utils.TimeUUID;
*
* See {@link Global#instance} for the main production path
*/
// TODO (required, clarity): move under Util since this is a class with shared logic
// TODO (expected): move under Util since this is a class with shared logic
public interface SharedContext
{
InetAddressAndPort broadcastAddressAndPort();

View File

@ -311,7 +311,7 @@ class PreviewRepairedState extends AutoRepairState
public RepairCoordinator getRepairRunnable(String keyspace, List<String> tables, Set<Range<Token>> ranges, boolean primaryRangeOnly)
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false);
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges, false, false, PreviewKind.REPAIRED, false, true, true, false, false, false, false);
option.getColumnFamilies().addAll(tables);
@ -331,7 +331,7 @@ class IncrementalRepairState extends AutoRepairState
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, true, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges,
false, false, PreviewKind.NONE, true, true, true, false, false, false);
false, false, PreviewKind.NONE, true, true, true, false, false, false, false);
option.getColumnFamilies().addAll(filterOutUnsafeTables(keyspace, tables));
@ -376,7 +376,7 @@ class FullRepairState extends AutoRepairState
{
RepairOption option = new RepairOption(RepairParallelism.PARALLEL, primaryRangeOnly, false, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), ranges,
false, false, PreviewKind.NONE, true, true, true, false, false, false);
false, false, PreviewKind.NONE, true, true, true, false, false, false, false);
option.getColumnFamilies().addAll(tables);

View File

@ -63,6 +63,7 @@ public class RepairOption
public static final String REPAIR_PAXOS_KEY = "repairPaxos";
public static final String NO_TOMBSTONE_PURGING = "nopurge";
public static final String REPAIR_ACCORD_KEY = "repairAccord";
public static final String PERMIT_NO_QUORUM_KEY = "permitNoQuorum";
// we don't want to push nodes too much for repair
public static final int MAX_JOB_THREADS = 4;
@ -209,6 +210,7 @@ public class RepairOption
logger.info("Overriding and disabling Accord repair because Accord is not enabled");
repairAccord = false;
}
boolean permitNoQuorum = Boolean.parseBoolean(options.get(PERMIT_NO_QUORUM_KEY));
if (previewKind != PreviewKind.NONE)
{
@ -233,7 +235,7 @@ public class RepairOption
boolean asymmetricSyncing = Boolean.parseBoolean(options.get(OPTIMISE_STREAMS_KEY));
RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, pullRepair, force, previewKind, asymmetricSyncing, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord);
RepairOption option = new RepairOption(parallelism, primaryRange, incremental, trace, jobThreads, ranges, pullRepair, force, previewKind, asymmetricSyncing, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord, permitNoQuorum);
// data centers
String dataCentersStr = options.get(DATACENTERS_KEY);
@ -316,13 +318,14 @@ public class RepairOption
private final boolean repairPaxos;
private final boolean dontPurgeTombstones;
private final boolean repairAccord;
private final boolean permitNoQuorum; // has an effect only for Accord at present
private final Collection<String> columnFamilies = new HashSet<>();
private final Collection<String> dataCenters = new HashSet<>();
private final Collection<String> hosts = new HashSet<>();
private final Collection<Range<Token>> ranges = new HashSet<>();
public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean pullRepair, boolean forceRepair, PreviewKind previewKind, boolean optimiseStreams, boolean ignoreUnreplicatedKeyspaces, boolean repairData, boolean repairPaxos, boolean dontPurgeTombstones, boolean repairAccord)
public RepairOption(RepairParallelism parallelism, boolean primaryRange, boolean incremental, boolean trace, int jobThreads, Collection<Range<Token>> ranges, boolean pullRepair, boolean forceRepair, PreviewKind previewKind, boolean optimiseStreams, boolean ignoreUnreplicatedKeyspaces, boolean repairData, boolean repairPaxos, boolean dontPurgeTombstones, boolean repairAccord, boolean permitNoQuorum)
{
checkArgument(repairData || repairAccord || repairPaxos, "Repair needs to repair at least one of data, Paxos, or Accord");
this.parallelism = parallelism;
@ -340,6 +343,7 @@ public class RepairOption
this.repairPaxos = repairPaxos;
this.dontPurgeTombstones = dontPurgeTombstones;
this.repairAccord = repairAccord;
this.permitNoQuorum = permitNoQuorum;
}
public RepairParallelism getParallelism()
@ -460,6 +464,11 @@ public class RepairOption
return repairAccord;
}
public boolean permitNoQuorum()
{
return repairAccord;
}
public boolean dontPurgeTombstones()
{
return dontPurgeTombstones;
@ -514,6 +523,7 @@ public class RepairOption
options.put(REPAIR_PAXOS_KEY, Boolean.toString(repairPaxos));
options.put(NO_TOMBSTONE_PURGING, Boolean.toString(dontPurgeTombstones));
options.put(REPAIR_ACCORD_KEY, Boolean.toString(repairAccord));
options.put(PERMIT_NO_QUORUM_KEY, Boolean.toString(permitNoQuorum));
return options;
}
}

View File

@ -132,7 +132,7 @@ public class CoordinatorState extends AbstractState<CoordinatorState.State, Time
{
if (neighborsAndRanges == null)
return null;
return neighborsAndRanges.filterCommonRanges(keyspace, getColumnFamilyNames());
return neighborsAndRanges.filterCommonRanges(keyspace, getColumnFamilyNames()).commonRanges;
}
@Override

View File

@ -458,6 +458,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
boolean excludedDeadNodes,
String keyspace,
RepairParallelism parallelismDegree,
boolean allReplicas,
boolean isIncremental,
boolean pullRepair,
PreviewKind previewKind,
@ -466,6 +467,7 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
boolean repairPaxos,
boolean dontPurgeTombstones,
boolean repairAccord,
boolean permitNoQuorum,
ExecutorPlus executor,
Scheduler validationScheduler,
String... cfnames)
@ -481,9 +483,9 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
final RepairSession session = new RepairSession(ctx, validationScheduler, parentRepairSession,
range, excludedDeadNodes, keyspace,
parallelismDegree, isIncremental, pullRepair,
parallelismDegree, allReplicas, isIncremental, pullRepair,
previewKind, optimiseStreams, repairData, repairPaxos,
dontPurgeTombstones, repairAccord, cfnames);
dontPurgeTombstones, repairAccord, permitNoQuorum, cfnames);
repairs.getIfPresent(parentRepairSession).register(session.state);
sessions.put(session.getId(), session);
@ -567,6 +569,19 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
public EndpointsForRange getNeighbors(String keyspaceName, Iterable<Range<Token>> keyspaceLocalRanges,
Range<Token> toRepair, Collection<String> dataCenters,
Collection<String> hosts)
{
return filterNeighbors(getNeighbors(keyspaceName, keyspaceLocalRanges, toRepair), toRepair, dataCenters, hosts);
}
/**
* Return all of the neighbors with whom we share the provided range.
*
* @param keyspaceName keyspace to repair
* @param keyspaceLocalRanges local-range for given keyspaceName
* @param toRepair token to repair
* @return neighbors with whom we share the provided range
*/
public EndpointsForRange getNeighbors(String keyspaceName, Iterable<Range<Token>> keyspaceLocalRanges, Range<Token> toRepair)
{
StorageService ss = StorageService.instance;
EndpointsByRange replicaSets = ss.getRangeToAddressMap(keyspaceName);
@ -589,7 +604,22 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
return EndpointsForRange.empty(toRepair);
// same as withoutSelf(), but done this way for testing
EndpointsForRange neighbors = replicaSets.get(rangeSuperSet).filter(r -> !ctx.broadcastAddressAndPort().equals(r.endpoint()));
return replicaSets.get(rangeSuperSet).filter(r -> !ctx.broadcastAddressAndPort().equals(r.endpoint()));
}
/**
* Return all of the neighbors in the listed data center or host lists
*
* @param toRepair token to repair
* @param dataCenters the data centers to involve in the repair
* @return neighbors with whom we share the provided range
*/
public EndpointsForRange filterNeighbors(EndpointsForRange neighbors, Range<Token> toRepair, Collection<String> dataCenters,
Collection<String> hosts)
{
if (neighbors.isEmpty())
return neighbors;
ClusterMetadata metadata = ClusterMetadata.current();
if (dataCenters != null && !dataCenters.isEmpty())

View File

@ -74,6 +74,7 @@ import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Locator;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.tcm.CMSOperations;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.RegistrationStatus;
@ -91,6 +92,7 @@ import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.MultiStepOperation;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JMXServerUtils;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -343,11 +345,12 @@ public class CassandraDaemon
PaxosState.initializeTrackers();
// replay the log if necessary
// TODO samt - when restarting a previously running instance, this needs to happen after reconstructing schema
// from the cluster metadata log or all mutations will throw IncompatibleSchemaException on deserialisation
try
{
CommitLog.instance.recoverSegmentsOnDisk();
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
}
catch (IOException e)
{

View File

@ -847,7 +847,10 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
Gossiper.waitToSettle();
NodeId self = Register.maybeRegister();
AccordService.startup(self);
if (!AccordService.isSetupOrStarting())
AccordService.localStartup(self);
AccordService.distributedStartup();
RegistrationStatus.instance.onRegistration();
Startup.maybeExecuteStartupTransformation(self);
@ -3165,7 +3168,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
true, // repairData
false, // repairPaxos
true, // dontPurgeTombstones
false // repairAccord
false, // repairAccord
false // permit no quorum
);
return new RepairCoordinator(this, cmd, options, keyspace);
@ -3846,8 +3850,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
transientMode = Optional.of(Mode.DRAINING);
}
if (AccordService.isSetup())
AccordService.instance().markShuttingDown();
if (AccordService.isSetupOrStarting())
AccordService.unsafeInstance().markShuttingDown();
// In-progress writes originating here could generate hints to be written,
// which is currently scheduled on the mutation stage. So shut down MessagingService
@ -3863,14 +3867,14 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
logger.error("Messaging service timed out shutting down", t);
}
if (AccordService.isSetup())
if (AccordService.isSetupOrStarting())
{
logger.info("Flushing Accord caches");
if (!AccordService.instance().flushCaches().awaitUninterruptibly(1, MINUTES))
if (!AccordService.unsafeInstance().flushCaches().awaitUninterruptibly(1, MINUTES))
logger.error("Could not flush Accord caches promptly");
if (AccordColumnFamilyStores.commandsForKey != null)
AccordColumnFamilyStores.commandsForKey.forceBlockingFlush(INTERNALLY_FORCED);
AccordService.instance().shutdownAndWait(1, MINUTES);
AccordService.unsafeInstance().shutdownAndWait(1, MINUTES);
}
// ScheduledExecutors shuts down after MessagingService, as MessagingService may issue tasks to it.

View File

@ -28,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
@ -60,6 +61,7 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.metrics.LogLinearHistogram;
import org.apache.cassandra.metrics.ShardedHitRate;
import org.apache.cassandra.service.accord.AccordCache.Adapter.Shrink;
import org.apache.cassandra.service.accord.AccordCacheEntry.LoadExecutor;
import org.apache.cassandra.service.accord.AccordCacheEntry.Status;
import org.apache.cassandra.service.accord.events.CacheEvents;
@ -100,11 +102,14 @@ public class AccordCache implements CacheSize
public interface Adapter<K, V, S>
{
enum Shrink { EVICT, DONE, PERFORM_WITHOUT_LOCK }
@Nullable V load(AccordCommandStore commandStore, K key);
@Nullable Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk);
default boolean canSave(@Nullable V value, @Nullable Object shrunk) { return true; }
// a result of null means we can immediately evict, without saving
@Nullable V quickShrink(V value);
Shrink decideFullShrink(K key, V value);
// a result of null means we cannot shrink, and should save/evict as appropriate
@Nullable Object fullShrink(K key, V value);
@Nullable V inflate(AccordCommandStore commandStore, K key, Object shrunk);
@ -149,6 +154,7 @@ public class AccordCache implements CacheSize
private long bytesCached;
private int noEvictGeneration;
private boolean shrinkingOn = true;
private boolean tryShrinkOrEvict;
public AccordCache(AccordCacheEntry.SaveExecutor saveExecutor, long maxSizeInBytes)
{
@ -156,11 +162,12 @@ public class AccordCache implements CacheSize
this.maxSizeInBytes = maxSizeInBytes;
}
// note: only affects current contents after lock is released
@Override
public void setCapacity(long sizeInBytes)
{
maxSizeInBytes = sizeInBytes;
maybeShrinkOrEvictSomeNodes();
tryShrinkOrEvict = true;
}
public void setShrinkingOn(boolean shrinkingOn)
@ -206,31 +213,56 @@ public class AccordCache implements CacheSize
* Roughly respects LRU semantics when evicting. Might consider prioritising keeping MODIFIED nodes around
* for longer to maximise the chances of hitting system tables fewer times (or not at all).
*/
private void maybeShrinkOrEvictSomeNodes()
void tryShrinkOrEvict(Lock lock)
{
if (!tryShrinkOrEvict)
return;
while (bytesCached > maxSizeInBytes && !evictQueue.isEmpty())
{
AccordCacheEntry<?, ?> node = evictQueue.peek();
shrinkOrEvict(node);
shrinkOrEvict(lock, node);
}
tryShrinkOrEvict = false;
}
@VisibleForTesting
private <K, V> void shrinkOrEvict(AccordCacheEntry<K, V> node)
private <K, V> void shrinkOrEvict(Lock lock, AccordCacheEntry<K, V> node)
{
require(node.references() == 0);
// TODO (desired): special-case evict queue with 1 element as no point shrinking first
if (shrinkingOn && node.tryShrink())
Shrink shrink = shrinkingOn ? node.tryShrink() : Shrink.EVICT;
if (shrink == Shrink.EVICT)
{
tryEvict(node);
}
else
{
IntrusiveLinkedList<AccordCacheEntry<?,?>> queue;
queue = node.isNoEvict() ? noEvictQueue : evictQueue;
node.unlink();
if (shrink == Shrink.DONE)
{
queue.addLast(node);
}
else
{
tryEvict(node);
K key = node.key();
V cur = node.tryGetExclusive();
Object upd = null;
lock.unlock();
try
{
upd = node.owner.parent().adapter().fullShrink(key, cur);
}
finally
{
//noinspection LockAcquiredButNotSafelyReleased
lock.lock();
node.tryApplyShrink(cur, upd);
queue.addLast(node);
}
}
}
}
@ -463,7 +495,7 @@ public class AccordCache implements CacheSize
node.initSize(parent());
++size;
node.notifyListeners(Listener::onAdd);
maybeShrinkOrEvictSomeNodes();
tryShrinkOrEvict = true;
return node;
}
@ -557,7 +589,7 @@ public class AccordCache implements CacheSize
}
}
maybeShrinkOrEvictSomeNodes();
tryShrinkOrEvict = true;
}
AccordCacheEntry<K, ?> remove(K key)
@ -1009,6 +1041,12 @@ public class AccordCache implements CacheSize
return quickShrink.apply(value);
}
@Override
public Shrink decideFullShrink(K key, V value)
{
return Shrink.DONE;
}
@Override
public Object fullShrink(K key, V value)
{
@ -1087,6 +1125,7 @@ public class AccordCache implements CacheSize
@Override public V load(AccordCommandStore commandStore, K key) { return null; }
@Override public Runnable save(AccordCommandStore commandStore, K key, @Nullable V value, @Nullable Object shrunk) { return null; }
@Override public V quickShrink(V value) { return null; }
@Override public Shrink decideFullShrink(K key, V value) { return Shrink.DONE; }
@Override public Object fullShrink(K key, V value) { return null; }
@Override public V inflate(AccordCommandStore commandStore, K key, Object shrunk) { return null; }
@Override public long estimateHeapSize(V value) { return 0; }
@ -1129,6 +1168,18 @@ public class AccordCache implements CacheSize
return value;
}
@Override
public Shrink decideFullShrink(RoutingKey key, CommandsForKey value)
{
if (value.isEmpty() || value.isLoadingPruned())
return Shrink.EVICT;
if (value.size() < 64)
return Shrink.DONE;
return Shrink.PERFORM_WITHOUT_LOCK;
}
@Override
public Object fullShrink(RoutingKey key, CommandsForKey value)
{
@ -1249,6 +1300,19 @@ public class AccordCache implements CacheSize
}
}
@Override
public Shrink decideFullShrink(TxnId txnId, Command value)
{
if (txnId.is(Txn.Kind.EphemeralRead))
Invariants.expect(value.saveStatus().compareTo(SaveStatus.ReadyToExecute) < 0);
// TODO (expected): improve heuristics and consider transaction size
if (value.partialDeps() == null || value.partialDeps().txnIds().size() < 64)
return Shrink.DONE;
return Shrink.PERFORM_WITHOUT_LOCK;
}
@Override
public @Nullable Command inflate(AccordCommandStore commandStore, TxnId key, Object serialized)
{

View File

@ -34,6 +34,7 @@ import accord.utils.IntrusiveLinkedListNode;
import accord.utils.Invariants;
import accord.utils.async.Cancellable;
import org.apache.cassandra.service.accord.AccordCache.Adapter;
import org.apache.cassandra.service.accord.AccordCache.Adapter.Shrink;
import org.apache.cassandra.utils.ObjectSizes;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED;
@ -462,16 +463,26 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
state = null;
}
boolean tryShrink()
Shrink tryShrink()
{
if (!isLoaded())
return false;
return Shrink.EVICT;
AccordCache.Type<K, V, ?> parent = owner.parent();
if (!tryShrink(key, parent.adapter()))
return false;
updateSize(parent);
return true;
Adapter<K, V, ?> adapter = parent.adapter();
if (isShrunk() || state == null)
return Shrink.EVICT;
V cur = (V)unwrap();
Shrink shrink = adapter.decideFullShrink(key, cur);
if (shrink == Shrink.PERFORM_WITHOUT_LOCK)
return Shrink.PERFORM_WITHOUT_LOCK;
Object upd = adapter.fullShrink(key, cur);
if (upd == null || upd == cur)
return Shrink.EVICT;
applyShrink(parent, cur, upd);
return Shrink.DONE;
}
V tryGetFull()
@ -584,20 +595,18 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
return ((FailedToSave)state).cause;
}
private boolean tryShrink(K key, Adapter<K, V, ?> adapter)
void tryApplyShrink(Object cur, Object upd)
{
if (isShrunk() || state == null)
return false;
Object cur = unwrap();
Object upd = adapter.fullShrink(key, (V)cur);
if (upd == null || upd == cur)
return false;
if (isLoaded() && unwrap() == cur && upd != cur && upd != null)
applyShrink(owner.parent(), cur, upd);
}
private void applyShrink(AccordCache.Type<K, V, ?> parent, Object cur, Object upd)
{
if (isNested()) ((Nested)this.state).state = upd;
else this.state = upd;
status |= SHRUNK;
return true;
updateSize(parent);
}
private void inflate(AccordCommandStore commandStore, K key, Adapter<K, V, ?> adapter)

View File

@ -142,6 +142,7 @@ public class AccordCommandStore extends CommandStore
@Override
public void close()
{
global().tryShrinkOrEvict(lock);
lock.unlock();
}
}
@ -208,10 +209,14 @@ public class AccordCommandStore extends CommandStore
ranges = update.newRangesForEpoch;
Invariants.require(ranges != null, "CommandStore %d created with no ranges", id);
}
tableId = (TableId)ranges.all().stream().map(r -> r.start().prefix()).reduce((a, b) -> {
Invariants.require(a.equals(b), "CommandStore created with multiple distinct TableId (%s and %s)", a, b);
return a;
}).orElseThrow(() -> Invariants.illegalState("CommandStore %d created with no ranges", id));
if (AccordService.isStarted())
progressLog.unsafeStart();
}
static Factory factory(IntFunction<AccordExecutor> executorFactory)

View File

@ -32,6 +32,7 @@ import accord.local.SequentialAsyncExecutor;
import accord.local.ShardDistributor;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.AccordSpec.QueueShardModel;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory;
@ -66,6 +67,14 @@ public class AccordCommandStores extends CommandStores implements CacheSize
maxQueuedRangeLoads = DatabaseDescriptor.getAccordMaxQueuedRangeLoadCount();
shrinkingOn = DatabaseDescriptor.getAccordCacheShrinkingOn();
refreshCapacities();
ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> {
for (AccordExecutor executor : executors)
{
executor.executeDirectlyWithLock(() -> {
executor.cacheExclusive().processNoEvictQueue();
});
}
}, 1L, 1L, TimeUnit.SECONDS);
}
static Factory factory()
@ -204,6 +213,9 @@ public class AccordCommandStores extends CommandStores implements CacheSize
for (AccordExecutor executor : executors)
{
executor.shutdown();
}
for (AccordExecutor executor : executors)
{
try
{
executor.awaitTermination(1, TimeUnit.MINUTES);

View File

@ -96,7 +96,16 @@ public class AccordDataStore implements DataStore
default: throw new UnhandledEnum(kind);
case Image:
{
AccordFetchCoordinator coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
AccordFetchCoordinator coordinator;
try
{
coordinator = new AccordFetchCoordinator(node, ranges, syncPoint, callback, safeStore.commandStore());
}
catch (Throwable t)
{
return new FetchResult.Failure(t);
}
coordinator.start();
return coordinator.result();
}

View File

@ -24,7 +24,6 @@ import java.util.List;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
@ -66,7 +65,6 @@ import accord.utils.async.Cancellable;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.DebuggableTask;
import org.apache.cassandra.concurrent.DebuggableTask.DebuggableTaskRunner;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.metrics.AccordCacheMetrics;
import org.apache.cassandra.metrics.AccordExecutorMetrics;
@ -130,6 +128,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
public void close()
{
executor.beforeUnlock();
global.tryShrinkOrEvict(lock);
lock.unlock();
}
}
@ -235,9 +234,6 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
this.elapsedRunning = AccordExecutorMetrics.INSTANCE.elapsedRunning.forShard(histogramsShard);
this.keys = AccordExecutorMetrics.INSTANCE.keys.forShard(histogramsShard);
this.replicaMetrics = new AccordReplicaMetrics.Shard(histogramsShard);
ScheduledExecutors.scheduledFastTasks.scheduleAtFixedRate(() -> {
executeDirectlyWithLock(cache::processNoEvictQueue);
}, 1L, 1L, TimeUnit.SECONDS);
}
public int executorId()
@ -493,7 +489,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
catch (Throwable t)
{
agent.onUncaughtException(t);
agent.onException(t);
}
}
@ -663,6 +659,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
if (waitingForCompletion != null && waitingForCompletion.peek().maybeNotify - position >= 0)
maybeNotifyWaitingForCompletion();
cache.tryShrinkOrEvict(lock);
}
}
@ -745,7 +743,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return;
try { task.failExclusive(fail); }
catch (Throwable t) { agent.onUncaughtException(t); }
catch (Throwable t) { agent.onException(t); }
finally
{
task.unqueueIfQueued();
@ -1197,7 +1195,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return false;
try { run.run(); }
catch (Throwable t) { agent.onUncaughtException(t); }
catch (Throwable t) { agent.onException(t); }
finally
{
if (owner == null)
@ -1322,7 +1320,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
queue.remove(this);
completeTaskExclusive(this);
try { fail(new CancellationException()); }
catch (Throwable t) { agent.onUncaughtException(t); }
catch (Throwable t) { agent.onException(t); }
}
}
@ -1373,7 +1371,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
{
if (result != null)
result.tryFailure(t);
agent.onUncaughtException(t);
agent.onException(t);
}
@Override
@ -1606,7 +1604,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
}
catch (Throwable t)
{
agent.onUncaughtException(t);
agent.onException(t);
return;
}
}
@ -1621,7 +1619,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
catch (Throwable t)
{
fail.addSuppressed(t);
agent.onUncaughtException(fail);
agent.onException(fail);
}
}
}

View File

@ -204,7 +204,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
pauseExclusive();
exitLockExclusive();
try { agent.onUncaughtException(t); }
try { agent.onException(t); }
catch (Throwable t2) { }
}
finally
@ -271,13 +271,13 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
catch (Throwable t2) { t.addSuppressed(t2); }
try { completeTaskExclusive(task); }
catch (Throwable t2) { t.addSuppressed(t2); }
try { agent.onUncaughtException(t); }
try { agent.onException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }
task = null;
}
else
{
try { agent.onUncaughtException(t); }
try { agent.onException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }
}
if (isHeldByExecutor)
@ -302,7 +302,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
try
{
t2.addSuppressed(t);
agent.onUncaughtException(t2);
agent.onException(t2);
}
catch (Throwable t3)
{

View File

@ -50,6 +50,7 @@ import accord.primitives.SyncPoint;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.TopologyException;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
@ -78,6 +79,7 @@ import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.TimeUUID;
import static accord.primitives.Routables.Slice.Minimal;
import static org.apache.cassandra.utils.CollectionSerializers.deserializeMap;
import static org.apache.cassandra.utils.CollectionSerializers.serializeMap;
import static org.apache.cassandra.utils.CollectionSerializers.serializedMapSize;
@ -219,7 +221,6 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
Invariants.nonNull(future);
Invariants.require(this.future == null, "future was not null: %s", this.future);
this.future = future;
logger.info("StreamFuture for plan {} received for bootstrap of {} from {}", planId, range, from);
maybeListen();
}
@ -229,6 +230,8 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
if (range == null || future == null)
return;
logger.info("StreamFuture for plan {} received for bootstrap of {} from {}", planId, range, from);
Invariants.nonNull(from);
future.addCallback((state, fail) -> {
if (fail == null)
@ -328,10 +331,10 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
}
@Override
public Read slice(Ranges ranges) { return new StreamingRead(to, this.ranges.slice(ranges)); }
public Read slice(Ranges ranges) { return new StreamingRead(to, this.ranges.slice(ranges, Minimal)); }
@Override
public Read intersecting(Participants<?> participants) { return new StreamingRead(to, this.ranges.slice(ranges)); }
public Read intersecting(Participants<?> participants) { return new StreamingRead(to, this.ranges.slice(ranges, Minimal)); }
@Override
public Read merge(Read other) { throw new UnsupportedOperationException(); }
@ -391,7 +394,7 @@ public class AccordFetchCoordinator extends AbstractFetchCoordinator implements
private final Map<TimeUUID, IncomingStream> streams = new HashMap<>();
public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore)
public AccordFetchCoordinator(Node node, Ranges ranges, SyncPoint syncPoint, DataStore.FetchRanges fetchRanges, CommandStore commandStore) throws TopologyException
{
super(node, node.someSequentialExecutor(), ranges, syncPoint, fetchRanges, commandStore);
}

View File

@ -798,6 +798,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public static @Nullable ByteBuffer asSerializedChange(Command before, Command after, Version userVersion) throws IOException
{
// TODO (expected): reusable buffer to build, or pre-size
try (DataOutputBuffer out = new DataOutputBuffer())
{
Writer writer = Writer.make(before, after);
@ -1149,7 +1150,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion);
break;
case PARTIAL_DEPS:
// TODO (required): this optimisation will be easily disabled;
// TODO (expected): this optimisation will be easily disabled;
// should either operate natively on ByteBuffer
// or else use some explicit API for copying bytes while skipping
if (deserializeDeps || !(in instanceof DataInputBuffer))

View File

@ -36,7 +36,6 @@ import org.apache.cassandra.service.accord.serializers.Version;
import static accord.local.CommandStores.RangesForEpoch;
// TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger
public class AccordJournalValueSerializers
{
public interface FlyweightImage

View File

@ -25,8 +25,6 @@ import java.util.Set;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.AsyncExecutor;
import accord.api.MessageSink;
@ -48,7 +46,6 @@ import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.TimeoutStrategy;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.utils.Clock;
import static accord.messages.MessageType.StandardMessage.*;
@ -59,8 +56,6 @@ import static org.apache.cassandra.service.accord.api.AccordWaitStrategies.slowR
public class AccordMessageSink implements MessageSink
{
private static final Logger logger = LoggerFactory.getLogger(AccordMessageSink.class);
public enum AccordMessageType implements MessageType
{
INTEROP_READ_REQ(Verb.ACCORD_INTEROP_READ_REQ),
@ -150,22 +145,20 @@ public class AccordMessageSink implements MessageSink
}
}
private final AccordAgent agent;
private final MessageDelivery messaging;
private final AccordEndpointMapper endpointMapper;
private final RequestCallbacks callbacks;
public AccordMessageSink(AccordAgent agent, MessageDelivery messaging, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
public AccordMessageSink(MessageDelivery messaging, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
{
this.agent = agent;
this.messaging = messaging;
this.endpointMapper = endpointMapper;
this.callbacks = callbacks;
}
public AccordMessageSink(AccordAgent agent, AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
public AccordMessageSink(AccordEndpointMapper endpointMapper, RequestCallbacks callbacks)
{
this(agent, MessagingService.instance(), endpointMapper, callbacks);
this(MessagingService.instance(), endpointMapper, callbacks);
}
@Override
@ -177,7 +170,7 @@ public class AccordMessageSink implements MessageSink
InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(to, message);
if (endpoint == null)
return;
logger.trace("Sending {} {} to {}", verb, message.payload, endpoint);
messaging.send(message, endpoint);
}
@ -222,7 +215,6 @@ public class AccordMessageSink implements MessageSink
return null;
}
logger.trace("Sending {} {} to {}", verb, message.payload, endpoint);
Cancellable cancellable = callbacks.registerAt(message.id(), executor, callback, to, nowNanos, slowAtNanos, expiresAtNanos, NANOSECONDS);
messaging.send(message, endpoint);
return cancellable;
@ -240,7 +232,6 @@ public class AccordMessageSink implements MessageSink
if (endpoint == null)
return;
logger.trace("Replying {} {} to {}", message.verb(), message.payload, endpoint);
messaging.send(message, endpoint);
}
@ -252,7 +243,7 @@ public class AccordMessageSink implements MessageSink
InetAddressAndPort endpoint = endpointMapper.mappedEndpointOrNull(replyingTo, message);
if (endpoint == null)
return;
logger.trace("Replying with failure {} {} to {}", message.verb(), message.payload, endpoint);
messaging.send(message, endpoint);
}

View File

@ -54,12 +54,6 @@ class AccordResponseVerbHandler<T extends Reply> implements IVerbHandler<T>
@Override
public void doVerb(Message message)
{
if (!AccordService.instance().shouldAcceptMessages())
{
dropping.debug(message.verb(), message.from());
return;
}
Node.Id from = endpointMapper.mappedIdOrNull(message.from(), message);
if (from == null)
{

View File

@ -32,10 +32,10 @@ import accord.coordinate.CoordinationFailed;
import accord.coordinate.Exhausted;
import accord.coordinate.Preempted;
import accord.coordinate.Timeout;
import accord.coordinate.TopologyMismatch;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.TxnId;
import accord.topology.TopologyMismatch;
import accord.utils.UnhandledEnum;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.ReadFailureException;
@ -152,7 +152,19 @@ public class AccordResult<V> extends AsyncFuture<V> implements BiConsumer<V, Thr
{
report = bookkeeping.newExhausted(txnId, keysOrRanges);
}
else if (isTxnRequest && coordinationFailed instanceof TopologyMismatch)
else if (fail instanceof TimeoutException)
{
report = bookkeeping.newTimeout(txnId, keysOrRanges);
}
else
{
report = bookkeeping.newFailed(txnId, keysOrRanges);
}
// this case happens when a non-timeout exception is seen, and we are unable to move forward
if (txnId != null && txnId.isSyncPoint())
AccordAgent.onFailedBarrier(txnId, fail);
}
else if (isTxnRequest && fail instanceof TopologyMismatch)
{
// Excluding bugs topology mismatch can occur because a table was dropped in between creating the txn
// and executing it.
@ -172,18 +184,6 @@ public class AccordResult<V> extends AsyncFuture<V> implements BiConsumer<V, Thr
trySuccess((V) RetryWithNewProtocolResult.instance);
return false;
}
else if (fail instanceof TimeoutException)
{
report = bookkeeping.newTimeout(txnId, keysOrRanges);
}
else
{
report = bookkeeping.newFailed(txnId, keysOrRanges);
}
// this case happens when a non-timeout exception is seen, and we are unable to move forward
if (txnId != null && txnId.isSyncPoint())
AccordAgent.onFailedBarrier(txnId, fail);
}
else if (fail instanceof RequestTimeoutException || fail instanceof TimeoutException)
{
report = bookkeeping.newTimeout(txnId, keysOrRanges);

View File

@ -70,7 +70,6 @@ import accord.local.durability.DurabilityService;
import accord.local.durability.ShardDurability;
import accord.messages.Reply;
import accord.messages.Request;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Seekable;
@ -149,6 +148,7 @@ import static org.apache.cassandra.db.SystemKeyspace.BootstrapState.COMPLETED;
import static org.apache.cassandra.journal.Params.ReplayMode.RESET;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordReadBookkeeping;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.accordWriteBookkeeping;
import static org.apache.cassandra.service.accord.AccordTopology.tcmIdToAccord;
import static org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter.getTableMetadata;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -233,7 +233,7 @@ public class AccordService implements IAccordService, Shutdownable
private final AccordResponseVerbHandler<? extends Reply> responseHandler;
@GuardedBy("this")
private State state = State.INIT;
private volatile State state = State.INIT;
private static final IAccordService NOOP_SERVICE = new NoOpAccordService();
@ -254,11 +254,26 @@ public class AccordService implements IAccordService, Shutdownable
unsafeInstance = instance = NOOP_SERVICE;
}
public static IAccordService tryGetUnsafe()
{
return unsafeInstance;
}
public static boolean isSetup()
{
return instance != null;
}
public static boolean isSetupOrStarting()
{
return unsafeInstance != null;
}
public static boolean isStarted()
{
return isSetup() && unsafeInstance instanceof AccordService && ((AccordService) unsafeInstance).state == State.STARTED;
}
public static IVerbHandler<Void> watermarkHandlerOrNoop()
{
if (!isSetup()) return ignore -> {};
@ -268,38 +283,47 @@ public class AccordService implements IAccordService, Shutdownable
public static IVerbHandler<? extends Request> requestHandlerOrNoop()
{
if (!isSetup()) return ignore -> {};
return instance().requestHandler();
if (instance == null) return ignore -> {};
return instance.requestHandler();
}
public static IVerbHandler<? extends Reply> responseHandlerOrNoop()
{
if (!isSetup()) return ignore -> {};
return instance().responseHandler();
if (unsafeInstance == null) return ignore -> {};
return unsafeInstance.responseHandler();
}
@VisibleForTesting
public synchronized static AccordService startup(NodeId tcmId)
public synchronized static void localStartup(NodeId tcmId)
{
Invariants.require(instance == null);
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
{
unsafeInstance = instance = NOOP_SERVICE;
return null;
}
else
{
AccordService as = new AccordService(tcmIdToAccord(tcmId));
unsafeInstance = as;
as.localStartup();
}
}
if (instance != null)
return (AccordService) instance;
public synchronized static AccordService distributedStartup()
{
if (unsafeInstance == NOOP_SERVICE)
return null;
AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmId));
unsafeInstance = as;
as.startup();
AccordService as = (AccordService) unsafeInstance;
if (as.state != State.INIT)
return as;
as.distributedStartupInternal();
instance = as;
AccordReplicaMetrics.touch();
AccordSystemMetrics.touch();
AccordViolationHandler.setup();
WatermarkCollector.fetchAndReportWatermarksAsync(as.topology());
return as;
}
@ -323,12 +347,6 @@ public class AccordService implements IAccordService, Shutdownable
return true;
}
@Override
public boolean shouldAcceptMessages()
{
return state == State.STARTED && journal.started();
}
public static IAccordService instance()
{
IAccordService i = instance;
@ -365,7 +383,7 @@ public class AccordService implements IAccordService, Shutdownable
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.messageSink = new AccordMessageSink(endpointMapper, callbacks);
this.node = new Node(localId,
messageSink,
topologyService,
@ -391,7 +409,7 @@ public class AccordService implements IAccordService, Shutdownable
}
@Override
public synchronized void startup()
public synchronized void localStartup()
{
if (state != State.INIT)
return;
@ -428,8 +446,13 @@ public class AccordService implements IAccordService, Shutdownable
{
node.unsafeSetReplaying(false);
}
}
private void distributedStartupInternal()
{
finishTopologyInitialization();
WatermarkCollector.fetchAndReportWatermarksAsync(topology());
finishInitialization();
catchup();
fastPathCoordinator.start();
@ -439,9 +462,12 @@ public class AccordService implements IAccordService, Shutdownable
Ints.checkedCast(getAccordShardDurabilityMaxSplits()),
Ints.checkedCast(getAccordShardDurabilityCycle(SECONDS)), SECONDS);
node.durability().global().setGlobalCycleTime(Ints.checkedCast(getAccordGlobalDurabilityCycle(SECONDS)), SECONDS);
// Only enable durability scheduling _after_ we have fully replayed journal
// Only enable durability scheduling and progress logs _after_ we have fully replayed journal
node.durability().start();
// we set ourselves to STARTED before starting progress logs as this is the condition we use to decide if we
// start the progress log on command store initialisation (so creates a synchronisation point)
state = State.STARTED;
node.commandStores().forAll("", safeStore -> safeStore.progressLog().start());
}
void catchup()
@ -477,13 +503,18 @@ public class AccordService implements IAccordService, Shutdownable
Throwable failed = f.cause();
if (failed != null)
{
if (spec.catchup_on_start_exit_on_failure)
throw new RuntimeException("Could not catch up with peers", failed);
logger.error("Could not catch up with peers; continuing to startup");
break;
}
long end = nanoTime();
double seconds = NANOSECONDS.toMillis(end - start)/1000.0;
logger.info("Finished catching up with all quorums. {}s elapsed.", String.format("%.2f", seconds));
// TODO (expected): make configurable
if (seconds <= spec.catchup_on_start_success_latency.toSeconds())
break;
@ -511,9 +542,9 @@ public class AccordService implements IAccordService, Shutdownable
/**
* Startup is broken up in two phases: local and distributed startup. During local startup, we replay up to
* the latest epoch known to the node prior to restart. After that, we replay journal itself, and only after
* that we finish initializaiton and replay the rest of epochs.
* that we finish initialization and replay the rest of epochs.
*/
void finishInitialization()
void finishTopologyInitialization()
{
endpointMapper.updateMapping(ClusterMetadata.current());
TopologyManager topology = node.topology();
@ -686,10 +717,9 @@ public class AccordService implements IAccordService, Shutdownable
private AsyncChain<Void> syncInternal(Timestamp minBound, Keys keys, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote)
{
TxnId txnId = node.nextTxnId(minBound, keys, Write);
FullRoute<?> route = node.computeRoute(txnId, keys);
return node.withEpochAtLeast(txnId.epoch(), null, () -> {
Txn txn = new Txn.InMemory(Write, keys, TxnRead.createNoOpRead(keys), TxnQuery.UNSAFE_EMPTY, TxnUpdate.empty(), new TableMetadatasAndKeys(TableMetadatas.none(), keys));
return CoordinateTransaction.coordinate(node, route, txnId, txn)
return CoordinateTransaction.coordinate(node, txnId, txn)
.mapToNull();
});
}
@ -782,7 +812,7 @@ public class AccordService implements IAccordService, Shutdownable
return keys;
TableId tableId = tableId(keys, r -> ((AccordRoutableKey)r).table());
return sliceToAccord(tableId, keys, Keys::slice);
return sliceToAccord(tableId, keys, Keys::overlapping);
}
public static Ranges intersecting(Ranges ranges)
@ -791,7 +821,7 @@ public class AccordService implements IAccordService, Shutdownable
return ranges;
TableId tableId = tableId(ranges, r -> ((TokenRange)r).table());
return sliceToAccord(tableId, ranges, Ranges::slice);
return sliceToAccord(tableId, ranges, Ranges::overlapping);
}
private static <C extends Seekables<?, ?>> C sliceToAccord(TableId tableId, C collection, BiFunction<C, Ranges, C> slice)
@ -974,7 +1004,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public void shutdownAndWait(long timeout, TimeUnit unit)
{
if (!ExecutorUtils.shutdownSequentiallyAndWait(shutdownableSubsystems(), timeout, unit))
if (!ExecutorUtils.shutdownThenWait(shutdownableSubsystems(), timeout, unit))
logger.error("One or more subsystems did not shut down cleanly.");
}

View File

@ -728,12 +728,12 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
public void fail(Throwable throwable)
{
commandStore.agent().onUncaughtException(throwable);
commandStore.agent().onException(throwable);
if (state.isComplete())
return;
if (commandStore.hasSafeStore())
commandStore.agent().onUncaughtException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
commandStore.agent().onException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
state(FAILING);
if (callback != null)
@ -747,12 +747,12 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
{
if (newFailure)
{
commandStore.agent().onUncaughtException(throwable);
commandStore.agent().onException(throwable);
if (state.isComplete())
return;
if (commandStore.hasSafeStore())
commandStore.agent().onUncaughtException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
commandStore.agent().onException(new IllegalStateException(String.format("Failure to cleanup safe store for %s; status=%s", this, state), throwable));
}
state(FAILED);
@ -860,7 +860,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
catch (Throwable t)
{
releaseResourcesSlow(caches, t);
commandStore.agent().onUncaughtException(t);
commandStore.agent().onException(t);
}
}

View File

@ -760,6 +760,9 @@ public class AccordTracing extends AccordCoordinatorMetrics.Listener
public Tracing trace(TxnId txnId, @Nullable Participants<?> participants, CoordinationKind kind)
{
if (kind == CoordinationKind.FetchDurableBefore)
return (cs, msg) -> logger.info("Catchup/FetchDurableBefore: {}", msg);
if (!txnIdMap.containsKey(txnId) && null == maybeTrace(txnId, participants, kind, NewOrFailure.NEW, traceNewPatterns))
return null;

View File

@ -46,12 +46,6 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
@Override
public void doVerb(Message<T> message) throws IOException
{
if (!AccordService.instance().shouldAcceptMessages())
{
dropping.debug(message.verb(), message.from());
return;
}
logger.trace("Receiving {} from {}", message.payload, message.from());
T request = message.payload;

View File

@ -108,6 +108,7 @@ public interface IAccordService
}
boolean isEnabled();
long currentEpoch();
void setCacheSize(long kb);
@ -115,7 +116,7 @@ public interface IAccordService
TopologyManager topology();
void startup();
void localStartup();
Future<Void> flushCaches();
void markShuttingDown();
@ -186,8 +187,6 @@ public interface IAccordService
Params journalConfiguration();
boolean shouldAcceptMessages();
Node node();
/**
@ -274,7 +273,7 @@ public interface IAccordService
}
@Override
public void startup()
public void localStartup()
{
try
{
@ -369,12 +368,6 @@ public interface IAccordService
throw new UnsupportedOperationException("Cannot return configuration when accord.enabled = false in cassandra.yaml");
}
@Override
public boolean shouldAcceptMessages()
{
return true;
}
@Override
public Node node()
{
@ -490,9 +483,9 @@ public interface IAccordService
}
@Override
public void startup()
public void localStartup()
{
delegate.startup();
delegate.localStartup();
}
@Override
@ -573,12 +566,6 @@ public interface IAccordService
return delegate.journalConfiguration();
}
@Override
public boolean shouldAcceptMessages()
{
return delegate.shouldAcceptMessages();
}
@Override
public Node node()
{

View File

@ -36,7 +36,7 @@ public class ImmediateAsyncExecutor implements AbstractAsyncExecutor
}
catch (Throwable t)
{
AccordAgent.handleUncaughtException(t);
AccordAgent.handleException(t);
}
}
}

View File

@ -139,7 +139,6 @@ public class WatermarkCollector implements TopologyListener
return;
Snapshot snapshot = m.payload;
long minEpoch = topologyManager.minEpoch();
forEachEpoch(topologyManager::onEpochClosed, snapshot.closed);
forEachEpoch(topologyManager::onEpochRetired, snapshot.retired);
for (Map.Entry<Long, Long> e : snapshot.synced.entrySet())

View File

@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord.api;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import javax.annotation.Nullable;
@ -36,6 +37,7 @@ import accord.api.ProgressLog.BlockedUntil;
import accord.api.RoutingKey;
import accord.api.Tracing;
import accord.coordinate.Coordination;
import accord.coordinate.Timeout;
import accord.local.Command;
import accord.local.Node;
import accord.local.SafeCommand;
@ -193,24 +195,23 @@ public class AccordAgent implements Agent, OwnershipEventListener
logger.error("This replica has become stale for {} as of {}", ranges, staleSince);
}
@Override
public void onUncaughtException(Throwable t)
public static void handleException(Throwable t)
{
handleUncaughtException(t);
}
public static void handleUncaughtException(Throwable t)
{
if (t instanceof RequestTimeoutException || t instanceof CancellationException)
if (t instanceof RequestTimeoutException || t instanceof CancellationException || t instanceof TimeoutException || t instanceof Timeout)
return;
JVMStabilityInspector.uncaughtException(Thread.currentThread(), t);
}
@Override
public void onCaughtException(Throwable t, String context)
public void onException(Throwable t)
{
logger.warn(context, t);
JVMStabilityInspector.inspectThrowable(t);
handleException(t);
}
@Override
public void onException(Throwable t, String context)
{
handleException(t);
}
@Override

View File

@ -106,8 +106,18 @@ public class AccordInteropAdapter extends TxnAdapter
if (updateKind != AccordUpdate.Kind.UNRECOVERABLE_REPAIR && (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ONE || txn.read().keys().isEmpty()))
return false;
new AccordInteropExecution(node, txnId, txn, updateKind, route, ballot, executeAt, deps, callback, executor, consistencyLevel, endpointMapper)
.start();
AccordInteropExecution execution;
try
{
execution = new AccordInteropExecution(node, txnId, txn, updateKind, route, ballot, executeAt, deps, callback, executor, consistencyLevel, endpointMapper);
}
catch (Throwable t)
{
callback.accept(null, t);
return true;
}
execution.start();
return true;
}
@ -119,9 +129,19 @@ public class AccordInteropAdapter extends TxnAdapter
if (consistencyLevel == null || consistencyLevel == ConsistencyLevel.ANY || writes.isEmpty())
return false;
AccordInteropPersist persist;
try
{
Topologies all = execution(node, any, sendTo, selectSendTo, fullRoute, txnId, executeAt);
new AccordInteropPersist(node, executor, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, CoordinationFlags.none(), informDurableOnDone, Minimal, callback)
.start();
persist = new AccordInteropPersist(node, executor, all, txnId, require, ballot, txn, executeAt, deps, writes, result, fullRoute, consistencyLevel, CoordinationFlags.none(), informDurableOnDone, Minimal, callback);
}
catch (Throwable t)
{
callback.accept(null, t);
return true;
}
persist.start();
return true;
}
}

View File

@ -178,7 +178,7 @@ public class AccordInteropApply extends Apply implements LocalListeners.ComplexL
else if (failure != null)
{
node.reply(replyTo, replyContext, null, failure);
node.agent().onUncaughtException(failure);
node.agent().onException(failure);
fail();
}

View File

@ -50,6 +50,7 @@ import accord.topology.ActiveEpochs;
import accord.topology.Shard;
import accord.topology.Topologies;
import accord.topology.Topology;
import accord.topology.TopologyException;
import accord.utils.Invariants;
import accord.utils.UnhandledEnum;
import accord.utils.async.AsyncChain;
@ -139,7 +140,7 @@ public class AccordInteropExecution implements ReadCoordinator
private volatile long uniqueHlc;
public AccordInteropExecution(Node node, TxnId txnId, Txn txn, AccordUpdate.Kind updateKind, FullRoute<?> route, Ballot ballot, Timestamp executeAt, Deps deps, BiConsumer<? super Result, Throwable> callback,
SequentialAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper)
SequentialAsyncExecutor executor, ConsistencyLevel consistencyLevel, AccordEndpointMapper endpointMapper) throws TopologyException
{
requireArgument(!txn.read().keys().isEmpty() || updateKind == AccordUpdate.Kind.UNRECOVERABLE_REPAIR);
this.node = node;

View File

@ -53,7 +53,6 @@ import org.apache.cassandra.utils.concurrent.Future;
import static accord.local.durability.DurabilityService.SyncLocal.NoLocal;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.local.durability.DurabilityService.SyncRemote.Quorum;
import static accord.primitives.Timestamp.mergeMax;
import static accord.primitives.Timestamp.minForEpoch;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ -79,15 +78,15 @@ public class AccordRepair
private volatile Throwable shouldAbort = null;
private volatile Thread waiting;
public AccordRepair(SharedContext ctx, ColumnFamilyStore cfs, TimeUUID repairId, String keyspace, Collection<Range<Token>> ranges, boolean requireAllEndpoints, List<InetAddressAndPort> endpoints)
public AccordRepair(SharedContext ctx, ColumnFamilyStore cfs, TimeUUID repairId, String keyspace, Collection<Range<Token>> ranges, SyncRemote syncRemote, List<InetAddressAndPort> endpoints)
{
this.ctx = ctx;
this.cfs = cfs;
this.repairId = repairId;
this.syncRemote = syncRemote;
// TODO (desired): support unsafe configuration where we permit less than a quorum,
// but this is challenging to do safely as repair has no concept of participants from earlier epochs
this.syncRemote = requireAllEndpoints ? All : Quorum;
if (endpoints != null)
if (syncRemote != All && endpoints != null)
{
including = new ArrayList<>(endpoints.size());
AccordEndpointMapper mapper = AccordService.instance().endpointMapper();

View File

@ -97,34 +97,38 @@ public abstract class AbstractSortedCollector<T, C> extends AbstractList<T>
return add;
}
public void clear()
{
if (count > 1)
cachedAny().forceDiscard((Object[])buffer, count);
buffer = null;
count = 0;
}
public C build()
{
C result;
if (count == 0)
{
return empty();
result = empty();
}
else if (count == 1)
{
return of((T)buffer);
result = of((T)buffer);
}
else if (count < BTREE_THRESHOLD)
{
C result = copy((Object[])buffer, count);
result = copy((Object[])buffer, count);
cachedAny().forceDiscard((Object[])buffer, count);
return result;
}
else
{
return copyBtree((Object[])buffer, count);
result = copyBtree((Object[])buffer, count);
}
buffer = null;
count = 0;
return result;
}
public void clear()
{
if (count > 1 && count < BTREE_THRESHOLD)
cachedAny().forceDiscard((Object[])buffer, count);
buffer = null;
count = 0;
}
@Override

View File

@ -168,7 +168,7 @@ public class CommandSerializers
if (executeAt.equals(Timestamp.NONE))
return size + TypeSizes.sizeofUnsignedVInt(0L);
size += TypeSizes.sizeofUnsignedVInt(executeAt.epoch() - txnId.epoch());
size += TypeSizes.sizeofUnsignedVInt(1 + executeAt.epoch() - txnId.epoch());
}
size += TypeSizes.sizeofUnsignedVInt(executeAt.hlc() - txnId.hlc());
size += TypeSizes.sizeofUnsignedVInt(executeAt.node.id);

View File

@ -67,6 +67,7 @@ import org.apache.cassandra.utils.CollectionSerializers;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.Pair;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.utils.ArrayBuffers.cachedInts;
import static accord.utils.Invariants.requireArgument;
import static accord.utils.SortedArrays.Search.CEIL;
@ -637,13 +638,13 @@ public class TxnUpdate extends AccordUpdate
@Override
public TxnUpdate slice(Ranges ranges)
{
return getTxnUpdate(keys -> keys.slice(ranges));
return getTxnUpdate(keys -> keys.slice(ranges, Minimal));
}
@Override
public TxnUpdate intersecting(Participants<?> participants)
{
return getTxnUpdate(keys -> keys.intersecting(participants));
return getTxnUpdate(keys -> keys.intersecting(participants, Minimal));
}
@VisibleForTesting

View File

@ -68,7 +68,6 @@ import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.service.accord.serializers.TableMetadatas;
import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.utils.BooleanSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.SimpleBitSetSerializers;
@ -545,7 +544,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
public void skip(Seekables keys, DataInputPlus in, Version version) throws IOException
{
TableMetadatas tables = TableMetadatas.deserializeSelf(in);
BooleanSerializer.serializer.deserialize(in);
SimpleBitSetSerializers.any.deserialize(in);
skipArray(new TableMetadatasAndKeys(tables, keys), in, version, Update.serializer);
}

View File

@ -380,7 +380,7 @@ public abstract class ConsensusTableMigration
boolean optimiseStreams = false;
boolean ignoreUnreplicatedKeyspaces = true;
boolean dontPurgeTombstones = false;
RepairOption repairOption = new RepairOption(RepairParallelism.PARALLEL, primaryRange, incremental, trace, numJobThreads, intersectingRanges, pullRepair, forceRepair, PreviewKind.NONE, optimiseStreams, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord);
RepairOption repairOption = new RepairOption(RepairParallelism.PARALLEL, primaryRange, incremental, trace, numJobThreads, intersectingRanges, pullRepair, forceRepair, PreviewKind.NONE, optimiseStreams, ignoreUnreplicatedKeyspaces, repairData, repairPaxos, dontPurgeTombstones, repairAccord, false);
tables.forEach(table -> repairOption.getColumnFamilies().add(table.tableName));
return repairOption;
}

View File

@ -117,6 +117,9 @@ public class Repair extends AbstractCommand
@Option(paramLabel = "accord-only", names = { "-accord-only", "--accord-only" }, description = "If the --accord-only flag is included, no table data is repaired, only accord operations..")
private boolean accordOnly = false;
@Option(paramLabel = "permit-no-quorum", names = { "-permit-no-quorum", "--permit-no-quorum" }, description = "If the --permit-no-quorum flag is included, accord repair is permitted to proceed without a quorum, including only the specified nodes")
private boolean permitNoQuorum = false;
@Option(paramLabel = "skip-accord", names = { "-skip-accord", "--skip-accord" }, description = "If the --skip-accord flag is included, the Accord repair step is skipped. Accord repair is also skipped for preview repairs.")
private boolean skipAccord = false;
@ -202,6 +205,7 @@ public class Repair extends AbstractCommand
checkArgument(!(skipAccord && accordOnly), "Can't specify both skip-accord and accord-only");
boolean repairAccord = !skipAccord && !paxosOnly && getPreviewKind() == PreviewKind.NONE;
options.put(RepairOption.REPAIR_ACCORD_KEY, Boolean.toString(repairAccord));
options.put(RepairOption.PERMIT_NO_QUORUM_KEY, Boolean.toString(permitNoQuorum));
boolean repairData = false;
if (getPreviewKind() == PreviewKind.NONE)
{

View File

@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
@ -103,13 +104,49 @@ public class ExecutorUtils
else
throw new IllegalArgumentException(executor.toString());
}
catch (Throwable t)
catch (InterruptedException t)
{
throw new UncheckedInterruptedException(t);
}
}
return shutdown;
}
public static boolean shutdownThenWait(Iterable<?> executors, long timeout, TimeUnit unit)
{
long deadline = nanoTime() + unit.toNanos(timeout);
for (Object executor : executors)
{
if (executor instanceof ExecutorService) ((ExecutorService) executor).shutdown();
else if (executor instanceof Shutdownable) ((Shutdownable) executor).shutdown();
else throw new IllegalArgumentException(executor.toString());
}
for (Object executor : executors)
{
try
{
if (executor instanceof ExecutorService)
{
if (!((ExecutorService) executor).awaitTermination(Math.max(0, deadline - nanoTime()), NANOSECONDS))
return false;
}
else if (executor instanceof Shutdownable)
{
if (!((Shutdownable) executor).awaitTermination(Math.max(0, deadline - nanoTime()), NANOSECONDS))
return false;
}
else throw new IllegalArgumentException(executor.toString());
}
catch (InterruptedException t)
{
throw new IllegalStateException("Caught interrupt while shutting down " + executor, t);
}
}
return shutdown;
return true;
}
public static void shutdown(ExecutorService ... executors)

View File

@ -45,6 +45,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
@ -197,7 +198,12 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
CassandraRelevantProperties.TEST_FLUSH_LOCAL_SCHEMA_CHANGES.reset();
CassandraRelevantProperties.NON_GRACEFUL_SHUTDOWN.reset();
CassandraRelevantProperties.IO_NETTY_TRANSPORT_NONATIVE.setBoolean(false);
withInstanceInitializer((classLoader, threadGroup, i, i1) -> {
withInstanceInitializer(defaultInitializer());
}
private IInstanceInitializer defaultInitializer()
{
return (classLoader, threadGroup, i, i1) -> {
try
{
Class<?> ef = classLoader.loadClass(ExecutorFactory.class.getName());
@ -215,10 +221,21 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
else
logger.info("Unable to set ExecutorFactory for instance {}", i, e);
}
catch (NoSuchMethodException | InvocationTargetException | InstantiationException | IllegalAccessException e)
catch (NoSuchMethodException | InvocationTargetException | InstantiationException |
IllegalAccessException e)
{
throw new RuntimeException(e);
}
};
}
@Override
public B withInstanceInitializer(BiConsumer<ClassLoader, Integer> instanceInitializer)
{
IInstanceInitializer wrap = defaultInitializer();
return withInstanceInitializer((classLoader, threadGroup, num, v) -> {
wrap.initialise(classLoader, threadGroup, num, v);
instanceInitializer.accept(classLoader, num);
});
}
@ -573,6 +590,10 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
protected AbstractCluster(AbstractBuilder<I, ? extends ICluster<I>, ?> builder)
{
// start the JNA cleaner on the system class loader to avoid pinning an instance
// (we do it here because startup() isn't always called)
com.sun.jna.internal.Cleaner.getCleaner();
this.root = builder.getRootPath();
this.sharedClassLoader = builder.getSharedClassLoader();
this.sharedClassPredicate = builder.getSharedClasses();
@ -1068,8 +1089,6 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
public void startup()
{
// start the JNA cleaner on the system class loader to avoid pinning an instance
com.sun.jna.internal.Cleaner.getCleaner();
try (AllMembersAliveMonitor monitor = new AllMembersAliveMonitor())
{
monitor.startPolling();

View File

@ -51,6 +51,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.apache.cassandra.Util;
import org.apache.cassandra.audit.AuditLogManager;
@ -64,6 +65,7 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.SharedExecutorPool;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
@ -822,6 +824,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
try
{
CommitLog.instance.recoverSegmentsOnDisk();
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
}
catch (IOException e)
{
@ -868,7 +873,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
ClusterMetadataService.instance().processor().fetchLogAndWait();
NodeId self = Register.maybeRegister();
RegistrationStatus.instance.onRegistration();
AccordService.startup(self);
if (!AccordService.isSetupOrStarting())
AccordService.localStartup(self);
AccordService.distributedStartup();
boolean joinRing = config.get(Constants.KEY_DTEST_JOIN_RING) == null || (boolean) config.get(Constants.KEY_DTEST_JOIN_RING);
if (ClusterMetadata.current().directory.peerState(self) != NodeState.JOINED && joinRing)
{
@ -1028,8 +1036,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
);
error = parallelRun(error, executor, () -> {
if (!AccordService.isSetup()) return;
AccordService.instance().shutdownAndWait(1l, MINUTES);
if (AccordService.isSetupOrStarting())
AccordService.unsafeInstance().shutdownAndWait(1L, MINUTES);
});
// CommitLog must shut down after Stage, or threads from the latter may attempt to use the former.
@ -1064,6 +1072,10 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
try
{
future.get();
ThreadGroup group = Thread.currentThread().getThreadGroup();
int active = group.activeCount();
Invariants.expect(group.getParent().activeCount() <= active
|| CassandraRelevantProperties.DTEST_IGNORE_SHUTDOWN_THREADCOUNT.getBoolean());
return null;
}
finally

View File

@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import org.junit.BeforeClass;
import org.junit.Test;
import net.bytebuddy.ByteBuddy;
@ -53,6 +54,12 @@ import static org.junit.Assert.fail;
public class DecommissionTest extends TestBaseImpl
{
@BeforeClass
public static void before()
{
CassandraRelevantProperties.DTEST_IGNORE_SHUTDOWN_THREADCOUNT.setBoolean(true);
}
@Test
public void testDecommission() throws Throwable
{

View File

@ -48,6 +48,7 @@ import org.slf4j.LoggerFactory;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
import accord.topology.TopologyException;
import org.apache.cassandra.config.Config.PaxosVariant;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ast.Symbol;
@ -709,7 +710,9 @@ 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().active().withUnsyncedEpochs(routables, epoch, epoch);
Topologies topology;
try { topology = AccordService.instance().topology().active().withUnsyncedEpochs(routables, epoch, epoch); }
catch (TopologyException e) { throw new RuntimeException(e); }
// 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);
});

View File

@ -38,6 +38,7 @@ import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.TopologyException;
import accord.utils.ImmutableBitSet;
import accord.utils.LargeBitSet;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -127,7 +128,9 @@ public class AccordCommandStoreTryExecuteListeningTest extends TestBaseImpl
PartitionKey key = pk(1, "ks", "tbl");
Txn txn = node.agent().emptySystemTxn(Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range);
TxnId txnId = node.nextTxnId(txn);
FullRoute<?> route = node.computeRoute(txnId, Ranges.of(key.asRange()));
FullRoute<?> route;
try { route = node.computeRoute(txnId, Ranges.of(key.asRange())); }
catch (TopologyException e) { throw new RuntimeException(e); }
AccordCommandStore commandStore = (AccordCommandStore) node.commandStores().unsafeForKey(key.toUnseekable());
int[] rangesToTxnIds = new int[dependencies.length + 1];
rangesToTxnIds[0] = rangesToTxnIds.length;

View File

@ -194,7 +194,6 @@ public class AccordMetricsTest extends AccordTestBase
assertClientMetrics(0, "AccordRead", 1, 0);
assertClientMetrics(1, "AccordRead", 0, 0);
// TODO (required): reenable or remove metric
assertCoordinatorMetrics(0, "ro", 0, 0, 1, 0, 0);
assertCoordinatorMetrics(1, "ro" , 0, 0, 0, 0, 1);
assertReplicaMetrics(0, "ro", 0, 0, 0);

View File

@ -110,7 +110,7 @@ public class AccordSimpleFastPathTest extends TestBaseImpl
long epoch = cm.epoch.getEpoch();
TopologyManager tm = AccordService.instance().topology();
Topology topology = tm.active().globalForEpoch(epoch);
Topology topology = tm.active().getKnown(epoch).global();
Assert.assertFalse(topology.shards().isEmpty());
topology.shards().forEach(shard -> Assert.assertEquals(idSet(1, 2, 3), shard.nodes.without(shard.notInFastPath)));
return cm.epoch.getEpoch();

View File

@ -43,6 +43,7 @@ import accord.primitives.SyncPoint;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Writes;
import accord.topology.TopologyException;
import accord.utils.ImmutableBitSet;
import accord.utils.LargeBitSet;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -114,7 +115,9 @@ public class AccordJournalReplayTest extends TestBaseImpl
Txn syncPointTxn = node.agent().emptySystemTxn(Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range);
TxnId syncPointId = node.nextTxnId(syncPointTxn);
TxnId txnId = node.nextTxnId(txn);
FullRoute<?> route = node.computeRoute(txnId, txn.keys());
FullRoute<?> route;
try { route = node.computeRoute(txnId, txn.keys()); }
catch (TopologyException e) { throw new RuntimeException(e); }
AccordCommandStore commandStore = (AccordCommandStore) node.commandStores().unsafeForKey(key.toUnseekable());
Deps deps = new Deps(KeyDeps.NONE, RangeDeps.SerializerSupport.create(new accord.primitives.Range[] { key.asRange() }, new TxnId[] { syncPointId }, new int[] { 2, 0 }, new int[] { 2, 0 }));
Command.WaitingOn waitingOn; {

View File

@ -16,7 +16,8 @@ SYNOPSIS
[(-iuk | --ignore-unreplicated-keyspaces)]
[(-j <job_threads> | --job-threads <job_threads>)]
[(-local | --in-local-dc)] [(-os | --optimise-streams)]
[(-paxos-only | --paxos-only)] [(-pl | --pull)]
[(-paxos-only | --paxos-only)]
[(-permit-no-quorum | --permit-no-quorum)] [(-pl | --pull)]
[(-pr | --partitioner-range)] [(-prv | --preview)]
[(-seq | --sequential)] [(-skip-accord | --skip-accord)]
[(-skip-paxos | --skip-paxos)]
@ -76,6 +77,11 @@ OPTIONS
If the --paxos-only flag is included, no table data is repaired,
only paxos operations..
-permit-no-quorum, --permit-no-quorum
If the --permit-no-quorum flag is included, accord repair is
permitted to proceed without a quorum, including only the specified
nodes
-pl, --pull
Use --pull to perform a one way repair where data is only streamed
from a remote node to this node.

View File

@ -129,7 +129,7 @@ public class OnInstanceRepair extends ClusterAction
{
Collection<Range<Token>> ranges = rangesSupplier.call();
// no need to wait for completion, as we track all task submissions and message exchanges, and ensure they finish before continuing to next action
StorageService.instance.repair(keyspaceName, new RepairOption(RepairParallelism.SEQUENTIAL, isPrimaryRangeOnly, repairType.incremental, false, 1, ranges, false, force, PreviewKind.NONE, false, true, repairType.repairData, repairType.repairPaxos, false, repairType.repairAccord), singletonList((tag, event) -> {
StorageService.instance.repair(keyspaceName, new RepairOption(RepairParallelism.SEQUENTIAL, isPrimaryRangeOnly, repairType.incremental, false, 1, ranges, false, force, PreviewKind.NONE, false, true, repairType.repairData, repairType.repairPaxos, false, repairType.repairAccord, false), singletonList((tag, event) -> {
if (event.getType() == ProgressEventType.COMPLETE)
listener.run();
}));

View File

@ -279,10 +279,14 @@ public class CompactionAccordIteratorsTest
private static void flush(AccordCommandStore commandStore)
{
long cacheSize;
try (AccordExecutor.ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
long cacheSize = cache.global.capacity();
cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
}
try (AccordExecutor.ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
cache.global.setCapacity(cacheSize);
}
commandsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
@ -313,7 +317,7 @@ public class CompactionAccordIteratorsTest
Txn txn = txnId.kind().isWrite() ? writeTxn : readTxn;
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.slice(commandStore.unsafeGetRangesForEpoch().currentRanges());
Route<?> partialRoute = route.overlapping(commandStore.unsafeGetRangesForEpoch().currentRanges());
getBlocking(commandStore.execute(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, (a, b) -> {});
}));

View File

@ -37,6 +37,7 @@ import org.junit.Test;
import accord.api.ProtocolModifiers;
import accord.messages.NoWaitRequest;
import accord.api.RoutingKey;
import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.SaveStatus;
@ -209,7 +210,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
CQLTester.setUpClass();
CassandraDaemon.getInstanceForTesting().setupVirtualKeyspaces();
AccordService.startup(ClusterMetadata.current().myNodeId());
AccordService.localStartup(ClusterMetadata.current().myNodeId());
AccordService.distributedStartup();
requireNetwork();
}
@ -235,8 +237,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
MessagingService.instance().outboundSink.add(filter);
try
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
execute(SET_TRACE, 1, "{WaitProgress}", id.toString());
@ -276,8 +278,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
MessagingService.instance().outboundSink.add(filter);
try
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
execute(SET_TRACE_REMOTE, 1, "{WaitProgress}", nodeId, id.toString());
@ -339,8 +341,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.toString(), "*", "{}", "*", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count + 1 ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
@ -356,8 +358,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.asRange().toString(), "{KE}", "{}", "{PreAccept}", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
assertRows(execute(QUERY_TRACE, id.toString()));
}
@ -365,8 +367,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
List<TxnId> txnIds = new ArrayList<>();
for (int i = 0 ; i < count + 1 ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.EphemeralRead, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("SELECT * FROM %s.%s WHERE k = ? AND c = ?", KEYSPACE, tableName)), 0, i);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.EphemeralRead, Routable.Domain.Key);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
@ -379,8 +381,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
}
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, "" + txn.keys().get(0).toUnseekable(), "{KW}", "*", "{}", "leaky", 1, 1, "{}", 1);
AccordMsgFilter filter = new AccordMsgFilter();
@ -461,7 +463,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Keys.of(), Txn.Kind.Write, Routable.Domain.Key);
try
{
execute(ERASE_JOURNAL_REMOTE, nodeId, 1, id.toString());
@ -479,8 +481,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
AccordService accord = accord();
int nodeId = accord.nodeId().id;
AccordMsgFilter filter = new AccordMsgFilter();
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
filter.dropVerbs = Set.of();
MessagingService.instance().outboundSink.add(filter);
@ -562,7 +564,6 @@ public class AccordDebugKeyspaceTest extends CQLTester
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
" IF r IS NULL THEN\n " +
@ -570,6 +571,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
" END IF\n" +
"COMMIT TRANSACTION", KEYSPACE, tableName, KEYSPACE, tableName);
Txn txn = createTxn(insertTxn, 0, 0, 0, 0, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
accord.node().coordinate(id, txn).beginAsResult();
@ -603,16 +605,17 @@ public class AccordDebugKeyspaceTest extends CQLTester
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId first = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
" IF r IS NULL THEN\n " +
" INSERT INTO %s.%s (k, c, v) VALUES (?, ?, ?);\n" +
" END IF\n" +
"COMMIT TRANSACTION", KEYSPACE, tableName, KEYSPACE, tableName);
Txn txn = createTxn(insertTxn, 0, 0, 0, 0, 0);
TxnId first = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(first);
accord.node().coordinate(first, createTxn(insertTxn, 0, 0, 0, 0, 0)).beginAsResult();
accord.node().coordinate(first, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
@ -627,10 +630,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
filter.reset();
TxnId second = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
TxnId second = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.reset();
filter.appliesTo(second);
accord.node().coordinate(second, createTxn(insertTxn, 0, 0, 0, 0, 0)).beginAsResult();
accord.node().coordinate(second, txn).beginAsResult();
filter.commit.awaitThrowUncheckedOnInterrupt();
@ -695,9 +698,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
KEYSPACE,
tableName);
AccordService accord = accord();
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(insertTxn, 0, 0, 0);
TxnId id = accord.node().nextTxnIdWithDefaultFlags(txn.keys(), Txn.Kind.Write, Routable.Domain.Key);
filter.appliesTo(id);
accord.node().coordinate(id, createTxn(insertTxn, 0, 0, 0)).beginAsResult();
accord.node().coordinate(id, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
String QUERY_JOURNAL = String.format("SELECT txn_id, save_status, command_store_id FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);

View File

@ -0,0 +1,351 @@
/*
* 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.db.virtual;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.IntUnaryOperator;
import java.util.stream.IntStream;
import org.junit.Test;
import accord.utils.Invariants;
import accord.utils.LargeBitSet;
import accord.utils.RandomSource;
import accord.utils.RandomTestRunner;
import accord.utils.UnhandledEnum;
import net.openhft.chronicle.core.util.IntBiPredicate;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.schema.TableMetadata;
import org.assertj.core.api.Assertions;
import static org.apache.cassandra.db.virtual.AbstractLazyVirtualTable.OnTimeout.FAIL;
import static org.apache.cassandra.db.virtual.LazyVirtualTableTest.Cmp.CMPS;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.ASC;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.DESC;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.UNSORTED;
public class LazyVirtualTableTest extends CQLTester
{
private static final String KEYSPACE = "system_accord_not_really";
@Test
public void test()
{
DatabaseDescriptor.setRpcTimeout(TimeUnit.MINUTES.toMillis(5L));
RandomTestRunner.test().withSeed(3606318398900145565L).check(this::testOne);
for (int i = 0 ; i < 10 ; ++i)
RandomTestRunner.test().check(this::testOne);
}
private void testOne(RandomSource rnd)
{
QueryProcessor.clearInternalStatementsCache();
QueryProcessor.clearPreparedStatementsCache();
int keyCount = rnd.nextInt(1, 5);
int valCount = rnd.nextInt(0, 3);
int rowCount = rnd.nextInt(0, 12) - 1;
rowCount = rowCount < 0 ? 0 : rnd.nextInt(1 << rowCount, 2 << rowCount);
int[] keyDomains = new int[keyCount];
Arrays.fill(keyDomains, 1);
int uniqueKeys = Math.max(1, rowCount * rnd.nextInt(2, 5));
for (int k = keyDomains.length - 1 ; k > 0 && uniqueKeys > 1 ; --k)
{
int maxScale = 31 - Integer.numberOfLeadingZeros(uniqueKeys);
if (1 << maxScale != uniqueKeys)
++maxScale;
int avgScale = maxScale / (1 + k);
int scale = rnd.nextBiasedInt(0, avgScale, maxScale);
keyDomains[k] = rnd.nextInt(1 << scale, Math.min(uniqueKeys, 2 << scale));
uniqueKeys = (uniqueKeys + keyDomains[k] - 1) / keyDomains[k];
}
keyDomains[0] = uniqueKeys;
int[] valDomains = new int[valCount];
Arrays.fill(valDomains, Integer.MAX_VALUE);
int queryCount = rnd.nextInt(1, Math.max(2, Math.min(1000, rowCount)));
testOne(rnd, keyDomains, valDomains, rowCount, queryCount);
}
private void testOne(RandomSource rnd, int[] keyDomains, int[] valDomains, int rowCount, int queryLoops)
{
{
int uniqueKeys = 1;
for (int keyDomain : keyDomains)
uniqueKeys *= keyDomain;
Invariants.require(uniqueKeys >= 2 * rowCount);
}
int keyCount = keyDomains.length;
int valCount = valDomains.length;
String[] keyNames = IntStream.rangeClosed(1, keyCount).mapToObj(i -> "k" + i).toArray(String[]::new);
String[] valNames = IntStream.rangeClosed(1, valCount).mapToObj(i -> "v" + i).toArray(String[]::new);
int[][] allKeys = new int[rowCount][], allVals = new int[rowCount][];
{
TreeMap<int[], int[]> unique = new TreeMap<>(Arrays::compare);
for (int r = 0 ; r < rowCount ; ++r)
{
int[] vals = new int[valCount];
for (int v = 0 ; v < valCount ; ++v)
vals[v] = rnd.nextInt(valDomains[v]);
int[] keys = new int[keyCount];
do
{
for (int k = 0 ; k < keyCount ; ++k)
keys[k] = rnd.nextInt(keyDomains[k]);
}
while (null != unique.putIfAbsent(keys, vals));
}
int count = 0;
for (Map.Entry<int[], int[]> e : unique.entrySet())
{
allKeys[count] = e.getKey();
allVals[count] = e.getValue();
++count;
}
}
class TestTable extends AbstractLazyVirtualTable
{
final int[][] copyOfKeys = new int[rowCount][], copyOfVals = new int[rowCount][];
TestTable(Sorted sorted, Sorted sortedByPartitionKey)
{
super(LazyVirtualTableTest.metadata(keyCount, valCount, sorted + "_" + sortedByPartitionKey), FAIL, sorted, sortedByPartitionKey);
IntUnaryOperator index;
switch (sorted)
{
default: throw new UnhandledEnum(sorted);
case SORTED: throw new UnsupportedOperationException(); // TODO (desired): test this
case ASC:
index = i -> i;
break;
case UNSORTED:
int[] indexes = IntStream.range(0, rowCount).toArray();
index = i -> indexes[i];
switch (sortedByPartitionKey)
{
default: throw new UnhandledEnum(sortedByPartitionKey);
case DESC: case SORTED: throw new UnsupportedOperationException(); // TODO (desired): test this
case UNSORTED:
shuffle(indexes, 0, indexes.length, rnd);
break;
case ASC:
int prev = 0;
for (int i = 1; i < rowCount ; ++i)
{
if (allKeys[prev][0] != allKeys[i][0])
{
shuffle(indexes, prev, i, rnd);
prev = i;
}
}
shuffle(indexes, prev, indexes.length, rnd);
}
break;
case DESC:
index = i -> rowCount - (i + 1);
break;
}
for (int i = 0 ; i < rowCount ; ++i)
{
copyOfKeys[i] = allKeys[index.applyAsInt(i)];
copyOfVals[i] = allVals[index.applyAsInt(i)];
}
}
@Override
protected void collect(PartitionsCollector collector)
{
for (int i = 0 ; i < copyOfKeys.length ; ++i)
{
int[] keys = copyOfKeys[i];
int[] vals = copyOfVals[i];
Object[] ckeys = new Object[keyCount - 1];
for (int k = 1; k < keyCount ; ++k)
ckeys[k - 1] = keys[k];
collector.partition(keys[0])
.collect(rows -> rows.add(ckeys).lazyCollect(cols -> {
for (int j = 0 ; j < vals.length ; ++j)
cols.add(valNames[j], vals[j]);
}));
}
}
}
List<VirtualTable> tables = new ArrayList<>();
tables.add(new TestTable(ASC, ASC));
tables.add(new TestTable(DESC, DESC));
tables.add(new TestTable(UNSORTED, ASC));
tables.add(new TestTable(UNSORTED, UNSORTED));
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KEYSPACE, tables));
LargeBitSet matchingRows = new LargeBitSet(rowCount);
int fieldCount = keyCount + valCount;
LargeBitSet filteringKeys = new LargeBitSet(fieldCount);
StringBuilder where = new StringBuilder();
for (int q = 0 ; q < queryLoops ; q++)
{
matchingRows.setRange(0, rowCount);
filteringKeys.setRange(0, fieldCount);
where.setLength(0);
do
{
int f;
do
{
int min = filteringKeys.nextSetBit(0);
f = rnd.nextInt(min, fieldCount);
} while (!filteringKeys.unset(f));
int val;
if (rowCount == 0 || rnd.nextBoolean())
{
val = rnd.nextInt(f < keyCount ? keyDomains[f] : valDomains[f - keyCount]);
}
else
{
int i = rnd.nextInt(rowCount);
val = f < keyCount ? allKeys[i][f] : allVals[i][f - keyCount];
}
Cmp cmp = rnd.pick(CMPS);
if (where.length() > 0)
where.append(" AND ");
where.append(f < keyCount ? keyNames[f] : valNames[f - keyCount]);
where.append(' ');
where.append(cmp.str);
where.append(' ');
where.append(val);
IntBiPredicate pred = cmp.predicate();
for (int i = matchingRows.nextSetBit(0); i >= 0; i = matchingRows.nextSetBit(i + 1, -1))
{
int rowVal = f < keyCount ? allKeys[i][f] : allVals[i][f - keyCount];
if (!pred.test(rowVal, val))
matchingRows.unset(i);
}
int matchCount = matchingRows.getSetBitCount();
int limit = rnd.nextBoolean() ? 0 : rnd.nextInt(1, Math.max(2, 2 * matchCount));
for (VirtualTable table : tables)
{
UntypedResultSet results = execute("select * from " + table.metadata() + " where " + where + (limit <= 0 ? "" : " LIMIT " + limit));
int i = -1;
for (UntypedResultSet.Row row : results)
{
i = matchingRows.nextSetBit(i + 1);
for (int k = 0 ; k < keyCount ; ++k)
Assertions.assertThat(allKeys[i][k]).isEqualTo(row.getInt(keyNames[k]));
for (int v = 0 ; v < valCount ; ++v)
Assertions.assertThat(allVals[i][v]).isEqualTo(row.getInt(valNames[v]));
}
i = matchingRows.nextSetBit(i + 1);
if (limit >= matchCount || limit == 0) Assertions.assertThat(-1).isEqualTo(i);
else Assertions.assertThat(i).isNotEqualTo(-1);
Assertions.assertThat(results.size()).isEqualTo(limit == 0 ? matchCount : Math.min(matchCount, limit));
}
} while (matchingRows.getSetBitCount() > 0 && filteringKeys.getSetBitCount() > 0);
}
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(KEYSPACE, tables));
}
enum Cmp
{
LT("<"), LE("<="), NE("!="), EQ("="), GE(">="), GT(">");
static final Cmp[] CMPS = values();
final String str;
Cmp(String str)
{
this.str = str;
}
IntBiPredicate predicate()
{
switch (this)
{
default: throw new UnhandledEnum(this);
case LT: return (a, b) -> a < b;
case LE: return (a, b) -> a <= b;
case NE: return (a, b) -> a != b;
case EQ: return (a, b) -> a == b;
case GE: return (a, b) -> a >= b;
case GT: return (a, b) -> a > b;
}
}
}
private static TableMetadata metadata(int keys, int vals, String name)
{
return CreateTableStatement.parse(createTable(keys, vals, name), KEYSPACE)
.comment("")
.kind(TableMetadata.Kind.VIRTUAL)
.partitioner(new LocalPartitioner(Int32Type.instance))
.build();
}
private static String createTable(int keys, int vals, String name)
{
StringBuilder sb = new StringBuilder("CREATE TABLE ");
sb.append(name).append(" (");
for (int i = 1 ; i <= keys ; ++i)
sb.append('k').append(i).append(" int, ");
for (int i = 1 ; i <= vals ; ++i)
sb.append('v').append(i).append(" int, ");
sb.append(" PRIMARY KEY (");
for (int i = 1 ; i <= keys ; ++i)
{
if (i > 1) sb.append(", ");
sb.append('k').append(i);
}
sb.append("))");
return sb.toString();
}
private static void shuffle(int[] array, int start, int end, RandomSource rnd)
{
while (end - start > 1)
{
int i = start + rnd.nextInt(end - start);
int tmp = array[start];
array[start] = array[i];
array[i] = tmp;
++start;
}
}
}

View File

@ -269,7 +269,7 @@ public class LocalRepairTablesTest extends CQLTester
private static RepairCoordinator.NeighborsAndRanges neighbors()
{
return new RepairCoordinator.NeighborsAndRanges(false, ADDRESSES, ImmutableList.of(COMMON_RANGE));
return new RepairCoordinator.NeighborsAndRanges(false, false, ADDRESSES, ImmutableList.of(COMMON_RANGE));
}
private static Range<Token> range(long a, long b)

View File

@ -521,7 +521,8 @@ public class RouteIndexTest extends CQLTester
AccordService.MetadataChangeListener.instance.resetForTesting(metadata);
NodeId tcmNodeId = metadata.myNodeId();
AccordService.unsafeSetNewAccordService(null);
return AccordService.startup(tcmNodeId);
AccordService.localStartup(tcmNodeId);
return AccordService.distributedStartup();
}
TxnId nextTxnId(Domain domain)

View File

@ -40,9 +40,9 @@ public class NeighborsAndRangesTest extends AbstractRepairTest
public void filterCommonIncrementalRangesNotForced()
{
CommonRange cr = new CommonRange(PARTICIPANTS, Collections.emptySet(), ALL_RANGES);
NeighborsAndRanges nr = new NeighborsAndRanges(false, PARTICIPANTS, Collections.singletonList(cr));
NeighborsAndRanges nr = new NeighborsAndRanges(false, false, PARTICIPANTS, Collections.singletonList(cr));
List<CommonRange> expected = Lists.newArrayList(cr);
List<CommonRange> actual = nr.filterCommonRanges(null, null);
List<CommonRange> actual = nr.filterCommonRanges(null, null).commonRanges;
Assert.assertEquals(expected, actual);
}
@ -59,8 +59,8 @@ public class NeighborsAndRangesTest extends AbstractRepairTest
new CommonRange(Sets.newHashSet(PARTICIPANT2, PARTICIPANT3), Collections.emptySet(), Sets.newHashSet(RANGE3), true),
new CommonRange(Sets.newHashSet(PARTICIPANT2, PARTICIPANT3), Collections.emptySet(), Sets.newHashSet(RANGE2), false));
NeighborsAndRanges nr = new NeighborsAndRanges(true, liveEndpoints, initial);
List<CommonRange> actual = nr.filterCommonRanges(null, null);
NeighborsAndRanges nr = new NeighborsAndRanges(true, false, liveEndpoints, initial);
List<CommonRange> actual = nr.filterCommonRanges(null, null).commonRanges;
Assert.assertEquals(expected, actual);
}

View File

@ -128,8 +128,8 @@ public class RepairJobTest
boolean dontPurgeTombstones, boolean repairAccord, String... cfnames)
{
super(SharedContext.Global.instance, new Scheduler.NoopScheduler(),
parentRepairSession, commonRange, excludedDeadNodes, keyspace, parallelismDegree, isIncremental, pullRepair,
previewKind, optimiseStreams, repairData, repairPaxos, dontPurgeTombstones, repairAccord, cfnames);
parentRepairSession, commonRange, excludedDeadNodes, keyspace, parallelismDegree, false, isIncremental, pullRepair,
previewKind, optimiseStreams, repairData, repairPaxos, dontPurgeTombstones, repairAccord, false, cfnames);
}
@Override

View File

@ -66,8 +66,8 @@ public class RepairSessionTest
RepairSession session = new RepairSession(SharedContext.Global.instance, new Scheduler.NoopScheduler(), parentSessionId,
new CommonRange(endpoints, Collections.emptySet(), Arrays.asList(repairRange)),
false, "Keyspace1", RepairParallelism.SEQUENTIAL,
false, false, PreviewKind.NONE, false,
false, false, false, false, "Standard1");
false, false, false, PreviewKind.NONE, false,
false, false, false, false, false, "Standard1");
// perform convict
session.convict(remote, Double.MAX_VALUE);

View File

@ -37,8 +37,6 @@ import org.apache.cassandra.cql3.statements.schema.TableAttributes;
import org.apache.cassandra.repair.RepairCoordinator;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.repair.RepairParallelism;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.schema.SystemDistributedKeyspace;
import org.apache.cassandra.service.StorageService;
@ -62,7 +60,6 @@ import org.apache.cassandra.metrics.AutoRepairMetrics;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.AutoRepairService;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.progress.ProgressEvent;
import org.apache.cassandra.utils.progress.ProgressEventType;
@ -811,9 +808,6 @@ public class AutoRepairParameterizedTest extends CQLTester
@Test
public void testSchedulerIgnoresErrorsFromUnrelatedRepairRunables()
{
RepairOption options = new RepairOption(RepairParallelism.PARALLEL, true, repairType == AutoRepairConfig.RepairType.INCREMENTAL, false,
AutoRepairService.instance.getAutoRepairConfig().getRepairThreads(repairType), Collections.emptySet(),
false, false, PreviewKind.NONE, false, true, true, false, false, false);
AutoRepairState repairState = AutoRepair.instance.repairStates.get(repairType);
AutoRepairState spyState = spy(repairState);
AtomicReference<AutoRepair.RepairProgressListener> failingListener = new AtomicReference<>();

View File

@ -280,7 +280,8 @@ public class AccordCacheTest
public void testEvictionOnAcquire()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
NoOpLock lock = new NoOpLock();
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 5);
@ -297,6 +298,7 @@ public class AccordCacheTest
testLoad(executor, safeString, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(safeString.key()));
instance.release(safeString, null);
cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 0, 5, nodeSize(1) * 5);
@ -305,6 +307,7 @@ public class AccordCacheTest
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
SafeString safeString = instance.acquire("5");
cache.tryShrinkOrEvict(lock);
Assert.assertTrue(instance.isReferenced(safeString.key()));
// since it's not loaded, only the node size is counted here
@ -327,7 +330,8 @@ public class AccordCacheTest
public void testEvictionOnRelease()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
NoOpLock lock = new NoOpLock();
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 4);
@ -343,6 +347,7 @@ public class AccordCacheTest
items[i] = safeString;
testLoad(executor, safeString, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(safeString.key()));
cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 5, 5, nodeSize(1) * 5);
@ -351,12 +356,14 @@ public class AccordCacheTest
Assert.assertNull(cache.tail());
instance.release(items[2], null);
cache.tryShrinkOrEvict(lock);
assertCacheState(cache, 4, 4, nodeSize(1) * 4);
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
Assert.assertNull(cache.head());
Assert.assertNull(cache.tail());
instance.release(items[4], null);
cache.tryShrinkOrEvict(lock);
assertCacheState(cache, 3, 4, nodeSize(1) * 4);
assertCacheMetrics(cacheMetrics, 0, 5, 5, 5);
Assert.assertSame(items[4].global, cache.head());
@ -401,7 +408,8 @@ public class AccordCacheTest
public void evictionBlockedOnSaving()
{
AccordCacheMetrics cacheMetrics = new AccordCacheMetrics(nextMetricId());
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(new NoOpLock());
NoOpLock lock = new NoOpLock();
AccordCacheMetrics.Shard shard = cacheMetrics.newShard(lock);
ManualExecutor executor = new ManualExecutor();
AccordCache cache = new AccordCache(saveExecutor(executor), nodeSize(1) * 3 + nodeSize(3));
@ -422,6 +430,7 @@ public class AccordCacheTest
testLoad(executor, item, Integer.toString(i));
Assert.assertTrue(instance.isReferenced(item.key()));
instance.release(item, null);
cache.tryShrinkOrEvict(lock);
}
assertCacheState(cache, 0, 4, nodeSize(1) * 3 + nodeSize(2));
@ -430,6 +439,7 @@ public class AccordCacheTest
// force cache eviction
instance.acquire(Integer.toString(0));
cache.setCapacity(0);
cache.tryShrinkOrEvict(lock);
// all should have been evicted except 0
assertCacheState(cache, 1, 1, nodeSize(2));

View File

@ -103,7 +103,7 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
Route<?> route = fullRoute.slice(fullRange(txn));
Route<?> route = fullRoute.overlapping(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept = PreAccept.SerializerSupport.create(txnId, route, 1, 1, 1, partialTxn, null, false, fullRoute);
@ -195,7 +195,7 @@ public class AccordCommandTest
Key key = (Key)txn.keys().get(0);
RoutingKey homeKey = key.toUnseekable();
FullRoute<?> fullRoute = txn.keys().toRoute(homeKey);
Route<?> route = fullRoute.slice(fullRange(txn));
Route<?> route = fullRoute.overlapping(fullRange(txn));
PartialTxn partialTxn = txn.intersecting(route, true);
PreAccept preAccept1 = PreAccept.SerializerSupport.create(txnId1, route, 1, 1, 1, partialTxn, null, false, fullRoute);

View File

@ -28,7 +28,6 @@ import accord.messages.ReadData;
import accord.messages.ReadData.CommitOrReadNack;
import accord.topology.TopologyUtils;
import org.apache.cassandra.service.accord.AccordFetchCoordinator.AccordFetchRequest;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.AccordTimeService;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
@ -72,7 +71,7 @@ public class AccordMessageSinkTest
DatabaseDescriptor.clientInitialization();
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
ClusterMetadataService.initializeForClients();
sink = new AccordMessageSink(Mockito.mock(AccordAgent.class), messaging, mapping, new RequestCallbacks(new AccordTimeService()));
sink = new AccordMessageSink(messaging, mapping, new RequestCallbacks(new AccordTimeService()));
}
@Test

View File

@ -195,7 +195,7 @@ public class AccordTaskTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
route.slice(ranges);
route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
@ -205,10 +205,15 @@ public class AccordTaskTest
}));
// clear cache
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
long cacheSize;
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches())
{
long cacheSize = cache.global.capacity();
cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
}
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches())
{
cache.global.setCapacity(cacheSize);
}
@ -238,7 +243,7 @@ public class AccordTaskTest
RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable();
FullRoute<?> route = partialTxn.keys().toRoute(routingKey);
Ranges ranges = AccordTestUtils.fullRange(partialTxn.keys());
Route<?> partialRoute = route.slice(ranges);
Route<?> partialRoute = route.overlapping(ranges);
PartialDeps deps = PartialDeps.builder(ranges, true).build();
Command command = getBlocking(commandStore.submit(contextFor(txnId, route, SYNC, READ_WRITE, "Test"), safe -> {
@ -250,10 +255,14 @@ public class AccordTaskTest
}));
// clear cache
long cacheSize;
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
long cacheSize = cache.global.capacity();
cacheSize = cache.global.capacity();
cache.global.setCapacity(0);
}
try (ExclusiveGlobalCaches cache = commandStore.executor().lockCaches();)
{
cache.global.setCapacity(cacheSize);
}
@ -280,6 +289,7 @@ public class AccordTaskTest
.withExamples(50)
.forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 2))
.check((rs, ids) -> {
before(); // truncate tables
Participants<RoutingKey> participants = keys.toParticipants();

View File

@ -392,7 +392,7 @@ public class EpochSyncTest
.isFalse();
// validate topology manager
Ranges ranges = tm.active().globalForEpoch(epoch).ranges().mergeTouching();
Ranges ranges = tm.active().getKnown(epoch).global().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))
@ -404,7 +404,7 @@ public class EpochSyncTest
continue;
Assertions.assertThat(tm.active().hasEpoch(epoch)).describedAs("node%s does not have epoch %d", id, epoch).isTrue();
Topology topology = tm.active().globalForEpoch(epoch);
Topology topology = tm.active().getKnown(epoch).global();
Ranges ranges = topology.ranges().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

View File

@ -159,7 +159,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
this.topologies = new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, topology);
Ranges ranges = topology.ranges();
if (tableId != null)
ranges = ranges.slice(Ranges.of(TokenRange.create(TokenKey.min(tableId, getPartitioner()), TokenKey.max(tableId, getPartitioner()))));
ranges = ranges.overlapping(Ranges.of(TokenRange.create(TokenKey.min(tableId, getPartitioner()), TokenKey.max(tableId, getPartitioner()))));
CommandStores.RangesForEpoch rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), ranges);
updateHolder.add(topology.epoch(), rangesForEpoch, ranges);
@ -258,10 +258,10 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
@Override
public void onUncaughtException(Throwable t)
public void onException(Throwable t)
{
if (ignoreExceptions.test(t)) return;
super.onUncaughtException(t);
super.onException(t);
}
};

View File

@ -662,8 +662,8 @@ public class CommandsForKeySerializerTest
@Override public void shutdown() { }
@Override public <T> AsyncChain<T> chain(Callable<T> call) { throw new UnsupportedOperationException(); }
@Override public OwnershipEventListener ownershipEvents() { return null; }
@Override public void onUncaughtException(Throwable t) { throw new UnsupportedOperationException(); }
@Override public void onCaughtException(Throwable t, String context) { throw new UnsupportedOperationException(); }
@Override public void onException(Throwable t) { throw new UnsupportedOperationException(); }
@Override public void onException(Throwable t, String context) { throw new UnsupportedOperationException(); }
@Override public boolean rejectPreAccept(TimeService time, TxnId txnId) { throw new UnsupportedOperationException(); }
@Override public long cfkHlcPruneDelta() { return 0; }
@Override public int cfkPruneInterval() { return 0; }

View File

@ -181,7 +181,7 @@ public class KeySerializersTest
int count = superset.isEmpty() ? 0 : rs.nextInt(superset.size());
Participants<?> subset = selectSubset(rs, count, superset);
if (superset instanceof Route<?> && (!changeType || rs.nextBoolean()))
return superset.intersecting(subset);
return superset.overlapping(subset);
return subset;
}

View File

@ -122,7 +122,7 @@ public class TxnUpdateTest
}
// slice the same key should return the same block
TxnUpdate noUpdate = update.getTxnUpdate(k -> k.intersecting(update.keys()));
TxnUpdate noUpdate = update.getTxnUpdate(k -> k.overlapping(update.keys()));
for (int i = 0; i < update.blocks.size(); i++)
assertThat(noUpdate.blocks.get(i)).isSameAs(update.blocks.get(i));
@ -131,7 +131,7 @@ public class TxnUpdateTest
int keyIndex = rs.nextInt(0, update.keys().size());
Key key = update.keys().get(keyIndex);
Keys singleKey = Keys.of(key);
TxnUpdate singleKeyUpdate = update.getTxnUpdate(k -> k.intersecting(singleKey));
TxnUpdate singleKeyUpdate = update.getTxnUpdate(k -> k.overlapping(singleKey));
for (int i = 0; i < update.blocks.size(); i++)
{
var block = singleKeyUpdate.blocks.get(i);
@ -162,7 +162,7 @@ public class TxnUpdateTest
for (int i = 0; i < update.keys().size(); i++)
{
int finalI = i;
perKeyUpdate.add(update.getTxnUpdate(k -> k.intersecting(Keys.of(update.keys().get(finalI)))));
perKeyUpdate.add(update.getTxnUpdate(k -> k.overlapping(Keys.of(update.keys().get(finalI)))));
}
assertThat(update.merge(update)).isEqualTo(update); // merge with self produces self

View File

@ -75,7 +75,7 @@ public class PaxosStateTrackerTest
{
SchemaLoader.prepareServer();
DatabaseDescriptor.setAccordTransactionsEnabled(false);
AccordService.startup(null);
AccordService.localStartup(null);
ks = "coordinatorsessiontest";
cfm1 = TableMetadata.builder(ks, "tbl1").addPartitionKeyColumn("k", Int32Type.instance).addRegularColumn("v", Int32Type.instance).build();