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

Implement field saving/loading in AccordJournal 

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20114
This commit is contained in:
Alex Petrov 2024-11-26 15:26:54 +01:00 committed by David Capwell
parent c114fc46d2
commit 211249b8a9
16 changed files with 64 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 1401e93490de57c65a12020d85830fbba8ef8c26
Subproject commit 5785d86ddad62932e21eb427deede09c77352213

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

@ -611,9 +611,9 @@ public class AccordCommandStore extends CommandStore
unsafeSetSafeToRead(safeToRead);
}
void loadRangesForEpoch(CommandStores.RangesForEpoch.Snapshot rangesForEpoch)
void loadRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
{
if (rangesForEpoch != null)
unsafeSetRangesForEpoch(new CommandStores.RangesForEpoch(rangesForEpoch.epochs, rangesForEpoch.ranges, this));
unsafeSetRangesForEpoch(rangesForEpoch);
}
}

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

@ -241,9 +241,9 @@ public class AccordJournal implements IJournal, Shutdownable
}
@Override
public CommandStores.RangesForEpoch.Snapshot loadRangesForEpoch(int store)
public CommandStores.RangesForEpoch loadRangesForEpoch(int store)
{
IdentityAccumulator<RangesForEpoch.Snapshot> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store));
IdentityAccumulator<RangesForEpoch> accumulator = readAll(new JournalKey(TxnId.NONE, JournalKey.Type.RANGES_FOR_EPOCH, store));
return accumulator.get();
}

45
src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
View File

@ -28,6 +28,7 @@ import accord.local.RedundantBefore;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
@ -280,44 +281,50 @@ public class AccordJournalValueSerializers
}
public static class RangesForEpochSerializer
implements FlyweightSerializer<RangesForEpoch.Snapshot, IdentityAccumulator<RangesForEpoch.Snapshot>>
implements FlyweightSerializer<RangesForEpoch, IdentityAccumulator<RangesForEpoch>>
{
public IdentityAccumulator<RangesForEpoch.Snapshot> mergerFor(JournalKey key)
public IdentityAccumulator<RangesForEpoch> mergerFor(JournalKey key)
{
return new IdentityAccumulator<>(null);
}
@Override
public void serialize(JournalKey key, RangesForEpoch.Snapshot from, DataOutputPlus out, int userVersion) throws IOException
public void serialize(JournalKey key, RangesForEpoch from, DataOutputPlus out, int userVersion) throws IOException
{
out.writeUnsignedVInt32(from.ranges.length);
for (Ranges ranges : from.ranges)
KeySerializers.ranges.serialize(ranges, out, messagingVersion);
out.writeUnsignedVInt32(from.epochs.length);
for (long epoch : from.epochs)
out.writeLong(epoch);
out.writeUnsignedVInt32(from.size());
from.forEach((epoch, ranges) -> {
try
{
KeySerializers.ranges.serialize(ranges, out, messagingVersion);
out.writeLong(epoch);
}
catch (Throwable t)
{
throw new IllegalStateException("Serialization error", t);
}
});
}
@Override
public void reserialize(JournalKey key, IdentityAccumulator<RangesForEpoch.Snapshot> from, DataOutputPlus out, int userVersion) throws IOException
public void reserialize(JournalKey key, IdentityAccumulator<RangesForEpoch> from, DataOutputPlus out, int userVersion) throws IOException
{
serialize(key, from.get(), out, messagingVersion);
}
@Override
public void deserialize(JournalKey key, IdentityAccumulator<RangesForEpoch.Snapshot> into, DataInputPlus in, int userVersion) throws IOException
public void deserialize(JournalKey key, IdentityAccumulator<RangesForEpoch> into, DataInputPlus in, int userVersion) throws IOException
{
Ranges[] ranges = new Ranges[in.readUnsignedVInt32()];
int size = in.readUnsignedVInt32();
Ranges[] ranges = new Ranges[size];
long[] epochs = new long[size];
for (int i = 0; i < ranges.length; i++)
{
ranges[i] = KeySerializers.ranges.deserialize(in, messagingVersion);
long[] epochs = new long[in.readUnsignedVInt32()];
for (int i = 0; i < epochs.length; i++)
epochs[i] = in.readLong(); // TODO: assert lengths equal?
into.update(new RangesForEpoch.Snapshot(epochs, ranges));
epochs[i] = in.readLong();
}
Invariants.checkState(ranges.length == epochs.length);
into.update(new RangesForEpoch(epochs, ranges));
}
}
}

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

