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

Enrich Durability with each phase, so we can both prune unapplied dependencies to mitigate replicas that are behind causing Deps growth, 

and more reliably avoid initialising recovery progress log state of transactions that cannot yet make progress
Also fix:
 - Harden AccordExecutor state cleanup to failures
 - Handle SAVING state in AccordCache.tryEvict, as now possible to save for reasons besides eviction so normal to both be in evict queue and saving
 - Infer invalid in MaybeRecover and FetchData
 - MaybeRecover sometimes aborts before home shard knows outcome
 - Epoch sync with VisibilitySyncPoint
 - Retired implies synced
 - Don't interpret force repair as excluding nodes from Accord sync conditions
 - TxnData.without
Also improve:
 - Add Topology.removedNodes
 - If Durability implies we can fetch a status, update the waiting state to fetch it
 - DurableBefore debug table should have searchable txnId

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20832
This commit is contained in:
Benedict Elliott Smith 2025-08-08 10:25:17 +01:00
parent cdfe2c3dac
commit 81bc18ad63
26 changed files with 110 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 12da4693b449b30d0420673579a06025b7b3e484
Subproject commit b86550c154758e5dd223fa8101a2a83005b644a6

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

@ -129,7 +129,7 @@ import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STOR
import static accord.local.RedundantStatus.Property.LOCALLY_REDUNDANT;
import static accord.local.RedundantStatus.Property.LOCALLY_SYNCED;
import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
import static accord.local.RedundantStatus.Property.MAJORITY_APPLIED;
import static accord.local.RedundantStatus.Property.QUORUM_APPLIED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
import static accord.utils.async.AsyncChains.getBlockingAndRethrow;
@ -459,8 +459,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
" table_name text,\n" +
" token_start 'TokenUtf8Type',\n" +
" token_end 'TokenUtf8Type',\n" +
" majority_before text,\n" +
" universal_before text,\n" +
" quorum 'TxnIdUtf8Type',\n" +
" universal 'TxnIdUtf8Type',\n" +
" PRIMARY KEY (keyspace_name, table_name, token_start)" +
')', UTF8Type.instance));
}
@ -475,8 +475,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
TableMetadata tableMetadata = tableMetadata(tableId);
ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), printToken(start))
.column("token_end", printToken(end))
.column("majority_before", entry.majorityBefore.toString())
.column("universal_before", entry.universalBefore.toString());
.column("quorum", entry.quorumBefore.toString())
.column("universal", entry.universalBefore.toString());
return ds;
},
new SimpleDataSet(metadata()),
@ -751,7 +751,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
" end_epoch bigint,\n" +
" gc_before 'TxnIdUtf8Type',\n" +
" shard_applied 'TxnIdUtf8Type',\n" +
" majority_applied 'TxnIdUtf8Type',\n" +
" quorum_applied 'TxnIdUtf8Type',\n" +
" locally_applied 'TxnIdUtf8Type',\n" +
" locally_durable_to_command_store 'TxnIdUtf8Type',\n" +
" locally_durable_to_data_store 'TxnIdUtf8Type',\n" +
@ -786,7 +786,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
.column("end_epoch", entry.endEpoch)
.column("gc_before", entry.maxBound(GC_BEFORE).toString())
.column("shard_applied", entry.maxBound(SHARD_APPLIED).toString())
.column("majority_applied", entry.maxBound(MAJORITY_APPLIED).toString())
.column("quorum_applied", entry.maxBound(QUORUM_APPLIED).toString())
.column("locally_applied", entry.maxBound(LOCALLY_APPLIED).toString())
.column("locally_durable_to_command_store", entry.maxBound(LOCALLY_DURABLE_TO_COMMAND_STORE).toString())
.column("locally_durable_to_data_store", entry.maxBound(LOCALLY_DURABLE_TO_DATA_STORE).toString())

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

@ -340,7 +340,6 @@ public enum Verb
ACCORD_RECOVER_AWAIT_RSP (141, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverReply), AccordService::responseHandlerOrNoop ),
ACCORD_RECOVER_AWAIT_REQ (142, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverRequest), AccordService::requestHandlerOrNoop, ACCORD_RECOVER_AWAIT_RSP),
ACCORD_INFORM_DURABLE_REQ (143, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(InformSerializers.durable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ),
ACCORD_INFORM_DECIDED_REQ (171, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(InformSerializers.decided), AccordService::requestHandlerOrNoop ),
ACCORD_CHECK_STATUS_RSP (144, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.reply), AccordService::responseHandlerOrNoop ),
ACCORD_CHECK_STATUS_REQ (145, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ),
ACCORD_FETCH_DATA_RSP (146, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(FetchSerializers.reply), AccordService::responseHandlerOrNoop ),

