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

Journal reads must select segments before sstables to avoid compaction races 

Also Fix Cassandra:
 - In memory size calculation for CommandsForKey include Unmanaged
 - Accord load out-of-band cleanup should use SafeRedundantBefore
ALso Improve Cassandra:
 - Report replay information on begin replay
 - Improve AccordService shutdown
 - Log command store RedundantBefore on shutdown
 - Segment compaction should wait for readOrder barrier to replace segments, for additional safety
 - Journal segments should share readOrder with sstables
Also Improve Accord:
 - Iterate LocalListeners in order, so can query more effectively on node
 - Refine AbstractReplay.minReplay/shouldReplay

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-21804
This commit is contained in:
Benedict Elliott Smith 2025-12-16 12:52:55 +00:00
parent c295c33a26
commit e14816e244
17 changed files with 194 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit f6b0a6998faca767e6951976097dec704c306b0e
Subproject commit f09a12da76bbc195ceb05ad859912aeb0a432dda

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

@ -692,7 +692,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
" waiting_until text,\n" +
" waiter 'TxnIdUtf8Type',\n" +
" PRIMARY KEY (command_store_id, waiting_on, waiting_until, waiter)" +
')', Int32Type.instance), FAIL, UNSORTED, ASC);
')', Int32Type.instance), FAIL, ASC, ASC);
}
@Override
@ -726,10 +726,15 @@ public class AccordDebugKeyspace extends VirtualKeyspace
LocalListeners listeners = safeStore.commandStore().unsafeGetListeners();
for (LocalListeners.TxnListener listener : listeners.txnListeners())
{
rows.add(listener.waitingOn.toString(), listener.awaitingStatus.name(), listener.waiter.toString())
rows.add(listener.waitingOn.toString(), ordered(listener.awaitingStatus), listener.waiter.toString())
.eagerCollect(ignore -> {});
}
}
private String ordered(SaveStatus saveStatus)
{
return (saveStatus.ordinal() <= 9 ? "0" : "") + saveStatus.ordinal() + '_' + saveStatus.name();
}
}
public static final class ListenersLocalTable extends AbstractLazyVirtualTable
@ -1796,7 +1801,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace
case TRY_EXECUTE:
run(txnId, commandStoreId, safeStore -> {
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
Commands.maybeExecute(safeStore, safeCommand, safeCommand.current(), true, true, NotifyWaitingOnPlus.adapter(ignore -> {}, true, true));
Command command = safeCommand.current();
if (command.saveStatus() == SaveStatus.Applying)
return Commands.applyChain(safeStore, (Command.Executed) command);
Commands.maybeExecute(safeStore, safeCommand, command, true, true, NotifyWaitingOnPlus.adapter(ignore -> {}, true, true));
return AsyncChains.success(null);
});
break;
@ -1896,7 +1904,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
AccordService.getBlocking(accord.node()
.commandStores()
.forId(commandStoreId)
.chain(PreLoadContext.contextFor(txnId, TXN_OPS), apply));
.chain(PreLoadContext.contextFor(txnId, TXN_OPS), apply)
.flatMap(i -> i));
}
private void cleanup(TxnId txnId, int commandStoreId, Cleanup cleanup)