@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord;
import java.util.Collections;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
import java.util.function.BiFunction;
import javax.annotation.Nullable;
@ -34,12 +33,9 @@ import accord.api.RoutingKey;
import accord.impl.AbstractSafeCommandStore;
import accord.impl.CommandsSummary;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.NodeCommandStoreService;
import accord.local.RedundantBefore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.AbstractKeys;
import accord.primitives.Ranges;
import accord.primitives.Routables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -48,7 +44,6 @@ import accord.primitives.Unseekables;
import accord.utils.Invariants;
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
import static accord.api.Journal.*;
import static accord.utils.Invariants.illegalState;
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
@ -56,20 +51,26 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
final AccordTask<?> task;
private final @Nullable CommandsForRanges commandsForRanges;
private final AccordCommandStore commandStore;
private RangesForEpoch ranges;
private FieldUpdates fieldUpdates;
private AccordSafeCommandStore(AccordTask<?> task,
@Nullable CommandsForRanges commandsForRanges,
AccordCommandStore commandStore)
{
super(task.preLoadContext());
super(task.preLoadContext(), commandStore);
this.task = task;
this.commandsForRanges = commandsForRanges;
this.commandStore = commandStore;
commandStore.updateRangesForEpoch(this);
if (this.ranges == null)
this.ranges = Invariants.nonNull(commandStore.unsafeRangesForEpoch());
}
@Override
public CommandStores.RangesForEpoch ranges()
{
CommandStores.RangesForEpoch ranges = super.ranges();
if (ranges != null)
return ranges;
return commandStore.unsafeGetRangesForEpoch();
}
public static AccordSafeCommandStore create(AccordTask<?> operation,
@ -196,12 +197,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return commandStore.node();
}
@Override
public RangesForEpoch ranges()
{
return ranges;
}
private <O> O mapReduce(Routables<?> keysOrRanges, BiFunction<CommandsSummary, O, O> map, O accumulate)
{
Invariants.checkState(context.keys().containsAll(keysOrRanges), "Attempted to access keysOrRanges outside of what was asked for; asked for %s, accessed %s", context.keys(), keysOrRanges);
@ -278,87 +273,4 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
{
return "AccordSafeCommandStore(id=" + commandStore().id() + ")";
}
@Override
public void upsertRedundantBefore(RedundantBefore addRedundantBefore)
{
// TODO (required): this is a temporary measure, see comment on AccordJournalValueSerializers; upsert instead
// when modifying, only modify together with AccordJournalValueSerializers
ensureFieldUpdates().newRedundantBefore = RedundantBefore.merge(redundantBefore(), addRedundantBefore);
}
@Override
public void setBootstrapBeganAt(NavigableMap<TxnId, Ranges> newBootstrapBeganAt)
{
ensureFieldUpdates().newBootstrapBeganAt = newBootstrapBeganAt;
}
@Override
public void setSafeToRead(NavigableMap<Timestamp, Ranges> newSafeToRead)
{
ensureFieldUpdates().newSafeToRead = newSafeToRead;
}
@Override
public void setRangesForEpoch(CommandStores.RangesForEpoch rangesForEpoch)
{
ensureFieldUpdates().newRangesForEpoch = rangesForEpoch.snapshot();
ranges = rangesForEpoch;
}
@Override
public NavigableMap<TxnId, Ranges> bootstrapBeganAt()
{
if (fieldUpdates != null && fieldUpdates.newBootstrapBeganAt != null)
return fieldUpdates.newBootstrapBeganAt;
return super.bootstrapBeganAt();
}
@Override
public NavigableMap<Timestamp, Ranges> safeToReadAt()
{
if (fieldUpdates != null && fieldUpdates.newSafeToRead != null)
return fieldUpdates.newSafeToRead;
return super.safeToReadAt();
}
@Override
public RedundantBefore redundantBefore()
{
if (fieldUpdates != null && fieldUpdates.newRedundantBefore != null)
return fieldUpdates.newRedundantBefore;
return super.redundantBefore();
}
private FieldUpdates ensureFieldUpdates()
{
if (fieldUpdates == null) fieldUpdates = new FieldUpdates();
return fieldUpdates;
}
public FieldUpdates fieldUpdates()
{
return fieldUpdates;
}
public void postExecute()
{
if (fieldUpdates == null)
return;
if (fieldUpdates.newRedundantBefore != null)
super.unsafeSetRedundantBefore(fieldUpdates.newRedundantBefore);
if (fieldUpdates.newBootstrapBeganAt != null)
super.setBootstrapBeganAt(fieldUpdates.newBootstrapBeganAt);
if (fieldUpdates.newSafeToRead != null)
super.setSafeToRead(fieldUpdates.newSafeToRead);
if (fieldUpdates.newRangesForEpoch != null)
super.setRangesForEpoch(ranges);
}
}

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