2
src/java/org/apache/cassandra/repair/RepairJob.java
View File

@ -192,7 +192,7 @@ public class RepairJob extends AsyncFuture<RepairResult> implements Runnable
requireAllEndpoints = true;
}
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);
AccordRepair repair = new AccordRepair(ctx, cfs, desc.sessionId, desc.keyspace, desc.ranges, requireAllEndpoints);
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)

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

@ -267,9 +267,11 @@ public class AccordCache implements CacheSize
break;
case MODIFIED:
node.save();
case SAVING: // we can be in evict queue and already be saving if save was requested for durability rather than eviction
boolean evict = node.status() == LOADED;
node.unlink();
if (evict) evict(node, true);
break;
}
}

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

@ -452,13 +452,11 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
{
long epoch = topology.epoch();
EpochState epochState = getOrCreateEpochState(epoch);
if (!startSync || epochState.syncStatus != SyncStatus.NOT_STARTED)
return;
synchronized (this)
{
if (epochState.syncStatus != SyncStatus.NOT_STARTED)
if (!startSync || epochState.syncStatus != SyncStatus.NOT_STARTED)
return;
epochState.setSyncStatus(SyncStatus.NOTIFYING);
}
@ -651,7 +649,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
public Future<Void> unsafeLocalSyncNotified(long epoch)
{
AsyncPromise<Void> promise = new AsyncPromise<>();
getOrCreateEpochState(epoch).localSyncNotified().invoke((result, failure) -> {
getOrCreateEpochState(epoch).localSyncNotified().begin((result, failure) -> {
if (failure != null) promise.tryFailure(failure);
else promise.trySuccess(result);
});

38
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

@ -566,8 +566,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
void submitExclusive(AccordTask<?> task)
{
assignQueuePosition(task);
++tasks;
task.setupExclusive();
++tasks;
updateQueue(task);
enqueueLoadsExclusive();
}
@ -616,21 +616,26 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
task.queuePosition = parent.queuePosition;
}
void completeTaskExclusive(Task task, boolean cleanupTask)
void completeTaskExclusive(Task task)
{
--tasks;
// for integration with SequentialExecutor, we must :
// - first take the position so that represents the just-executed task
// - call cleanup to submit any following task on the relevant sub-queue
// - remove the previous task from the running collection only if still present (SequentialExecutor will have removed it)
int position = task.queuePosition;
if (cleanupTask) task.cleanupExclusive();
if (running.contains(task))
running.remove(task);
try
{
task.cleanupExclusive();
}
finally
{
--tasks;
if (running.contains(task))
running.remove(task);
if (waitingForCompletion != null && waitingForCompletion.peek().maybeNotify - position >= 0)
maybeNotifyWaitingForCompletion();
if (waitingForCompletion != null && waitingForCompletion.peek().maybeNotify - position >= 0)
maybeNotifyWaitingForCompletion();
}
}
private void maybeNotifyWaitingForCompletion()
@ -655,14 +660,15 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
return task == null ? min : Integer.min(task.queuePosition, min);
}
private void cancelExclusive(AccordTask<?> task)
void cancelExclusive(AccordTask<?> task)
{
switch (task.state())
{
default: throw new UnhandledEnum(task.state());
case INITIALIZED:
// we could be cancelled before we even reach the queue
task.cancelExclusive();
try { task.cancelExclusive(); }
finally { task.cleanupExclusive(); }
break;
case SCANNING_RANGES:
@ -671,8 +677,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
case WAITING_TO_SCAN_RANGES:
case WAITING_TO_RUN:
task.unqueueIfQueued();
task.cancelExclusive();
completeTaskExclusive(task, false);
try { task.cancelExclusive(); }
finally { completeTaskExclusive(task); }
break;
case FAILING:
@ -715,7 +721,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
finally
{
task.unqueueIfQueued();
completeTaskExclusive(task, true);
completeTaskExclusive(task);
}
}
@ -1000,9 +1006,9 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
owner = null;
running = false;
task = super.poll();
AccordExecutor.this.running.remove(selfTask);
if (task != null)
{
AccordExecutor.this.running.remove(selfTask);
selfTask.queuePosition = task.queuePosition;
waitingToRun.append(selfTask);
}
@ -1241,7 +1247,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
if (queue.contains(this))
{
queue.remove(this);
completeTaskExclusive(this, true);
completeTaskExclusive(this);
try { fail(new CancellationException()); }
catch (Throwable t) { agent.onUncaughtException(t); }
}

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

@ -180,7 +180,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
}
finally
{
completeTaskExclusive(task, true);
completeTaskExclusive(task);
clearRunning();
}
}
@ -234,7 +234,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
{
Task tmp = task;
task = null;
completeTaskExclusive(tmp, true);
completeTaskExclusive(tmp);
clearRunning();
}
else resumeExclusive();
@ -270,7 +270,7 @@ abstract class AccordExecutorAbstractLockLoop extends AccordExecutor
{
try { task.fail(t); }
catch (Throwable t2) { t.addSuppressed(t2); }
try { completeTaskExclusive(task, true); }
try { completeTaskExclusive(task); }
catch (Throwable t2) { t.addSuppressed(t2); }
try { agent.onUncaughtException(t); }
catch (Throwable t2) { /* nothing we can sensibly do after already reporting */ }

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

@ -96,7 +96,7 @@ class AccordExecutorSimple extends AccordExecutor
catch (Throwable t) { task.fail(t); }
finally
{
completeTaskExclusive(task, true);
completeTaskExclusive(task);
active = null;
}
}

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

@ -816,7 +816,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
out.writeByte(command.saveStatus().ordinal());
break;
case DURABILITY:
out.writeByte(command.durability().ordinal());
out.writeByte(command.durability().encoded());
break;
case ACCEPTED:
CommandSerializers.ballot.serialize(command.acceptedOrCommitted(), out);
@ -954,7 +954,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
break;
case DURABILITY:
Invariants.require(durability != null, "%s", this);
out.writeByte(durability.ordinal());
out.writeByte(durability.encoded());
break;
case ACCEPTED:
Invariants.require(acceptedOrCommitted != null, "%s", this);
@ -1035,7 +1035,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
saveStatus = SaveStatus.values()[in.readByte()];
break;
case DURABILITY:
durability = Durability.values()[in.readByte()];
durability = Durability.forEncoded(in.readUnsignedByte());
break;
case ACCEPTED:
acceptedOrCommitted = CommandSerializers.ballot.deserialize(in);

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

@ -120,7 +120,6 @@ public class AccordMessageSink implements MessageSink
builder.put(WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_WAIT_UNTIL_APPLIED_REQ);
builder.put(APPLY_THEN_WAIT_UNTIL_APPLIED_REQ, Verb.ACCORD_APPLY_AND_WAIT_REQ);
builder.put(INFORM_DURABLE_REQ, Verb.ACCORD_INFORM_DURABLE_REQ);
builder.put(INFORM_DECIDED_REQ, Verb.ACCORD_INFORM_DECIDED_REQ);
builder.put(CHECK_STATUS_REQ, Verb.ACCORD_CHECK_STATUS_REQ);
builder.put(CHECK_STATUS_RSP, Verb.ACCORD_CHECK_STATUS_RSP);
builder.put(FETCH_DATA_REQ, Verb.ACCORD_FETCH_DATA_REQ);

4
src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
View File

@ -221,7 +221,7 @@ public class AccordObjectSizes
}
}
private static final long EMPTY_TXN = measure(new PartialTxn.InMemory(null, null, null, null, null, TableMetadatasAndKeys.none(Domain.Key)));
private static final long EMPTY_TXN = measure(new PartialTxn.InMemory(Kind.Read, null, null, null, null, TableMetadatasAndKeys.none(Domain.Key)));
public static long txn(PartialTxn txn)
{
long size = EMPTY_TXN;
@ -316,7 +316,7 @@ public class AccordObjectSizes
builder.partialDeps(PartialDeps.NONE);
if (hasTxn)
builder.partialTxn(new PartialTxn.InMemory(null, null, null, null, null, TableMetadatasAndKeys.none(Domain.Key)));
builder.partialTxn(new PartialTxn.InMemory(Kind.Read, null, null, null, null, TableMetadatasAndKeys.none(Domain.Key)));
if (executes)
{

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

@ -139,6 +139,7 @@ import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
import static accord.primitives.Txn.Kind.Write;
import static accord.primitives.TxnId.Cardinality.cardinality;
import static accord.topology.TopologyManager.TopologyRange;
@ -533,7 +534,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public AsyncChain<Void> sync(Object requestedBy, Timestamp minBound, Ranges ranges, @Nullable Collection<Id> include, DurabilityService.SyncLocal syncLocal, DurabilityService.SyncRemote syncRemote, long timeout, TimeUnit timeoutUnits)
{
return node.durability().sync(requestedBy, minBound, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits);
return node.durability().sync(requestedBy, ExclusiveSyncPoint, minBound, ranges, include, syncLocal, syncRemote, timeout, timeoutUnits);
}
@Override
@ -1068,7 +1069,7 @@ public class AccordService implements IAccordService, Shutdownable
long startedAt = nanoTime();
long deadline = startedAt + timeout;
// TODO (required): relax this requirement - too expensive
getBlocking(node.durability().sync("Drop Keyspace/Table (Epoch " + epoch + ')', TxnId.minForEpoch(epoch), ranges, Self, All, DatabaseDescriptor.getAccordRangeSyncPointTimeoutNanos(), NANOSECONDS), ranges, new LatencyRequestBookkeeping(null), startedAt, deadline, false);
getBlocking(node.durability().sync("Drop Keyspace/Table (Epoch " + epoch + ')', ExclusiveSyncPoint, TxnId.minForEpoch(epoch), ranges, Self, All, DatabaseDescriptor.getAccordRangeSyncPointTimeoutNanos(), NANOSECONDS), ranges, new LatencyRequestBookkeeping(null), startedAt, deadline, false);
}
public Params journalConfiguration()

7
src/java/org/apache/cassandra/service/accord/AccordTask.java
View File

@ -754,9 +754,9 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
protected void cleanupExclusive()
{
releaseResources(commandStore.cachesExclusive());
if (state == FAILING)
state(FAILED);
releaseResources(commandStore.cachesExclusive());
}
@Nullable
@ -778,7 +778,6 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
public void cancelExclusive()
{
releaseResources(commandStore.cachesExclusive());
state(CANCELLED);
if (callback != null)
callback.accept(null, new CancellationException());
@ -786,7 +785,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
void cancelExclusive(AccordExecutor owner)
{
owner.cancel(this);
owner.cancelExclusive(this);
}
public State state()
@ -839,7 +838,7 @@ public abstract class AccordTask<R> extends SubmittableTask implements Function<
catch (Throwable t)
{
releaseResourcesSlow(caches, t);
throw t;
commandStore.agent().onUncaughtException(t);
}
}

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

@ -35,6 +35,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import accord.local.Node;
import accord.local.Node.Id;
import accord.primitives.Ranges;
import accord.topology.Shard;
@ -307,7 +308,12 @@ public class AccordTopology
}
res.sort((a, b) -> a.range.compare(b.range));
return new Topology(epoch.getEpoch(), SortedArrayList.copyUnsorted(staleReplicas.ids(), Id[]::new), res.toArray(new Shard[0]));
List<Node.Id> removed = directory.removedNodes().stream()
.filter(n -> n.removedIn.equals(epoch))
.map(n -> tcmIdToAccord(n.id))
.collect(Collectors.toList());
return new Topology(epoch.getEpoch(), SortedArrayList.copySorted(removed, Id[]::new), SortedArrayList.copyUnsorted(staleReplicas.ids(), Id[]::new), res.toArray(new Shard[0]));
}
public static Topology createAccordTopology(ClusterMetadata metadata, ShardLookup lookup)

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