45
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -121,7 +121,7 @@ public class Journal<K, V> implements Shutdownable
private final FlusherCallbacks flusherCallbacks;
final OpOrder readOrder = new OpOrder();
final OpOrder readOrder;
private class FlusherCallbacks implements Flusher.Callbacks
{
@ -177,7 +177,8 @@ public class Journal<K, V> implements Shutdownable
Params params,
KeySupport<K> keySupport,
ValueSerializer<K, V> valueSerializer,
SegmentCompactor<K, V> segmentCompactor)
SegmentCompactor<K, V> segmentCompactor,
OpOrder readOrder)
{
this.name = name;
this.directory = directory;
@ -185,6 +186,7 @@ public class Journal<K, V> implements Shutdownable
this.keySupport = keySupport;
this.valueSerializer = valueSerializer;
this.readOrder = readOrder;
this.metrics = new Metrics<>(name);
this.flusherCallbacks = new FlusherCallbacks();
@ -357,15 +359,25 @@ public class Journal<K, V> implements Shutdownable
return null;
}
public void readAll(K id, RecordConsumer<K> consumer)
public static <K, V> void readAll(K id, RecordConsumer<K> consumer, OpOrder.Group readGroup, Segments<K, V> segments)
{
EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
try (OpOrder.Group group = readOrder.start())
for (Segment<K, V> segment : segments.allSorted(false))
{
for (Segment<K, V> segment : segments.get().allSorted(false))
{
segment.readAll(id, holder, consumer);
}
segment.readAll(id, holder, consumer);
}
}
public void readAll(K id, RecordConsumer<K> consumer, OpOrder.Group readGroup)
{
readAll(id, consumer, readGroup, segments.get());
}
public void readAll(K id, RecordConsumer<K> consumer)
{
try (OpOrder.Group readGroup = readOrder.start())
{
readAll(id, consumer, readGroup);
}
}
@ -449,18 +461,15 @@ public class Journal<K, V> implements Shutdownable
* @return true if the record was found, false otherwise
*/
@SuppressWarnings("unused")
public boolean readLast(K id, RecordConsumer<K> consumer)
public static <K, V> boolean readLast(K id, RecordConsumer<K> consumer, OpOrder.Group readOrder, Segments<K, V> segments)
{
try (OpOrder.Group group = readOrder.start())
for (Segment<K, V> segment : segments.allSorted(false))
{
for (Segment<K, V> segment : segments.get().allSorted(false))
{
if (!segment.index().mayContainId(id))
continue;
if (!segment.index().mayContainId(id))
continue;
if (segment.readLast(id, consumer))
return true;
}
if (segment.readLast(id, consumer))
return true;
}
return false;
}
@ -728,7 +737,7 @@ public class Journal<K, V> implements Shutdownable
}
}
Segments<K, V> segments()
public Segments<K, V> segments()
{
return segments.get();
}

2
src/java/org/apache/cassandra/journal/Segments.java
View File