@ -31,6 +31,8 @@ public interface IJournal extends Journal
default SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId, SavedCommand.Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Command command = loadCommand(commandStoreId, txnId, redundantBefore, durableBefore);
if (command == null)
return null;
return new SavedCommand.MinimalCommand(command.txnId(), command.saveStatus(), command.participants(), command.durability(), command.executeAt(), command.writes());
}

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

@ -259,7 +259,7 @@ public class SavedCommand
// Special-cased for Journal BurnTest integration
if ((before != null && before.result() != null && before.result() != ResultSerializers.APPLIED) ||
(after.result() != null && after.result() != ResultSerializers.APPLIED))
(after != null && after.result() != null && after.result() != ResultSerializers.APPLIED))
{
flags = collectFlags(before, after, Command::writes, false, RESULT, flags);
}

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

@ -38,7 +38,6 @@ import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.utils.CollectionSerializers;
import org.apache.cassandra.utils.NullableSerializer;
@ -138,7 +137,7 @@ public class CommandStoreSerializers
@Override
public void serialize(RedundantBefore.Entry t, DataOutputPlus out, int version) throws IOException
{
TokenRange.serializer.serialize((TokenRange) t.range, out, version);
KeySerializers.range.serialize(t.range, out, version);
Invariants.checkState(t.startOwnershipEpoch <= t.endOwnershipEpoch);
out.writeUnsignedVInt(t.startOwnershipEpoch);
if (t.endOwnershipEpoch == Long.MAX_VALUE) out.writeUnsignedVInt(0L);
@ -156,7 +155,7 @@ public class CommandStoreSerializers
@Override
public RedundantBefore.Entry deserialize(DataInputPlus in, int version) throws IOException
{
Range range = TokenRange.serializer.deserialize(in, version);
Range range = KeySerializers.range.deserialize(in, version);
long startEpoch = in.readUnsignedVInt();
long endEpoch = in.readUnsignedVInt();
if (endEpoch == 0) endEpoch = Long.MAX_VALUE;
@ -175,7 +174,7 @@ public class CommandStoreSerializers
@Override
public long serializedSize(RedundantBefore.Entry t, int version)
{
long size = TokenRange.serializer.serializedSize((TokenRange) t.range, version);
long size = KeySerializers.range.serializedSize(t.range, version);
size += TypeSizes.sizeofUnsignedVInt(t.startOwnershipEpoch);
size += TypeSizes.sizeofUnsignedVInt(t.endOwnershipEpoch == Long.MAX_VALUE ? 0 : 1 + t.endOwnershipEpoch - t.startOwnershipEpoch);
size += CommandSerializers.txnId.serializedSize(t.locallyWitnessedOrInvalidatedBefore, version);

4
test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java
View File

@ -90,12 +90,12 @@ public class AccordJournalCompactionTest
DurableBeforeAccumulator durableBeforeAccumulator = new DurableBeforeAccumulator();
NavigableMap<Timestamp, Ranges> safeToReadAtAccumulator = ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY);
NavigableMap<TxnId, Ranges> bootstrapBeganAtAccumulator = ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY);
RangesForEpoch.Snapshot rangesForEpochAccumulator = null;
RangesForEpoch rangesForEpochAccumulator = null;
Gen<RedundantBefore> redundantBeforeGen = AccordGenerators.redundantBefore(DatabaseDescriptor.getPartitioner());
Gen<DurableBefore> durableBeforeGen = AccordGenerators.durableBeforeGen(DatabaseDescriptor.getPartitioner());
Gen<NavigableMap<Timestamp, Ranges>> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner());
Gen<RangesForEpoch.Snapshot> rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner());
Gen<RangesForEpoch> rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner());
Gen<Range> rangeGen = AccordGenerators.range(DatabaseDescriptor.getPartitioner());
Gen<Deps> historicalTransactionsGen = depsGen();

6
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -440,7 +440,7 @@ public class CompactionAccordIteratorsTest
if (durableBefore != null)
durableBefores.put(commandStore.id(), durableBefore);
Int2ObjectHashMap<CommandStores.RangesForEpoch> rangesForEpochs = new Int2ObjectHashMap<>();
rangesForEpochs.put(commandStore.id(), commandStore.unsafeRangesForEpoch());
rangesForEpochs.put(commandStore.id(), commandStore.unsafeGetRangesForEpoch());
when(mockAccordService.getCompactionInfo()).thenReturn(new IAccordService.CompactionInfo(redundantBefores, rangesForEpochs, durableBefores));
return mockAccordService;
}
@ -491,8 +491,8 @@ public class CompactionAccordIteratorsTest
{
Txn txn = txnId.kind().isWrite() ? writeTxn : readTxn;
PartialDeps partialDeps = Deps.NONE.intersecting(AccordTestUtils.fullRange(txn));
PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.slice(commandStore.unsafeGetRangesForEpoch().currentRanges());
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToKeyspace(commandStore));
}).beginAsResult());