@ -22,21 +22,15 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.Node;
import accord.local.durability.DurabilityService;
import accord.local.durability.DurabilityService.SyncRemote;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.LatencyMetrics;
import org.apache.cassandra.repair.SharedContext;
import org.apache.cassandra.schema.Schema;
@ -68,29 +62,25 @@ import static org.apache.cassandra.config.DatabaseDescriptor.getAccordRepairTime
*/
public class AccordRepair
{
private static final Logger logger = LoggerFactory.getLogger(AccordRepair.class);
private final SharedContext ctx;
private final ColumnFamilyStore cfs;
private final TimeUUID repairId;
private final Ranges ranges;
private final boolean requireAllEndpoints;
private final List<InetAddressAndPort> endpoints;
private final SyncRemote syncRemote;
private final Epoch minEpoch = ClusterMetadata.current().epoch;
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, boolean requireAllEndpoints)
{
this.ctx = ctx;
this.cfs = cfs;
this.repairId = repairId;
this.requireAllEndpoints = requireAllEndpoints;
this.endpoints = endpoints;
this.syncRemote = requireAllEndpoints ? All : Quorum;
this.ranges = AccordTopology.toAccordRanges(keyspace, ranges);
}
@ -151,9 +141,6 @@ public class AccordRepair
private Pair<List<accord.primitives.Range>, Long> repairRange(TokenRange range) throws Throwable
{
List<accord.primitives.Range> repairedRanges = new ArrayList<>();
List<Node.Id> ids = endpoints == null ? null : endpoints.stream().map(AccordService.instance().configService()::mappedId).collect(Collectors.toList());
DurabilityService.SyncRemote syncRemote = requireAllEndpoints ? All : Quorum;
if (shouldAbort != null)
throw shouldAbort;
@ -177,7 +164,7 @@ public class AccordRepair
long timeoutNanos = getAccordRepairTimeoutNanos();
long maxHlc = AccordService.getBlocking(service.maxConflict(ranges).flatMap(conflict -> {
Timestamp conflictMax = mergeMax(conflict, minForEpoch(this.minEpoch.getEpoch()));
return service.sync("[repairId #" + repairId + ']', conflictMax, Ranges.of(range), ids, NoLocal, syncRemote, timeoutNanos, NANOSECONDS).map(ignored -> conflictMax.hlc());
return service.sync("[repairId #" + repairId + ']', conflictMax, Ranges.of(range), null, NoLocal, syncRemote, timeoutNanos, NANOSECONDS).map(ignored -> conflictMax.hlc());
}), ranges, bookkeeping, start, start + timeoutNanos);
waiting = null;

3
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
View File

@ -913,7 +913,8 @@ public class CommandSerializers
public static final UnversionedSerializer<SaveStatus> saveStatus = EncodeAsVInt32.of(SaveStatus.class);
public static final UnversionedSerializer<Status> status = EncodeAsVInt32.of(Status.class);
public static final UnversionedSerializer<Durability> durability = EncodeAsVInt32.of(Durability.class);
public static final UnversionedSerializer<Durability> durability = EncodeAsVInt32.withoutNulls(Durability::encoded, Durability::forEncoded);
public static final UnversionedSerializer<Durability.HasOutcome> outcomeDurability = EncodeAsVInt32.of(Durability.HasOutcome.class);
public static final IVersionedSerializer<Writes> writes = new IVersionedSerializer<>()
{

17
src/java/org/apache/cassandra/service/accord/serializers/CommandStoreSerializers.java
View File

@ -117,22 +117,22 @@ public class CommandStoreSerializers
@Override
public void serialize(DurableBefore.Entry t, DataOutputPlus out) throws IOException
{
CommandSerializers.txnId.serialize(t.majorityBefore, out);
CommandSerializers.txnId.serialize(t.quorumBefore, out);
CommandSerializers.txnId.serialize(t.universalBefore, out);
}
@Override
public DurableBefore.Entry deserialize(DataInputPlus in) throws IOException
{
TxnId majorityBefore = CommandSerializers.txnId.deserialize(in);
TxnId quorumBefore = CommandSerializers.txnId.deserialize(in);
TxnId universalBefore = CommandSerializers.txnId.deserialize(in);
return new DurableBefore.Entry(majorityBefore, universalBefore);
return new DurableBefore.Entry(quorumBefore, universalBefore);
}
@Override
public long serializedSize(DurableBefore.Entry t)
{
return CommandSerializers.txnId.serializedSize(t.majorityBefore)
return CommandSerializers.txnId.serializedSize(t.quorumBefore)
+ CommandSerializers.txnId.serializedSize(t.universalBefore);
}
}), DurableBefore.Entry[]::new, DurableBefore.SerializerSupport::create);
@ -153,8 +153,11 @@ public class CommandStoreSerializers
{
CommandSerializers.txnId.serialize(bound, out);
}
for (int status : b.statuses)
out.writeShort(status);
for (int i = 0 ; i < b.bounds.length ; ++i)
{
out.writeShort(b.status(i * 2));
out.writeShort(b.status(i * 2 + 1));
}
}
@Override
@ -190,7 +193,7 @@ public class CommandStoreSerializers
{
size += CommandSerializers.txnId.serializedSize(bound);
}
size += 2L * b.statuses.length;
size += 2L * 2 * b.bounds.length;
return size;
}
};

