From b4f6c7c6173e946c8a7eee0f35061acf8a0b1d6d Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Fri, 5 Sep 2025 22:01:35 +0100 Subject: [PATCH] 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 --- modules/accord | 2 +- .../db/compaction/CompactionIterator.java | 42 ++++++--- .../db/virtual/AccordDebugKeyspace.java | 22 ++--- .../service/accord/AccordCacheEntry.java | 21 ++++- .../service/accord/AccordCommandStore.java | 8 -- .../service/accord/AccordJournal.java | 20 ++-- .../accord/AccordSafeCommandStore.java | 8 ++ .../service/accord/AccordService.java | 31 +++++-- .../service/accord/api/AccordAgent.java | 9 -- .../accord/api/AccordViolationHandler.java | 19 +++- .../accord/journal/AccordTopologyUpdate.java | 91 +++++++++++-------- .../serializers/CommandSerializers.java | 23 +++++ .../accord/serializers/DepsSerializers.java | 39 ++++++++ .../accord/serializers/SerializePacked.java | 6 +- .../journal/AccordTopologyUpdateTest.java | 6 +- .../CommandsForKeySerializerTest.java | 1 - 16 files changed, 242 insertions(+), 106 deletions(-) diff --git a/modules/accord b/modules/accord index c3a62d77ba..9c7f856c3e 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c3a62d77ba332f4a01220709c040af86dd3acf6a +Subproject commit 9c7f856c3e1f10c6f985495702edb844cfee2b80 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 79344bdc67..8facfe30da 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -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.TopologyImage lastChangedTopology; + TopologyImage lastImage; + boolean hasWritten; final long minEpoch; + TopologyCompactor(FlyweightSerializer 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); } } diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 423a7b3f6c..e69891cc6b 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -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())) diff --git a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java index 13cc1a4bac..9a39874083 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java @@ -335,6 +335,21 @@ public class AccordCacheEntry extends IntrusiveLinkedListNode this.onSuccess = new ArrayList<>(); this.onSuccess.add(onSuccess); } + + static void notify(List 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 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 extends IntrusiveLinkedListNode boolean saved(Object identity, Throwable fail) { + if (identity instanceof UniqueSave) + UniqueSave.notify(((UniqueSave) identity).onSuccess); + if (!is(SAVING)) return false; diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 20c86281dd..5a73e8a79f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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. diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 68a3b6c138..045acea8df 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -389,12 +389,17 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public TopologyUpdate next() { Journal.KeyRefs 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) diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 5cdff4f11f..f3ecdc43fd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -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 V getBlocking(AsyncChain async, Seekables keysOrRanges, RequestBookkeeping bookkeeping, long startedAt, long deadline) { return getBlocking(async, keysOrRanges, bookkeeping, startedAt, deadline, false); } + public static V getBlocking(AsyncChain async) + { + return asPromise(async).syncUninterruptibly().getNow(); + } + + public static Promise asPromise(AsyncChain async) + { + AsyncPromise 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 epochReady(Epoch epoch) { - AsyncPromise promise = new AsyncPromise<>(); - AsyncChain 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 diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 092801fcba..9204770a3d 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -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) { diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordViolationHandler.java b/src/java/org/apache/cassandra/service/accord/api/AccordViolationHandler.java index 44687e00f5..67ec0111a8 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordViolationHandler.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordViolationHandler.java @@ -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)); } } diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java index c32ca52d29..a01ddb9469 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java @@ -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 + 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)); } } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java index 3348fc1f2c..e9b42c338f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -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); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java index c77ce05993..f3b4d1107e 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java @@ -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 diff --git a/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java b/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java index 8321e147fd..b7262c1265 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java @@ -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); } } diff --git a/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java index c0c790ba47..8443a09741 100644 --- a/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java +++ b/test/unit/org/apache/cassandra/service/accord/journal/AccordTopologyUpdateTest.java @@ -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); } }; diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index fb62e7fd59..42fec34135 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -657,7 +657,6 @@ public class CommandsForKeySerializerTest @Override public AsyncChain build(PreLoadContext context, Function apply) { throw new UnsupportedOperationException(); } @Override public void shutdown() { } @Override public AsyncChain build(Callable 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(); }