8
test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java
View File

@ -350,9 +350,9 @@ public class AccordTestUtils
this.ranges = ranges;
}
private void set(CommandStore store)
private void set()
{
add(1, new CommandStores.RangesForEpoch(1, ranges, store), ranges);
add(1, new CommandStores.RangesForEpoch(1, ranges), ranges);
}
}
@ -378,7 +378,7 @@ public class AccordTestUtils
SingleEpochRanges holder = new SingleEpochRanges(Ranges.of(range));
InMemoryCommandStore.Synchronized result = new InMemoryCommandStore.Synchronized(0, time, new AccordAgent(),
null, null, cs -> null, holder);
holder.set(result);
holder.set();
return result;
}
@ -421,7 +421,7 @@ public class AccordTestUtils
cs -> new NoOpProgressLog(),
cs -> new DefaultLocalListeners(new NoOpRemoteListeners(), new NoOpNotifySink()),
holder, journal, executor);
holder.set(result);
holder.set();
result.unsafeUpdateRangesForEpoch();
return result;
}

2
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -220,7 +220,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
});
this.topology = AccordTopology.createAccordTopology(ClusterMetadata.current());
this.topologies = new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, topology);
var rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), topology.ranges(), commandStore);
CommandStores.RangesForEpoch rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), topology.ranges());
updateHolder.add(topology.epoch(), rangesForEpoch, topology.ranges());
updateHolder.updateGlobal(topology.ranges());
commandStore.unsafeUpdateRangesForEpoch();

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

@ -186,7 +186,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
FullRangeRoute rangeRoute = ranges.toRoute(pk.toUnseekable());
Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, ranges);
DepsModel model = new DepsModel(instance.commandStore.unsafeRangesForEpoch().currentRanges());
DepsModel model = new DepsModel(instance.commandStore.unsafeGetRangesForEpoch().currentRanges());
for (int i = 0; i < numSamples; i++)
{
instance.maybeCacheEvict(keyRoute, ranges);
@ -258,7 +258,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
Range left = tokenRange(tbl.id, token - 10, token + 5);
Range right = tokenRange(tbl.id, token - 5, token + 10);
DepsModel model = new DepsModel(instance.commandStore.unsafeRangesForEpoch().currentRanges());
DepsModel model = new DepsModel(instance.commandStore.unsafeGetRangesForEpoch().currentRanges());
for (int i = 0; i < numSamples; i++)
{
Ranges partialRange = Ranges.of(rs.nextBoolean() ? left : right);

2
test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java
View File

@ -71,7 +71,7 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe
Gen.IntGen keyCountGen = keyDistribution.next(rs);
Gen.IntGen rangeCountGen = rangeDistribution.next(rs);
DepsModel model = new DepsModel(instance.commandStore.unsafeRangesForEpoch().currentRanges());
DepsModel model = new DepsModel(instance.commandStore.unsafeGetRangesForEpoch().currentRanges());
for (int i = 0; i < numSamples; i++)
{

2
test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java
View File

@ -88,7 +88,7 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom
AccordKeyspace.unsafeClear();
this.instance = new SimulatedAccordCommandStore(rs);
this.instance.commandStore.executor().cacheUnsafe().setShrinkingOn(false);
this.model = new DepsModel(instance.commandStore.unsafeRangesForEpoch().currentRanges());
this.model = new DepsModel(instance.commandStore.unsafeGetRangesForEpoch().currentRanges());
}
@Override

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

@ -517,7 +517,7 @@ public class AccordGenerators
};
}
public static Gen<RangesForEpoch.Snapshot> rangesForEpoch(IPartitioner partitioner)
public static Gen<RangesForEpoch> rangesForEpoch(IPartitioner partitioner)
{
Gen<Ranges> rangesGen = ranges(partitioner);
@ -529,7 +529,7 @@ public class AccordGenerators
Ranges[] ranges = new Ranges[size];
for (int i = 0; i < size; i++)
ranges[i] = rangesGen.next(rs);
return new RangesForEpoch.Snapshot(epochs, ranges);
return new RangesForEpoch(epochs, ranges);
};
}