27
src/java/org/apache/cassandra/service/accord/serializers/InformSerializers.java
View File

@ -20,8 +20,6 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.api.RoutingKey;
import accord.messages.InformDecided;
import accord.messages.InformDurable;
import accord.primitives.Route;
import accord.primitives.Status;
@ -33,31 +31,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
public class InformSerializers
{
public static final IVersionedSerializer<InformDecided> decided = new IVersionedSerializer<>()
{
@Override
public void serialize(InformDecided t, DataOutputPlus out, Version version) throws IOException
{
CommandSerializers.txnId.serialize(t.txnId, out);
KeySerializers.routingKey.serialize(t.homeKey, out);
}
@Override
public InformDecided deserialize(DataInputPlus in, Version version) throws IOException
{
TxnId txnId = CommandSerializers.txnId.deserialize(in);
RoutingKey homeKey = KeySerializers.routingKey.deserialize(in);
return new InformDecided(txnId, homeKey);
}
@Override
public long serializedSize(InformDecided t, Version version)
{
return CommandSerializers.txnId.serializedSize(t.txnId)
+ KeySerializers.routingKey.serializedSize(t.homeKey);
}
};
public static final IVersionedSerializer<InformDurable> durable = new TxnRequestSerializer<>()
{
@Override

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

@ -39,21 +39,21 @@ public class SetDurableSerializers
public void serialize(SetShardDurable msg, DataOutputPlus out) throws IOException
{
syncPoint.serialize(msg.exclusiveSyncPoint, out);
CommandSerializers.durability.serialize(msg.durability, out);
CommandSerializers.outcomeDurability.serialize(msg.durability, out);
}
@Override
public SetShardDurable deserialize(DataInputPlus in) throws IOException
{
return new SetShardDurable(syncPoint.deserialize(in),
CommandSerializers.durability.deserialize(in));
CommandSerializers.outcomeDurability.deserialize(in));
}
@Override
public long serializedSize(SetShardDurable msg)
{
return syncPoint.serializedSize(msg.exclusiveSyncPoint)
+ CommandSerializers.durability.serializedSize(msg.durability);
+ CommandSerializers.outcomeDurability.serializedSize(msg.durability);
}
};