@ -31,7 +31,7 @@ import org.apache.cassandra.utils.concurrent.Refs;
* <p/>
* TODO (performance, expected): an interval/range structure for StaticSegment lookup based on min/max key bounds
*/
class Segments<K, V>
public class Segments<K, V>
{
private final Long2ObjectHashMap<Segment<K, V>> segments;
private SortedArrayList<Segment<K, V>> sorted;

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

@ -1321,7 +1321,7 @@ public class AccordCache implements CacheSize
try (DataInputBuffer buf = new DataInputBuffer(buffer, false))
{
builder.deserializeNext(buf, Version.LATEST);
return builder.construct(commandStore.unsafeGetRedundantBefore());
return builder.construct(commandStore.safeGetRedundantBefore());
}
catch (UnknownTableException e)
{

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

@ -317,7 +317,7 @@ public class AccordCommandStore extends CommandStore
CommandsForKey cfk = CommandsForKeyAccessor.load(id, (TokenKey) key);
if (cfk == null)
return null;
RedundantBefore.QuickBounds bounds = unsafeGetRedundantBefore().get(key);
RedundantBefore.QuickBounds bounds = safeGetRedundantBefore().get(key);
if (bounds == null)
return cfk; // TODO (required): I don't think this should be possible? but we hit it on some test
return cfk.withGcBeforeAtLeast(bounds.gcBefore, false);
@ -420,7 +420,7 @@ public class AccordCommandStore extends CommandStore
@VisibleForTesting
public Command loadCommand(TxnId txnId)
{
return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore());
return journal.loadCommand(id, txnId, safeGetRedundantBefore(), durableBefore());
}
@VisibleForTesting
@ -446,12 +446,12 @@ public class AccordCommandStore extends CommandStore
public Command.Minimal loadMinimal(TxnId txnId)
{
return journal.loadMinimal(id, txnId, unsafeGetRedundantBefore(), durableBefore());
return journal.loadMinimal(id, txnId, safeGetRedundantBefore(), durableBefore());
}
public Command.MinimalWithDeps loadMinimalWithDeps(TxnId txnId)
{
return journal.loadMinimalWithDeps(id, txnId, unsafeGetRedundantBefore(), durableBefore());
return journal.loadMinimalWithDeps(id, txnId, safeGetRedundantBefore(), durableBefore());
}
public AccordCompactionInfo getCompactionInfo()
@ -465,6 +465,11 @@ public class AccordCommandStore extends CommandStore
return new AccordCompactionInfo(id, redundantBefore, ranges, tableId);
}
public final RedundantBefore safeGetRedundantBefore()
{
return safeRedundantBefore.redundantBefore;
}
public RangeSearcher rangeSearcher()
{
return rangeSearcher;
@ -472,10 +477,10 @@ public class AccordCommandStore extends CommandStore
public AccordCommandStoreReplayer replayer()
{
boolean replayOnlyDurable = true;
boolean replayOnlyNonDurable = true;
if (journal instanceof AccordJournal)
replayOnlyDurable = ((AccordJournal)journal).configuration().replayMode() == ONLY_NON_DURABLE;
return new AccordCommandStoreReplayer(this, replayOnlyDurable);
replayOnlyNonDurable = ((AccordJournal)journal).configuration().replayMode() == ONLY_NON_DURABLE;
return new AccordCommandStoreReplayer(this, replayOnlyNonDurable);
}
static final AtomicLong nextDurabilityLoggingId = new AtomicLong();
@ -540,15 +545,22 @@ public class AccordCommandStore extends CommandStore
super.unsafeUpsertRedundantBefore(addRedundantBefore);
}
@VisibleForTesting
public void unsafeUpdateRangesForEpoch()
{
super.unsafeUpdateRangesForEpoch();
safeRedundantBefore = new SafeRedundantBefore(0, unsafeGetRedundantBefore());
}
public static class AccordCommandStoreReplayer extends AbstractReplayer
{
private final AccordCommandStore store;
private final AccordCommandStore commandStore;
private final boolean onlyNonDurable;
private AccordCommandStoreReplayer(AccordCommandStore store, boolean onlyNonDurable)
private AccordCommandStoreReplayer(AccordCommandStore commandStore, boolean onlyNonDurable)
{
super(store.unsafeGetRedundantBefore());
this.store = store;
super(commandStore, null);
this.commandStore = commandStore;
this.onlyNonDurable = onlyNonDurable;
}
@ -558,7 +570,7 @@ public class AccordCommandStore extends CommandStore
if (onlyNonDurable && !maybeShouldReplay(txnId))
return AsyncChains.success(null);
return store.chain(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
return commandStore.chain(PreLoadContext.contextFor(txnId, "Replay"), safeStore -> {
if (onlyNonDurable && !shouldReplay(txnId, safeStore.unsafeGet(txnId).current().participants()))
return null;
@ -574,12 +586,16 @@ public class AccordCommandStore extends CommandStore
void maybeLoadRedundantBefore(RedundantBefore redundantBefore)
{
Invariants.require(safeRedundantBefore == null);
if (redundantBefore != null)
{
loadRedundantBefore(redundantBefore);
Invariants.require(safeRedundantBefore == null);
safeRedundantBefore = new SafeRedundantBefore(0, redundantBefore);
}
else
{
safeRedundantBefore = new SafeRedundantBefore(0, this.unsafeGetRedundantBefore());
}
}
void maybeLoadBootstrapBeganAt(NavigableMap<TxnId, Ranges> bootstrapBeganAt)

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

@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import accord.api.Agent;
import accord.api.DataStore;
@ -33,6 +34,7 @@ import accord.local.ShardDistributor;
import accord.utils.RandomSource;
import org.apache.cassandra.cache.CacheSize;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.AccordSpec.QueueShardModel;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.accord.AccordExecutor.AccordExecutorFactory;
@ -43,8 +45,9 @@ import static org.apache.cassandra.config.DatabaseDescriptor.getAccordQueueSubmi
import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITHOUT_LOCK;
import static org.apache.cassandra.service.accord.AccordExecutor.Mode.RUN_WITH_LOCK;
import static org.apache.cassandra.service.accord.AccordExecutor.constant;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
public class AccordCommandStores extends CommandStores implements CacheSize
public class AccordCommandStores extends CommandStores implements CacheSize, Shutdownable
{
private final AccordExecutor[] executors;
private final int mask;
@ -206,25 +209,37 @@ public class AccordCommandStores extends CommandStores implements CacheSize
executor.waitForQuiescence();
}
@Override
public boolean isTerminated()
{
return Stream.of(executors).allMatch(AccordExecutor::isTerminated);
}
@Override
public synchronized void shutdown()
{
super.shutdown();
for (AccordExecutor executor : executors)
{
executor.shutdown();
}
}
@Override
public Object shutdownNow()
{
shutdown();
return null;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
{
long deadline = nanoTime() + units.toNanos(timeout);
for (AccordExecutor executor : executors)
{
try
{
executor.awaitTermination(1, TimeUnit.MINUTES);
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
long wait = Math.max(1, deadline - nanoTime());
if (!executor.awaitTermination(wait, TimeUnit.NANOSECONDS))
return false;
}
//TODO (expected): shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface)
return true;
}
}

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

@ -38,6 +38,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.impl.AbstractReplayer;
import accord.impl.CommandChange;
import accord.impl.CommandChange.Field;
import accord.local.Cleanup;
@ -109,6 +110,8 @@ import static accord.impl.CommandChange.toIterableSetFields;
import static accord.impl.CommandChange.unsetIterable;
import static accord.impl.CommandChange.validateFlags;
import static accord.local.Cleanup.Input.FULL;
import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator;
import static org.apache.cassandra.service.accord.JournalKey.Type.COMMAND_DIFF;
import static org.apache.cassandra.service.accord.journal.AccordTopologyUpdate.Accumulator;
@ -157,7 +160,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
throw new UnsupportedOperationException();
}
},
compactor(cfs, userVersion));
compactor(cfs, userVersion),
cfs.readOrdering);
this.journalTable = new AccordJournalTable<>(journal, JournalKey.SUPPORT, cfs, userVersion);
this.params = params;
}
@ -617,20 +621,23 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
class ReplayStream implements Closeable
{
final CommandStore commandStore;
final Replayer replayer;
final AbstractReplayer replayer;
final CloseableIterator<Journal.KeyRefs<JournalKey>> iter;
JournalKey prev;
public ReplayStream(CommandStore commandStore)
{
this.commandStore = commandStore;
this.replayer = commandStore.replayer();
this.replayer = (AbstractReplayer) commandStore.replayer();
// Keys in the index are sorted by command store id, so index iteration will be sequential
this.iter = journalTable.keyIterator(new JournalKey(TxnId.NONE, COMMAND_DIFF, commandStore.id()), new JournalKey(TxnId.MAX.withoutNonIdentityFlags(), COMMAND_DIFF, commandStore.id()), false);
this.iter = journalTable.keyIterator(new JournalKey(replayer.minReplay.withoutNonIdentityFlags(), COMMAND_DIFF, commandStore.id()), new JournalKey(TxnId.MAX.withoutNonIdentityFlags(), COMMAND_DIFF, commandStore.id()), false);
}
boolean replay()
{
logger.info("Beginning replay of {} with min={}, {}", commandStore, replayer.minReplay,
replayer.redundantBefore.map(b -> b == null ? null : b.maxBoundBoth(LOCALLY_DURABLE_TO_DATA_STORE, LOCALLY_DURABLE_TO_COMMAND_STORE), TxnId[]::new));
JournalKey key;
long[] segments;
while (true)

100
src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
View File

@ -76,6 +76,7 @@ import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.KeySupport;
import org.apache.cassandra.journal.RecordConsumer;
import org.apache.cassandra.journal.Segment;
import org.apache.cassandra.journal.Segments;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.service.RetryStrategy;
import org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns;
@ -347,21 +348,26 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
public void readAll(K key, RecordConsumer<K> reader)
{
try (TableKeyIterator table = readAllFromTable(key))
try (OpOrder.Group readOrder = cfs.readOrdering.start())
{
boolean hasTableData = table.advance();
long minSegment = hasTableData ? table.segment : Long.MIN_VALUE;
// First, read all journal entries newer than anything flushed into sstables
journal.readAll(key, (segment, position, key1, buffer, userVersion) -> {
if (segment > minSegment)
reader.accept(segment, position, key1, buffer, userVersion);
});
// Then, read SSTables
while (hasTableData)
// SELECT segments first, to avoid missing segments due to races compacting segment->sstable
Segments<K, V> segments = journal.segments();
try (TableKeyIterator table = readAllFromTable(key, readOrder))
{
reader.accept(table.segment, table.offset, key, table.value, table.userVersion);
hasTableData = table.advance();
boolean hasTableData = table.advance();
long minSegment = hasTableData ? table.segment : Long.MIN_VALUE;
// First, read all journal entries newer than anything flushed into sstables
Journal.readAll(key, (segment, position, key1, buffer, userVersion) -> {
if (segment > minSegment)
reader.accept(segment, position, key1, buffer, userVersion);
}, readOrder, segments);
// Then, read SSTables
while (hasTableData)
{
reader.accept(table.segment, table.offset, key, table.value, table.userVersion);
hasTableData = table.advance();
}
}
}
}
@ -373,32 +379,36 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
public void readLast(K key, RecordConsumer<K> reader)
{
try (TableKeyIterator table = readAllFromTable(key))
try (OpOrder.Group readOrder = cfs.readOrdering.start())
{
boolean hasTableData = table.advance();
long minSegment = hasTableData ? table.segment : Long.MIN_VALUE;
class JournalReader implements RecordConsumer<K>
Segments<K, V> segments = journal.segments();
try (TableKeyIterator table = readAllFromTable(key, readOrder))
{
boolean read;
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
boolean hasTableData = table.advance();
long minSegment = hasTableData ? table.segment : Long.MIN_VALUE;
class JournalReader implements RecordConsumer<K>
{
if (segment > minSegment)
boolean read;
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
{
reader.accept(segment, position, key, buffer, userVersion);
read = true;
if (segment > minSegment)
{
reader.accept(segment, position, key, buffer, userVersion);
read = true;
}
}
}
// First, read all journal entries newer than anything flushed into sstables
JournalReader journalReader = new JournalReader();
Journal.readLast(key, journalReader, readOrder, segments);
// Then, read SSTables, if we haven't found a record already
if (hasTableData && !journalReader.read)
reader.accept(table.segment, table.offset, key, table.value, table.userVersion);
}
// First, read all journal entries newer than anything flushed into sstables
JournalReader journalReader = new JournalReader();
journal.readLast(key, journalReader);
// Then, read SSTables, if we haven't found a record already
if (hasTableData && !journalReader.read)
reader.accept(table.segment, table.offset, key, table.value, table.userVersion);
}
}
@ -408,19 +418,17 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
final K key;
final List<UnfilteredRowIterator> unmerged;
final UnfilteredRowIterator merged;
final OpOrder.Group readOrder;
long segment;
int offset;
ByteBuffer value;
int userVersion;
TableKeyIterator(K key, List<UnfilteredRowIterator> unmerged, UnfilteredRowIterator merged, OpOrder.Group readOrder)
TableKeyIterator(K key, List<UnfilteredRowIterator> unmerged, UnfilteredRowIterator merged)
{
this.key = key;
this.unmerged = unmerged;
this.merged = merged;
this.readOrder = readOrder;
}
@Override
@ -455,16 +463,14 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
@Override
public void close()
{
readOrder.close();
if (merged != null)
merged.close();
}
}
private TableKeyIterator readAllFromTable(K key)
private TableKeyIterator readAllFromTable(K key, OpOrder.Group readOrder)
{
DecoratedKey pk = JournalColumns.decorate(key);
OpOrder.Group readOrder = cfs.readOrdering.start();
List<UnfilteredRowIterator> iters = new ArrayList<>(3);
try
{
@ -479,11 +485,10 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
iters.add(iter);
}
return new TableKeyIterator(key, iters, iters.isEmpty() ? null : UnfilteredRowIterators.merge(iters), readOrder);
return new TableKeyIterator(key, iters, iters.isEmpty() ? null : UnfilteredRowIterators.merge(iters));
}
catch (Throwable t)
{
readOrder.close();
for (UnfilteredRowIterator iter : iters)
{
try { iter.close(); }
@ -496,7 +501,10 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
@SuppressWarnings("resource") // Auto-closeable iterator will release related resources
public CloseableIterator<Journal.KeyRefs<K>> keyIterator(@Nullable K min, @Nullable K max, boolean includeActive)
{
return new JournalAndTableKeyIterator(min, max, includeActive);
try (OpOrder.Group readOrder = cfs.readOrdering.start())
{
return new JournalAndTableKeyIterator(min, max, includeActive);
}
}
private class TableIterator extends AbstractIterator<K> implements CloseableIterator<K>
@ -551,13 +559,19 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
private class JournalAndTableKeyIterator extends AbstractIterator<Journal.KeyRefs<K>> implements CloseableIterator<Journal.KeyRefs<K>>
{
final TableIterator tableIterator;
final Journal<K, V>.SegmentKeyIterator journalIterator;
final TableIterator tableIterator;
private JournalAndTableKeyIterator(K min, K max, boolean includeActive)
{
this.tableIterator = new TableIterator(min, max);
// We must initialise journal reader first, else we may race with segment->table compaction and miss some data
// that is, the following sequence could happen:
// - Select sstables to read
// - Segments compacted; segments removed and sstables added
// - Segment iterator created
// TODO (expected): segments should be sstables on creation
this.journalIterator = journal.segmentKeyIterator(min, max, includeActive ? ignore -> true : Segment::isStatic);
this.tableIterator = new TableIterator(min, max);
}
K prevFromTable = null;

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

@ -417,6 +417,7 @@ public class AccordObjectSizes
private static long EMPTY_CFK_SIZE = measure(new CommandsForKey(null));
private static long EMPTY_INFO_SIZE = measure(CommandsForKey.NO_INFO);
private static long EMPTY_UNMANAGED_SIZE = measure(new CommandsForKey.Unmanaged(null, TxnId.NONE, TxnId.NONE));
private static long EMPTY_INFO_EXTRA_ADDITIONAL_SIZE = measure(TxnInfo.create(TxnId.NONE, ACCEPTED, false, TxnId.NONE, NO_TXNIDS, Ballot.MAX)) - EMPTY_INFO_SIZE;
public static long commandsForKey(CommandsForKey cfk)
{
@ -438,6 +439,15 @@ public class AccordObjectSizes
size += ballot(infoExtra.ballot);
}
}
size += ObjectSizes.sizeOfReferenceArray(cfk.unmanagedCount());
size += cfk.unmanagedCount() * EMPTY_UNMANAGED_SIZE;
size += cfk.unmanagedCount() * TIMESTAMP_SIZE;
for (int i = 0 ; i < cfk.unmanagedCount() ; ++i)
{
CommandsForKey.Unmanaged unmanaged = cfk.getUnmanaged(i);
if (unmanaged.waitingUntil != unmanaged.txnId)
size += TIMESTAMP_SIZE;
}
return size;
}
}

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

@ -57,6 +57,8 @@ public class AccordSegmentCompactor<V> extends AbstractAccordSegmentCompactor<V>
cfs.addSSTables(writer.finish(true));
writer.close();
writer = null;
// await reads to complete before swapping segments, to provide additional guarantees against reading incomplete data
cfs.readOrdering.awaitNewBarrier();
}
@Override

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

