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

Topology compaction omits image records if minEpoch is ahead of last image 

Also Fix:
 - RegisteredCallback should remove itself from callback map when cancelled
 - Do not throw CancellationException when processing requests that have been aborted, as may be caused by a successful meaningful reply that can be overridden
 - system_accord_debug.{executors,coordinations} fail with ClassCastException
 - CommandStore.updateMinHlc eats up CPU as called much too often
 - AccordCache not notifying flushed on shutdown
Also Improve:
 - Support skipping Deps
 - Violation information reported
 - Sort CommandStore shards by id

patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-20896
This commit is contained in:
Benedict Elliott Smith 2025-09-05 22:01:35 +01:00
parent 965a39166c
commit b4f6c7c617
16 changed files with 242 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit c3a62d77ba332f4a01220709c040af86dd3acf6a
Subproject commit 9c7f856c3e1f10c6f985495702edb844cfee2b80

42
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -95,6 +95,7 @@ import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfos;
import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate;
import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.TopologyImage;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.service.paxos.PaxosRepairHistory;
import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
@ -118,6 +119,8 @@ import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
import static org.apache.cassandra.service.accord.AccordKeyspace.CFKAccessor;
import static org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns.getJournalKey;
import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.Kind.Image;
import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.Kind.Repeat;
/**
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
@ -868,7 +871,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
public AccordJournalPurger(AccordCompactionInfos compactionInfos, Version version, ColumnFamilyStore cfs)
{
this.userVersion = version;
this.infos = compactionInfos;
this.recordColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("record", false));
this.versionColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("user_version", false));
@ -943,8 +945,10 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
static class TopologyCompactor extends AccordMergingCompactor<AccordTopologyUpdate.Accumulator>
{
AccordTopologyUpdate.TopologyImage lastChangedTopology;
TopologyImage lastImage;
boolean hasWritten;
final long minEpoch;
TopologyCompactor(FlyweightSerializer<Object, AccordTopologyUpdate.Accumulator> serializer, Version userVersion, long minEpoch)
{
super(serializer, userVersion);
@ -960,21 +964,33 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
@Override
UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException
{
AccordTopologyUpdate.TopologyImage current = builder.get();
Invariants.require(lastImage != null || !hasWritten);
TopologyImage read = builder.read();
if (lastChangedTopology != null && current.getUpdate() != null && lastChangedTopology.getUpdate().isEquivalent(current.getUpdate()))
builder.update(current.asNoOp());
if (builder.get().kind() != AccordTopologyUpdate.Kind.NoOp)
if (read.epoch() < minEpoch)
{
lastChangedTopology = builder.get();
Invariants.nonNull(lastChangedTopology.getUpdate());
if (read.kind() == Image)
lastImage = read;
return null;
}
if (builder.get().epoch() >= minEpoch)
return super.result(journalKey, partitionKey);
else
return null;
TopologyImage write = read;
if (read.kind() == Repeat && !hasWritten)
{
Invariants.require(lastImage != null);
write = new TopologyImage(read.epoch(), Image, lastImage.getUpdate());
}
else if (hasWritten && read.kind() == Repeat && lastImage.getUpdate().isEquivalent(read.getUpdate()))
{
write = read.asRepeat();
}
if (write.kind() == Image)
lastImage = write;
hasWritten = true;
builder.write(write);
return super.result(journalKey, partitionKey);
}
}

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

@ -219,8 +219,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
" txn_id 'TxnIdUtf8Type',\n" +
" txn_id_additional 'TxnIdUtf8Type',\n" +
" keys text,\n" +
" keysLoad text,\n" +
" keysLoadFor text,\n" +
" keys_loading text,\n" +
" keys_loading_for text,\n" +
" PRIMARY KEY (executor_id, status, position, unique_position)" +
')', UTF8Type.instance));
}
@ -242,14 +242,14 @@ public class AccordDebugKeyspace extends VirtualKeyspace
else uniquePos = 0;
prev = info;
PreLoadContext preLoadContext = info.preLoadContext();
ds.row(executorId, info.status(), info.position(), uniquePos)
ds.row(executorId, Objects.toString(info.status()), info.position(), uniquePos)
.column("description", info.describe())
.column("command_store_id", info.commandStoreId())
.column("txn_id", preLoadContext == null ? null : preLoadContext.primaryTxnId())
.column("txn_id_additional", preLoadContext == null ? null : preLoadContext.additionalTxnId())
.column("keys", preLoadContext == null ? null : preLoadContext.keys())
.column("keysLoad", preLoadContext == null ? null : preLoadContext.loadKeys())
.column("keysLoadFor", preLoadContext == null ? null : preLoadContext.loadKeysFor())
.column("txn_id", preLoadContext == null ? null : toStringOrNull(preLoadContext.primaryTxnId()))
.column("txn_id_additional", preLoadContext == null ? null : toStringOrNull(preLoadContext.additionalTxnId()))
.column("keys", preLoadContext == null ? null : toStringOrNull(preLoadContext.keys()))
.column("keys_loading", preLoadContext == null ? null : toStringOrNull(preLoadContext.loadKeys()))
.column("keys_loading_for", preLoadContext == null ? null : toStringOrNull(preLoadContext.loadKeysFor()))
;
}
}
@ -265,9 +265,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace
super(parse(VIRTUAL_ACCORD_DEBUG, COORDINATIONS,
"Accord Coordination State",
"CREATE TABLE %s (\n" +
" txn_id int,\n" +
" txn_id 'TxnIdUtf8Type',\n" +
" kind text,\n" +
" coordination_id int,\n" +
" coordination_id bigint,\n" +
" description text,\n" +
" nodes text,\n" +
" nodes_inflight text,\n" +
@ -286,7 +286,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
SimpleDataSet ds = new SimpleDataSet(metadata());
for (Coordination c : coordinations)
{
ds.row(c.txnId(), c.kind().toString(), c.coordinationId())
ds.row(toStringOrNull(c.txnId()), c.kind().toString(), c.coordinationId())
.column("nodes", toStringOrNull(c.nodes()))
.column("nodes_inflight", toStringOrNull(c.inflight()))
.column("nodes_contacted", toStringOrNull(c.contacted()))

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

@ -335,6 +335,21 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
this.onSuccess = new ArrayList<>();
this.onSuccess.add(onSuccess);
}
static void notify(List<Runnable> onSuccess)
{
if (onSuccess != null)
{
onSuccess.forEach(run -> {
try { run.run(); }
catch (Throwable t)
{
Thread thread = Thread.currentThread();
thread.getUncaughtExceptionHandler().uncaughtException(thread, t);
}
});
}
}
}
public interface SaveExecutor
@ -506,8 +521,7 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
setStatus(LOADED);
if (waitingToSave != null)
this.state = state;
if (identity.onSuccess != null)
identity.onSuccess.forEach(Runnable::run);
UniqueSave.notify(identity.onSuccess);
return false;
}
else
@ -521,6 +535,9 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
boolean saved(Object identity, Throwable fail)
{
if (identity instanceof UniqueSave)
UniqueSave.notify(((UniqueSave) identity).onSuccess);
if (!is(SAVING))
return false;

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

@ -74,7 +74,6 @@ import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor
import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.service.paxos.PaxosState;
import org.apache.cassandra.utils.Clock;
import static accord.api.Journal.CommandUpdate;
@ -584,13 +583,6 @@ public class AccordCommandStore extends CommandStore
loadRangesForEpoch(rangesForEpoch);
}
@Override
public void updateRangesForEpoch(SafeCommandStore safeStore)
{
super.updateRangesForEpoch(safeStore);
updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
}
// TODO (expected): handle journal failures, and consider how we handle partial failures.
// Very likely we will not be able to safely or cleanly handle partial failures of this logic, but decide and document.
// TODO (desired): consider merging with PersistentField? This version is cheaper to manage which may be preferable at the CommandStore level.

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

@ -389,12 +389,17 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public TopologyUpdate next()
{
Journal.KeyRefs<JournalKey> ref = iter.next();
System.out.println(ref.key());
Accumulator read = readAll(ref.key());
if (read.accumulated.kind() == Kind.NoOp)
prev = read.accumulated.asImage(Invariants.nonNull(prev.getUpdate()));
else
prev = read.accumulated;
Accumulator reader = readAll(ref.key());
if (reader.read().kind() == Kind.Repeat)
{
if (prev == null)
{
logger.error("Encountered TopologyImage Repeat record for epoch {}, but no prior image record was found", ref.key().id.epoch());
return null;
}
prev = reader.read().asImage(Invariants.nonNull(prev.getUpdate()));
}
else prev = reader.read();
return new TopologyUpdate(prev.getUpdate().commandStores,
prev.getUpdate().global);
@ -411,6 +416,9 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
while (iter.hasNext())
{
TopologyUpdate next = iter.next();
if (next == null)
continue;
Invariants.require(prev == null || next.global.epoch() > prev.global.epoch());
// Due to partial compaction, we can clean up only some of the old epochs, creating gaps. We skip these epochs here.
if (prev != null && next.global.epoch() > prev.global.epoch() + 1)

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

@ -42,6 +42,7 @@ import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
import org.apache.cassandra.service.accord.AccordCommandStore.SafeRedundantBefore;
import org.apache.cassandra.service.paxos.PaxosState;
import static accord.utils.Invariants.illegalState;
@ -169,6 +170,13 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return commandsForKey.get(key);
}
@Override
public void setRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
{
super.setRangesForEpoch(rangesForEpoch);
commandStore.updateMinHlc(PaxosState.ballotTracker().getLowBound().unixMicros() + 1);
}
@Override
public AccordCommandStore commandStore()
{

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

@ -39,6 +39,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.service.accord.api.AccordViolationHandler;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.slf4j.Logger;
@ -132,6 +133,7 @@ import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.Promise;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.api.Journal.TopologyUpdate;
@ -315,7 +317,10 @@ public class AccordService implements IAccordService, Shutdownable
as.node.durability().start();
instance = as;
AccordReplicaMetrics.touch();
AccordViolationHandler.setup();
WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);
return as;
}
@ -596,11 +601,27 @@ public class AccordService implements IAccordService, Shutdownable
async.begin(result);
return result.awaitAndGet();
}
public static <V> V getBlocking(AsyncChain<V> async, Seekables<?, ?> keysOrRanges, RequestBookkeeping bookkeeping, long startedAt, long deadline)
{
return getBlocking(async, keysOrRanges, bookkeeping, startedAt, deadline, false);
}
public static <V> V getBlocking(AsyncChain<V> async)
{
return asPromise(async).syncUninterruptibly().getNow();
}
public static <V> Promise<V> asPromise(AsyncChain<V> async)
{
AsyncPromise<V> promise = new AsyncPromise<>();
async.begin((result, failure) -> {
if (failure == null) promise.trySuccess(result);
else promise.tryFailure(failure);
});
return promise;
}
public static Keys intersecting(Keys keys)
{
if (keys.isEmpty())
@ -974,7 +995,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public void ensureMinHlc(long minHlc)
{
node.updateMinHlc(minHlc >= 0 ? minHlc : 0);
asPromise(node.updateMinHlc(minHlc >= 0 ? minHlc : 0)).syncUninterruptibly();
}
public AccordJournal journal()
@ -985,13 +1006,7 @@ public class AccordService implements IAccordService, Shutdownable
@Override
public Future<Void> epochReady(Epoch epoch)
{
AsyncPromise<Void> promise = new AsyncPromise<>();
AsyncChain<Void> ready = configService.epochReady(epoch.getEpoch());
ready.begin((result, failure) -> {
if (failure == null) promise.trySuccess(result);
else promise.tryFailure(failure);
});
return promise;
return asPromise(configService.epochReady(epoch.getEpoch()));
}
@Override

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

@ -131,15 +131,6 @@ public class AccordAgent implements Agent
self = id;
}
@Override
public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next)
{
// TODO (expected): better reporting
AssertionError error = new AssertionError("Inconsistent execution timestamp detected for txnId " + command.txnId() + ": " + prev + " != " + next);
onUncaughtException(error);
throw error;
}
@Override
public void onFailedBootstrap(int attempts, String phase, Ranges ranges, Runnable retry, Throwable failure)
{

19
src/java/org/apache/cassandra/service/accord/api/AccordViolationHandler.java
View File

@ -18,28 +18,37 @@
package org.apache.cassandra.service.accord.api;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.ViolationHandler;
import accord.local.Command;
import accord.local.SafeCommandStore;
import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import static accord.utils.Invariants.illegalState;
public class AccordViolationHandler implements ViolationHandler
{
private static final Logger logger = LoggerFactory.getLogger(AccordViolationHandler.class);
static
public static void setup()
{
ViolationHandlerHolder.set(AccordViolationHandler::new);
}
@Override
public void onViolation(String message, Participants<?> participants, @Nullable TxnId notWitnessed, @Nullable Timestamp notWitnessedExecuteAt, @Nullable TxnId by, @Nullable Timestamp byEexecuteAt)
public void onTimestampViolation(SafeCommandStore safeStore, Command command, Participants<?> otherParticipants, Route<?> otherRoute, Timestamp otherExecuteAt)
{
logger.error(message);
throw illegalState(ViolationHandler.timestampViolationMessage(safeStore, command, otherParticipants, otherRoute, otherExecuteAt));
}
@Override
public void onDependencyViolation(Participants<?> participants, TxnId notWitnessed, Timestamp notWitnessedExecuteAt, TxnId by, Timestamp byExecuteAt)
{
logger.error(ViolationHandler.dependencyViolationMessage(participants, notWitnessed, notWitnessedExecuteAt, by, byExecuteAt));
}
}

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

@ -44,7 +44,7 @@ public interface AccordTopologyUpdate
Kind kind();
void applyTo(TopologyImage accumulator);
long epoch();
AccordTopologyUpdate asNoOp();
AccordTopologyUpdate asRepeat();
Journal.TopologyUpdate getUpdate();
static AccordTopologyUpdate newTopology(Journal.TopologyUpdate update)
@ -151,13 +151,13 @@ public interface AccordTopologyUpdate
out.writeUnsignedVInt32(t.kind().ordinal());
switch (t.kind())
{
case NewTopology:
case New:
{
TopologyUpdateSerializer.instance.serialize(((NewTopology) t).update, out);
break;
}
case NoOp:
case TopologyImage:
case Repeat:
case Image:
TopologyImage image = (TopologyImage) t;
out.writeBoolean(image.update != null);
if (image.update != null)
@ -183,14 +183,15 @@ public interface AccordTopologyUpdate
switch (kind)
{
case NewTopology:
case New:
return new NewTopology(TopologyUpdateSerializer.instance.deserialize(in));
case NoOp:
case TopologyImage:
TopologyImage image = new TopologyImage(epoch);
case Repeat:
case Image:
Journal.TopologyUpdate update = null;
if (in.readBoolean())
image.update = TopologyUpdateSerializer.instance.deserialize(in);
update = TopologyUpdateSerializer.instance.deserialize(in);
TopologyImage image = new TopologyImage(epoch, kind, update);
byte syncStateByte = in.readByte();
if (syncStateByte != Byte.MAX_VALUE)
image.syncStatus = AccordConfigurationService.SyncStatus.values()[syncStateByte];
@ -211,11 +212,11 @@ public interface AccordTopologyUpdate
switch (t.kind())
{
case NewTopology:
case New:
size += TopologyUpdateSerializer.instance.serializedSize(((NewTopology) t).update);
break;
case TopologyImage:
case NoOp:
case Image:
case Repeat:
TopologyImage image = (TopologyImage) t;
size += TypeSizes.BOOL_SIZE;
if (image.update != null)
@ -235,49 +236,52 @@ public interface AccordTopologyUpdate
enum Kind
{
// New Topology, written to journal when the node first learned about it
NewTopology,
New,
// Used when accumulating state during compaction or replay
TopologyImage,
Image,
// Effectively unchanged topology
// During compaction, we can write a no-op if we know that from Accord's perspective topology has not changed
// (see CompactionIterator$TopologyCompactor). During replay/deserialization, we collect last known changed
// epoch, and reconstruct its topology.
NoOp
Repeat
}
class TopologyImage implements AccordTopologyUpdate
{
private final long epoch;
private final Kind kind;
private Journal.TopologyUpdate update;
private AccordConfigurationService.SyncStatus syncStatus = null;
private Ranges closed = Ranges.EMPTY;
private Ranges retired = Ranges.EMPTY;
private final long epoch;
public TopologyImage(long epoch)
public TopologyImage(long epoch, Kind kind)
{
this.epoch = epoch;
this.kind = Invariants.requireArgument(kind, kind == Kind.Repeat);
}
public TopologyImage(long epoch, Journal.TopologyUpdate update)
public TopologyImage(long epoch, Kind kind, Journal.TopologyUpdate update)
{
this.epoch = epoch;
this.update = update;
this.kind = kind;
this.update = Invariants.requireArgument(update, update != null || kind == Kind.Repeat);
}
public TopologyImage asImage(Journal.TopologyUpdate update)
{
TopologyImage image = new TopologyImage(epoch, update.cloneWithEquivalentEpoch(epoch));
TopologyImage image = new TopologyImage(epoch, Kind.Image, update.cloneWithEquivalentEpoch(epoch));
image.closed = closed;
image.retired = retired;
return image;
}
public TopologyImage asNoOp()
public TopologyImage asRepeat()
{
TopologyImage image = new TopologyImage(epoch);
TopologyImage image = new TopologyImage(epoch, Kind.Repeat, update);
image.closed = closed;
image.retired = retired;
return image;
@ -298,14 +302,14 @@ public interface AccordTopologyUpdate
@Override
public Kind kind()
{
return update == null ? Kind.NoOp : Kind.TopologyImage;
return kind;
}
@Override
public void applyTo(TopologyImage accumulator)
{
Invariants.require(accumulator.epoch == epoch, "Expected %d but got %d", epoch, accumulator.epoch);
if (kind() == Kind.NoOp)
if (kind() == Kind.Repeat)
{
accumulator.update = null;
return;
@ -362,7 +366,7 @@ public interface AccordTopologyUpdate
@Override
public Kind kind()
{
return Kind.NewTopology;
return Kind.New;
}
@Override
@ -374,9 +378,9 @@ public interface AccordTopologyUpdate
}
@Override
public AccordTopologyUpdate asNoOp()
public AccordTopologyUpdate asRepeat()
{
return new TopologyImage(epoch);
return new TopologyImage(epoch, Kind.Repeat, update);
}
@Override
@ -395,24 +399,39 @@ public interface AccordTopologyUpdate
}
}
class Accumulator extends AccordJournalValueSerializers.Accumulator<TopologyImage, AccordTopologyUpdate>
class Accumulator implements AccordJournalValueSerializers.FlyweightImage
{
TopologyImage read, write;
public Accumulator()
{
super(null);
}
@Override
public void reset(JournalKey key)
{
accumulated = new TopologyImage(key.id.epoch());
read = write = null;
}
@Override
protected TopologyImage accumulate(TopologyImage acc, AccordTopologyUpdate update)
public TopologyImage read()
{
update.applyTo(acc);
return acc;
return read;
}
public void read(AccordTopologyUpdate update)
{
Invariants.require(read == null);
if (Objects.requireNonNull(update.kind()) == Kind.New)
read = new TopologyImage(update.epoch(), Kind.Image, update.getUpdate());
else
read = (TopologyImage) update;
write = read;
}
public void write(TopologyImage image)
{
Invariants.require(write == read);
this.write = image;
}
}
@ -435,13 +454,13 @@ public interface AccordTopologyUpdate
@Override
public void reserialize(JournalKey key, Accumulator from, DataOutputPlus out, Version version) throws IOException
{
serialize(key, from.get(), out, version);
serialize(key, from.write, out, version);
}
@Override
public void deserialize(JournalKey key, Accumulator into, DataInputPlus in, Version version) throws IOException
{
into.update(Serializer.instance.deserialize(in));
into.read(Serializer.instance.deserialize(in));
}
}
}

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

@ -761,6 +761,29 @@ public class CommandSerializers
return ts;
}
public int skipArray(DataInputPlus in) throws IOException
{
int length = in.readUnsignedVInt32();
if (length == 0)
return 0;
// we could pack these a bit more tightly if we wanted to
in.readUnsignedVInt();
in.readUnsignedVInt();
in.readUnsignedVInt32();
in.readUnsignedVInt32();
int epochBits = in.readByte();
int hlcBits = in.readByte();
int flagBits = in.readByte();
int nodeBits = in.readByte();
in.skipBytesFully(SerializePacked.serializedPackedSize(length, mask(epochBits))
+ SerializePacked.serializedPackedSize(length, mask(hlcBits))
+ SerializePacked.serializedPackedSize(length, mask(flagBits))
+ SerializePacked.serializedPackedSize(length, mask(nodeBits)));
return length;
}
private static long mask(int bits)
{
return bits == 0 ? 0 : -1L >>> (64 - bits);

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

@ -42,9 +42,11 @@ import static accord.primitives.KeyDeps.SerializerSupport.txnIdsToKeys;
import static accord.primitives.RangeDeps.SerializerSupport.ranges;
import static accord.primitives.RangeDeps.SerializerSupport.rangesToTxnIds;
import static accord.primitives.RangeDeps.SerializerSupport.txnIdsToRanges;
import static org.apache.cassandra.service.accord.serializers.KeySerializers.keys;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.deserializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedIntsSize;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedSize;
public class DepsSerializers
{
@ -204,6 +206,36 @@ public class DepsSerializers
return xtoy;
}
private static void skipPackedXtoY(int xCount, int yCount, DataInputPlus in) throws IOException
{
int length = in.readUnsignedVInt32();
if (!((xCount <= 1 || yCount <= 1) && (length == (xCount == 1 ? 1 + yCount : 2 * xCount) || xCount == 0 || yCount == 0)))
{
in.skipBytesFully(serializedPackedSize(xCount, length)
+ serializedPackedSize(length - xCount, yCount - 1));
}
}
@Override
public void skip(DataInputPlus in) throws IOException
{
int flags = in.readUnsignedVInt32();
{
int keyCount = KeySerializers.routingKeys.countAndSkip(in);
int txnIdCount = CommandSerializers.txnId.skipArray(in);
if (0 != (flags & KEYS_BY_TXNID)) skipPackedXtoY(txnIdCount, keyCount, in);
else skipPackedXtoY(keyCount, txnIdCount, in);
}
{
int rangeCount = KeySerializers.rangeArray.countAndSkip(in);
int txnIdCount = CommandSerializers.txnId.skipArray(in);
if (0 != (flags & RANGES_BY_TXNID)) skipPackedXtoY(txnIdCount, rangeCount, in);
else skipPackedXtoY(rangeCount, txnIdCount, in);
}
}
@Override
public long serializedSize(D deps)
{
@ -270,6 +302,13 @@ public class DepsSerializers
return super.serializedSize(partialDeps)
+ KeySerializers.participants.serializedSize(partialDeps.covering);
}
@Override
public void skip(DataInputPlus in) throws IOException
{
super.skip(in);
KeySerializers.participants.skip(in);
}
}
static class DepsSerializer extends AbstractDepsSerializer<Deps>

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

@ -133,13 +133,13 @@ public class SerializePacked
}
}
public static long serializedPackedSize(int count, long max)
public static int serializedPackedSize(int count, long max)
{
return serializedPackedBitsSize(count, BitUtils.numberOfBitsToRepresent(max));
}
public static long serializedPackedBitsSize(int count, int bitsPerEntry)
public static int serializedPackedBitsSize(int count, int bitsPerEntry)
{
return ((long)bitsPerEntry * count + 7)/8;
return (int) (((long)bitsPerEntry * count + 7) / 8);
}
}

6
test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java
View File

@ -140,9 +140,9 @@ public class AccordTopologyUpdateTest
AccordTopologyUpdate.Kind kind = kindGen.next(rs);
switch (kind)
{
case NewTopology: return new AccordTopologyUpdate.NewTopology(topologyUpdateGen.next(rs));
case TopologyImage: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs), topologyUpdateGen.next(rs));
case NoOp: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs));
case New: return new AccordTopologyUpdate.NewTopology(topologyUpdateGen.next(rs));
case Image: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs), AccordTopologyUpdate.Kind.Image, topologyUpdateGen.next(rs));
case Repeat: return new AccordTopologyUpdate.TopologyImage(epochGen.nextLong(rs), AccordTopologyUpdate.Kind.Repeat);
default: throw new AssertionError("Unknown kind: " + kind);
}
};

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

@ -657,7 +657,6 @@ public class CommandsForKeySerializerTest
@Override public <T> AsyncChain<T> build(PreLoadContext context, Function<? super SafeCommandStore, T> apply) { throw new UnsupportedOperationException(); }
@Override public void shutdown() { }
@Override public <T> AsyncChain<T> build(Callable<T> task) { throw new UnsupportedOperationException(); }
@Override public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next) { throw new UnsupportedOperationException(); }
@Override public void onFailedBootstrap(int attempts, String phase, Ranges ranges, Runnable retry, Throwable failure) { throw new UnsupportedOperationException(); }
@Override public void onStale(Timestamp staleSince, Ranges ranges) { throw new UnsupportedOperationException(); }
@Override public void onUncaughtException(Throwable t) { throw new UnsupportedOperationException(); }