12
src/java/org/apache/cassandra/service/accord/serializers/TopologySerializers.java
View File

@ -146,6 +146,7 @@ public class TopologySerializers
{
out.writeLong(topology.epoch());
CollectionSerializers.serializeList(topology.shards(), out, shard);
CollectionSerializers.serializeCollection(topology.removedIds(), out, TopologySerializers.nodeId);
CollectionSerializers.serializeCollection(topology.staleIds(), out, TopologySerializers.nodeId);
}
@ -154,8 +155,9 @@ public class TopologySerializers
{
long epoch = in.readLong();
Shard[] shards = ArraySerializers.deserializeArray(in, shard, Shard[]::new);
SortedArrayList<Node.Id> removedIds = CollectionSerializers.deserializeSortedArrayList(in, TopologySerializers.nodeId, Node.Id[]::new);
SortedArrayList<Node.Id> staleIds = CollectionSerializers.deserializeSortedArrayList(in, TopologySerializers.nodeId, Node.Id[]::new);
return new Topology(epoch, staleIds, shards);
return new Topology(epoch, removedIds, staleIds, shards);
}
@Override
@ -164,6 +166,7 @@ public class TopologySerializers
long size = 0;
size += TypeSizes.LONG_SIZE; // epoch
size += CollectionSerializers.serializedListSize(topology.shards(), shard);
size += CollectionSerializers.serializedCollectionSize(topology.removedIds(), TopologySerializers.nodeId);
size += CollectionSerializers.serializedCollectionSize(topology.staleIds(), TopologySerializers.nodeId);
return size;
}
@ -175,6 +178,7 @@ public class TopologySerializers
public void serialize(Topology topology, DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt(topology.epoch());
CollectionSerializers.serializeList(topology.removedIds(), out, TopologySerializers.nodeId);
CollectionSerializers.serializeList(topology.staleIds(), out, TopologySerializers.nodeId);
List<Shard> shards = topology.shards();
@ -228,6 +232,7 @@ public class TopologySerializers
public long serializedSize(Topology topology)
{
long size = TypeSizes.sizeofUnsignedVInt(topology.epoch());
size += CollectionSerializers.serializedListSize(topology.removedIds(), TopologySerializers.nodeId);
size += CollectionSerializers.serializedListSize(topology.staleIds(), TopologySerializers.nodeId);
List<Shard> shards = topology.shards();
@ -279,7 +284,8 @@ public class TopologySerializers
public Topology deserialize(DataInputPlus in) throws IOException
{
long epoch = in.readUnsignedVInt();
SortedArrays.SortedArrayList<Node.Id> staleNodes = SortedArrays.SortedArrayList.copySorted(CollectionSerializers.deserializeList(in, TopologySerializers.nodeId), Node.Id[]::new);
SortedArrays.SortedArrayList<Node.Id> removedIds = SortedArrays.SortedArrayList.copySorted(CollectionSerializers.deserializeList(in, TopologySerializers.nodeId), Node.Id[]::new);
SortedArrays.SortedArrayList<Node.Id> staleIds = SortedArrays.SortedArrayList.copySorted(CollectionSerializers.deserializeList(in, TopologySerializers.nodeId), Node.Id[]::new);
List<TokenRange> ranges = CollectionSerializers.deserializeList(in, TokenRange.noTableSerializer);
@ -300,7 +306,7 @@ public class TopologySerializers
int flags = in.readUnsignedVInt32();
shards[i] = Shard.SerializerSupport.create(range, nodes, fromSimpleBitSet(nodes, notInFastPath, Node.Id[]::new), fromSimpleBitSet(nodes, joining, Node.Id[]::new), new TinyEnumSet<>(flags));
}
return new Topology(epoch, staleNodes, shards);
return new Topology(epoch, removedIds, staleIds, shards);
}
};
}

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