@ -134,6 +134,8 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.api.Journal.TopologyUpdate;
import static accord.api.ProtocolModifiers.Toggles.FastExec.MAY_BYPASS_SAFESTORE;
import static accord.impl.progresslog.DefaultProgressLog.ModeFlag.CATCH_UP;
import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_COMMAND_STORE;
import static accord.local.RedundantStatus.Property.LOCALLY_DURABLE_TO_DATA_STORE;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
@ -252,7 +254,6 @@ public class AccordService implements IAccordService, Shutdownable
private enum State { INIT, STARTED, SHUTTING_DOWN, SHUTDOWN }
private final Node node;
private final Shutdownable nodeShutdown;
private final AccordMessageSink messageSink;
private final AccordEndpointMapper endpointMapper;
private final AccordTopologyService topologyService;
@ -435,7 +436,6 @@ public class AccordService implements IAccordService, Shutdownable
new AccordInteropFactory(endpointMapper),
journal.durableBeforePersister(),
journal);
this.nodeShutdown = toShutdownable(node);
this.requestHandler = new AccordVerbHandler<>(node, endpointMapper);
this.responseHandler = new AccordResponseVerbHandler<>(callbacks, endpointMapper);
}
@ -1042,7 +1042,7 @@ public class AccordService implements IAccordService, Shutdownable
private List<Shutdownable> shutdownableSubsystems()
{
return Arrays.asList(scheduler, nodeShutdown, journal, topologyService);
return Arrays.asList((AccordCommandStores)node.commandStores(), journal, topologyService, scheduler);
}
@VisibleForTesting
@ -1051,6 +1051,10 @@ public class AccordService implements IAccordService, Shutdownable
{
if (!ExecutorUtils.shutdownThenWait(shutdownableSubsystems(), timeout, unit))
logger.error("One or more subsystems did not shut down cleanly.");
node.commandStores().forAllUnsafe(commandStore -> {
logger.info("{} stopping with durability: {}", commandStore, commandStore.unsafeGetRedundantBefore().map(b -> b == null ? null : b.maxBoundBoth(LOCALLY_DURABLE_TO_DATA_STORE, LOCALLY_DURABLE_TO_COMMAND_STORE), TxnId[]::new));
});
}
@Override
@ -1119,43 +1123,6 @@ public class AccordService implements IAccordService, Shutdownable
sink.respond(Ok, message);
}
private static Shutdownable toShutdownable(Node node)
{
return new Shutdownable() {
private volatile boolean isShutdown = false;
@Override
public boolean isTerminated()
{
// we don't know about terminiated... so settle for shutdown!
return isShutdown;
}
@Override
public void shutdown()
{
isShutdown = true;
node.shutdown();
}
@Override
public Object shutdownNow()
{
// node doesn't offer shutdownNow
shutdown();
return null;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit units)
{
// TODO (required): expose awaitTermination in Node
// node doesn't offer
return true;
}
};
}
@VisibleForTesting
public AccordEndpointMapper endpointMapper()
{

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

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

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

@ -353,7 +353,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
}
}
public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
public CommandsForRanges.Loader loader(TxnId primaryTxnId, LoadKeysFor loadKeysFor, Unseekables<?> keysOrRanges)
{
RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
MaxDecidedRX maxDecidedRX = commandStore.unsafeGetMaxDecidedRX();

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

@ -125,7 +125,6 @@ public class AccordJournalCompactionTest
DurableBefore addDurableBefore = durableBeforeGen.next(rs);
// TODO: improve redundant before generator and re-enable
// updates.addRedundantBefore = redundantBeforeGen.next(rs);
// updates.newRedundantBefore = redundantBefore = RedundantBefore.merge(redundantBefore, updates.addRedundantBefore);
updates.newSafeToRead = safeToReadGen.next(rs);
updates.newRangesForEpoch = rangesForEpochGen.next(rs);

4
test/simulator/test/org/apache/cassandra/simulator/test/AccordJournalSimulationTest.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.ValueSerializer;
import org.apache.cassandra.utils.Isolated;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.apache.cassandra.utils.concurrent.OpOrder;
public class AccordJournalSimulationTest extends SimulationTestBase
{
@ -80,7 +81,8 @@ public class AccordJournalSimulationTest extends SimulationTestBase
spec,
new IdentityKeySerializer(),
new IdentityValueSerializer(),
SegmentCompactor.noop());
SegmentCompactor.noop(),
new OpOrder());
}),
() -> check());
}

5
test/unit/org/apache/cassandra/journal/JournalTest.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.OpOrder;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
import static org.junit.Assert.assertEquals;
@ -49,7 +50,7 @@ public class JournalTest
directory.deleteRecursiveOnExit();
Journal<TimeUUID, Long> journal =
new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop(), new OpOrder());
journal.start();
@ -70,7 +71,7 @@ public class JournalTest
journal.shutdown();
journal = new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop());
journal = new Journal<>("TestJournal", directory, TestParams.INSTANCE, TimeUUIDKeySupport.INSTANCE, LongSerializer.INSTANCE, SegmentCompactor.noop(), new OpOrder());
journal.start();
assertEquals(1L, (long) journal.readLast(id1));