@ -193,7 +193,14 @@ public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResul
else
{
if (result == null)
{
result = new TxnData();
for (Map.Entry<Integer, TxnDataValue> e2 : entrySet())
{
if (e2.getKey() == e.getKey()) break;
result.put(e2.getKey(), e2.getValue());
}
}
if (newValue != null)
result.put(e.getKey(), newValue);
}

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

@ -62,7 +62,7 @@ public class TxnDataKeyValue extends FilteredPartition implements TxnDataValue
@Override
public TxnDataValue without(Ranges ranges)
{
return ranges.contains(new TokenKey(metadata().id, partitionKey().getToken())) ? this : null;
return ranges.contains(new TokenKey(metadata().id, partitionKey().getToken())) ? null : this;
}
@Override

14
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -38,7 +38,7 @@ import accord.messages.TxnRequest;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Status.Durability.HasOutcome;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.async.AsyncChains;
@ -111,8 +111,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
private static final String QUERY_REDUNDANT_BEFORE =
String.format("SELECT * FROM %s.%s", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ =
String.format("SELECT * FROM %s.%s WHERE majority_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ =
String.format("SELECT * FROM %s.%s WHERE quorum_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ =
String.format("SELECT * FROM %s.%s WHERE shard_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
@ -199,13 +199,13 @@ public class AccordDebugKeyspaceTest extends CQLTester
Ranges ranges1 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(1)), new TokenKey(tableId, new LongToken(100))));
Ranges ranges2 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(100)), new TokenKey(tableId, new LongToken(200))));
AsyncChains.getBlocking(accord.node().commandStores().forEach((PreLoadContext.Empty)() -> "Test", safeStore -> {
safeStore.commandStore().markShardDurable(safeStore, syncId1, ranges1, Status.Durability.Universal);
safeStore.commandStore().markShardDurable(safeStore, syncId2, ranges2, Status.Durability.Majority);
safeStore.commandStore().markShardDurable(safeStore, syncId1, ranges1, HasOutcome.Universal);
safeStore.commandStore().markShardDurable(safeStore, syncId2, ranges2, HasOutcome.Quorum);
}));
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE).size()).isGreaterThan(0);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ, syncId1.toString()).size()).isEqualTo(2);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ, syncId2.toString()).size()).isEqualTo(1);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ, syncId1.toString()).size()).isEqualTo(2);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ, syncId2.toString()).size()).isEqualTo(1);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ, syncId1.toString()).size()).isEqualTo(1);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ, syncId2.toString()).size()).isEqualTo(0);
}

4
test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
View File

@ -70,7 +70,7 @@ import org.apache.cassandra.service.accord.txn.TxnUpdate;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.Pair;
import static accord.primitives.Status.Durability.Majority;
import static accord.primitives.Status.Durability.AllQuorums;
import static com.google.common.collect.Iterables.getOnlyElement;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
import static org.apache.cassandra.service.accord.AccordTestUtils.Commands.preaccepted;
@ -135,7 +135,7 @@ public class AccordCommandStoreTest
Command.WaitingOn waitingOn = new Command.WaitingOn(dependencies.keyDeps.keys(), dependencies.rangeDeps, new ImmutableBitSet(waitingOnApply), new ImmutableBitSet(2));
Pair<Writes, Result> result = AsyncChains.getBlocking(AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt));
Command expected = Command.Executed.executed(txnId, SaveStatus.Applied, Majority, StoreParticipants.all(route),
Command expected = Command.Executed.executed(txnId, SaveStatus.Applied, AllQuorums, StoreParticipants.all(route),
promised, executeAt, txn, dependencies, accepted,
waitingOn, result.left, ResultSerializers.APPLIED);
AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null));

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

@ -120,7 +120,7 @@ import static accord.api.ProtocolModifiers.Toggles.setTransitiveDependenciesAreV
import static accord.local.cfk.CommandsForKey.NO_BOUNDS_INFO;
import static accord.primitives.Known.KnownExecuteAt.ExecuteAtErased;
import static accord.primitives.Known.KnownExecuteAt.ExecuteAtUnknown;
import static accord.primitives.Status.Durability.Majority;
import static accord.primitives.Status.Durability.AllQuorums;
import static accord.primitives.Status.Durability.NotDurable;
import static accord.utils.Property.qt;
import static accord.utils.SortedArrays.Search.FAST;
@ -182,7 +182,7 @@ public class CommandsForKeySerializerTest
builder.partialTxn(txn);
builder.setParticipants(StoreParticipants.all(txn.keys().toRoute(txn.keys().get(0).someIntersectingRoutingKey(null))));
builder.durability(isDurable ? Majority : NotDurable);
builder.durability(isDurable ? AllQuorums : NotDurable);
if (saveStatus.known.deps().hasPreAcceptedOrProposedOrDecidedDeps())
{
try (KeyDeps.Builder keyBuilder = KeyDeps.builder();)
@ -196,7 +196,7 @@ public class CommandsForKeySerializerTest
builder.executeAt(executeAt);
builder.promised(ballot);
builder.acceptedOrCommitted(ballot);
builder.durability(isDurable ? Majority : NotDurable);
builder.durability(isDurable ? AllQuorums : NotDurable);
if (saveStatus.compareTo(SaveStatus.Stable) >= 0 && !saveStatus.hasBeen(Status.Truncated))
builder.waitingOn(Command.WaitingOn.empty(txnId.domain()));
@ -424,7 +424,7 @@ public class CommandsForKeySerializerTest
@Test
public void serde()
{
testOne(-4567266914751633833L);
testOne(7082228630293368049L);
Random random = new Random();
for (int i = 0 ; i < 10000 ; ++i)
{
@ -610,7 +610,7 @@ public class CommandsForKeySerializerTest
else unmanaged = CommandsForKey.NO_PENDING_UNMANAGED;
long maxUniqueHlc = rs.nextLong(0, Long.MAX_VALUE);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, maxUniqueHlc, unmanaged, TxnId.NONE, NO_BOUNDS_INFO);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, maxUniqueHlc, unmanaged, TxnId.NONE, NO_BOUNDS_INFO, true);
ByteBuffer buffer = Serialize.toBytesWithoutKey(expected);
CommandsForKey roundTrip = Serialize.fromBytes(pk, buffer);
@ -627,7 +627,7 @@ public class CommandsForKeySerializerTest
TxnId txnId = TxnId.fromValues(11,34052499,2,1);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk,
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_WITHOUT_DEPS, true, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) },
0, CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, NO_BOUNDS_INFO);
0, CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, NO_BOUNDS_INFO, true);
ByteBuffer buffer = Serialize.toBytesWithoutKey(expected);
CommandsForKey roundTrip = Serialize.fromBytes(pk, buffer);