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

Implement Nemesis compactor, improve partial compaction 

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20424.
This commit is contained in:
Alex Petrov 2025-03-06 16:59:17 +01:00 committed by David Capwell
parent 2999d8f235
commit 03053b14bf
8 changed files with 478 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit fe306fc5539b40d1c9d49f9afd0ca45bb74c49d3
Subproject commit 2025bcf1a6ec9f572177616fc5b6ee8714cf8f77

195
src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java Normal file
View File

@ -0,0 +1,195 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.PriorityQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.PartitionUpdate.SimpleBuilder;
import org.apache.cassandra.io.sstable.SSTableTxnWriter;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.StaticSegment.KeyOrderReader;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer;
/**
* Segment compactor: takes static segments and compacts them into a single SSTable.
*/
public abstract class AbstractAccordSegmentCompactor<V> implements SegmentCompactor<JournalKey, V>
{
protected static final Logger logger = LoggerFactory.getLogger(AbstractAccordSegmentCompactor.class);
protected final int userVersion;
protected final ColumnFamilyStore cfs;
public AbstractAccordSegmentCompactor(int userVersion, ColumnFamilyStore cfs)
{
this.userVersion = userVersion;
this.cfs = cfs;
}
void switchPartitions() {}
boolean considerWritingKey()
{
return false;
}
abstract void initializeWriter();
abstract SSTableTxnWriter writer();
abstract void finishAndAddWriter();
abstract Throwable cleanupWriter(Throwable t);
@Override
public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment<JournalKey, V>> segments)
{
Invariants.require(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size()));
logger.info("Compacting {} static segments: {}", segments.size(), segments);
PriorityQueue<KeyOrderReader<JournalKey>> readers = new PriorityQueue<>();
for (StaticSegment<JournalKey, V> segment : segments)
{
KeyOrderReader<JournalKey> reader = segment.keyOrderReader();
if (reader.advance())
readers.add(reader);
else
reader.close();
}
// nothing to compact (all segments empty, should never happen, but it is theoretically possible?) - exit early
// TODO: investigate how this comes to be, check if there is a cleanup issue
if (readers.isEmpty())
return Collections.emptyList();
initializeWriter();
JournalKey key = null;
Object builder = null;
FlyweightSerializer<Object, Object> serializer = null;
long firstDescriptor = -1, lastDescriptor = -1;
int firstOffset = -1, lastOffset = -1;
try
{
KeyOrderReader<JournalKey> reader;
while ((reader = readers.poll()) != null)
{
if (key == null || !reader.key().equals(key))
{
maybeWritePartition(key, builder, serializer, firstDescriptor, firstOffset);
switchPartitions();
key = reader.key();
serializer = (FlyweightSerializer<Object, Object>) key.type.serializer;
builder = serializer.mergerFor(key);
firstDescriptor = lastDescriptor = -1;
firstOffset = lastOffset = -1;
}
boolean advanced;
do
{
if (builder == null)
builder = serializer.mergerFor(key);
try (DataInputBuffer in = new DataInputBuffer(reader.record(), false))
{
if (lastDescriptor != -1)
{
Invariants.require(reader.descriptor.timestamp <= lastDescriptor,
"Descriptors were accessed out of order: %d was accessed after %d", reader.descriptor.timestamp, lastDescriptor);
Invariants.require(reader.descriptor.timestamp != lastDescriptor ||
reader.offset() < lastOffset,
"Offsets were accessed out of order: %d was accessed after %s", reader.offset(), lastOffset);
}
serializer.deserialize(key, builder, in, reader.descriptor.userVersion);
lastDescriptor = reader.descriptor.timestamp;
lastOffset = reader.offset();
if (firstDescriptor == -1)
{
firstDescriptor = lastDescriptor;
firstOffset = lastOffset;
}
}
if (considerWritingKey())
{
maybeWritePartition(key, builder, serializer, firstDescriptor, firstOffset);
builder = null;
firstDescriptor = lastDescriptor = -1;
firstOffset = lastOffset = -1;
}
}
while ((advanced = reader.advance()) && reader.key().equals(key));
if (advanced) readers.offer(reader); // there is more to this reader, but not with this key
else reader.close();
}
maybeWritePartition(key, builder, serializer, firstDescriptor, firstOffset);
switchPartitions();
}
catch (Throwable t)
{
t = cleanupWriter(t);
throw new RuntimeException(String.format("Caught exception while serializing. Last seen key: %s", key), t);
}
finishAndAddWriter();
return Collections.emptyList();
}
private JournalKey prevKey;
private DecoratedKey prevDecoratedKey;
private void maybeWritePartition(JournalKey key, Object builder, FlyweightSerializer<Object, Object> serializer, long descriptor, int offset) throws IOException
{
if (builder != null)
{
DecoratedKey decoratedKey = AccordKeyspace.JournalColumns.decorate(key);
if (prevKey != null)
{
Invariants.requireArgument((decoratedKey.compareTo(prevDecoratedKey) >= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1),
String.format("Partition key and JournalKey didn't have matching order, which may imply a serialization issue.\n%s (%s)\n%s (%s)",
key, decoratedKey, prevKey, prevDecoratedKey));
}
prevKey = key;
prevDecoratedKey = decoratedKey;
SimpleBuilder partitionBuilder = PartitionUpdate.simpleBuilder(cfs.metadata(), decoratedKey);
try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
{
serializer.reserialize(key, builder, out, userVersion);
partitionBuilder.row(descriptor, offset)
.add("record", out.asNewBuffer())
.add("user_version", userVersion);
}
writer().append(partitionBuilder.build().unfilteredIterator());
}
}
}

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

@ -63,6 +63,7 @@ import org.apache.cassandra.journal.Compactor;
import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.Params;
import org.apache.cassandra.journal.RecordPointer;
import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.ValueSerializer;
import org.apache.cassandra.net.MessagingService;
@ -78,7 +79,9 @@ import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.ExecutorUtils;
import static accord.impl.CommandChange.Field.CLEANUP;
import static accord.impl.CommandChange.anyFieldChanged;
import static accord.impl.CommandChange.describeFlags;
import static accord.impl.CommandChange.getFlags;
import static accord.impl.CommandChange.isChanged;
import static accord.impl.CommandChange.isNull;
@ -100,8 +103,10 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[JournalKeySupport.TOTAL_SIZE]);
private final Journal<JournalKey, Object> journal;
private final AccordJournalTable<JournalKey, Object> journalTable;
@VisibleForTesting
protected final Journal<JournalKey, Object> journal;
@VisibleForTesting
protected final AccordJournalTable<JournalKey, Object> journalTable;
private final Params params;
private final AccordAgent agent;
Node node;
@ -118,18 +123,6 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public AccordJournal(Params params, AccordAgent agent, File directory, ColumnFamilyStore cfs)
{
this.agent = agent;
AccordSegmentCompactor<Object> compactor = new AccordSegmentCompactor<>(params.userVersion(), cfs) {
@Nullable
@Override
public Collection<StaticSegment<JournalKey, Object>> compact(Collection<StaticSegment<JournalKey, Object>> staticSegments)
{
if (journalTable == null)
throw new IllegalStateException("Unsafe access to AccordJournal during <init>; journalTable was touched before it was published");
Collection<StaticSegment<JournalKey, Object>> result = super.compact(staticSegments);
journalTable.safeNotify(index -> index.remove(staticSegments));
return result;
}
};
this.journal = new Journal<>("AccordJournal", directory, params, JournalKey.SUPPORT,
// In Accord, we are using streaming serialization, i.e. Reader/Writer interfaces instead of materializing objects
new ValueSerializer<>()
@ -146,11 +139,27 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
throw new UnsupportedOperationException();
}
},
compactor);
compactor(cfs, params));
this.journalTable = new AccordJournalTable<>(journal, JournalKey.SUPPORT, cfs, params.userVersion());
this.params = params;
}
protected SegmentCompactor<JournalKey, Object> compactor(ColumnFamilyStore cfs, Params params)
{
return new AccordSegmentCompactor<>(params.userVersion(), cfs) {
@Nullable
@Override
public Collection<StaticSegment<JournalKey, Object>> compact(Collection<StaticSegment<JournalKey, Object>> staticSegments)
{
if (journalTable == null)
throw new IllegalStateException("Unsafe access to AccordJournal during <init>; journalTable was touched before it was published");
Collection<StaticSegment<JournalKey, Object>> result = super.compact(staticSegments);
journalTable.safeNotify(index -> index.remove(staticSegments));
return result;
}
};
}
@VisibleForTesting
public int inMemorySize()
{
@ -594,6 +603,12 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
{
return hasField(Field.PARTICIPANTS);
}
@Override
public String toString()
{
return after.saveStatus() + " " + describeFlags(flags);
}
}
public static class Builder extends CommandChange.Builder
@ -641,7 +656,9 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
switch (field)
{
default: throw new UnhandledEnum(field);
case CLEANUP: throw UnhandledEnum.invalid(field);
case CLEANUP:
out.writeByte(cleanup.ordinal());
break;
case EXECUTE_AT:
Invariants.require(txnId != null);
Invariants.require(executeAt != null);
@ -715,7 +732,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
{
// Since we are iterating in reverse order, we skip the fields that were
// set by entries writter later (i.e. already read ones).
if (isChanged(field, flags))
if (isChanged(field, flags) && field != CLEANUP)
skip(txnId, field, in, userVersion);
else
deserialize(field, in, userVersion);
@ -780,6 +797,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
{
switch (field)
{
default: throw new UnhandledEnum(field);
case EXECUTE_AT:
ExecuteAtSerializer.skip(txnId, in);
break;

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

@ -17,162 +17,50 @@
*/
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.PriorityQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.PartitionUpdate.SimpleBuilder;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableTxnWriter;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.StaticSegment.KeyOrderReader;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer;
/**
* Segment compactor: takes static segments and compacts them into a single SSTable.
*/
public class AccordSegmentCompactor<V> implements SegmentCompactor<JournalKey, V>
public class AccordSegmentCompactor<V> extends AbstractAccordSegmentCompactor<V>
{
private static final Logger logger = LoggerFactory.getLogger(AccordSegmentCompactor.class);
private final int userVersion;
private final ColumnFamilyStore cfs;
private SSTableTxnWriter writer;
public AccordSegmentCompactor(int userVersion, ColumnFamilyStore cfs)
{
this.userVersion = userVersion;
this.cfs = cfs;
super(userVersion, cfs);
}
@Override
public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment<JournalKey, V>> segments)
void initializeWriter()
{
Invariants.require(segments.size() >= 2, () -> String.format("Can only compact 2 or more segments, but got %d", segments.size()));
logger.info("Compacting {} static segments: {}", segments.size(), segments);
PriorityQueue<KeyOrderReader<JournalKey>> readers = new PriorityQueue<>();
for (StaticSegment<JournalKey, V> segment : segments)
{
KeyOrderReader<JournalKey> reader = segment.keyOrderReader();
if (reader.advance())
readers.add(reader);
else
reader.close();
}
// nothing to compact (all segments empty, should never happen, but it is theoretically possible?) - exit early
// TODO: investigate how this comes to be, check if there is a cleanup issue
if (readers.isEmpty())
return Collections.emptyList();
Descriptor descriptor = cfs.newSSTableDescriptor(cfs.getDirectories().getDirectoryForNewSSTables());
SerializationHeader header = new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS);
try (SSTableTxnWriter writer = SSTableTxnWriter.create(cfs, descriptor, 0, 0, null, false, header))
{
JournalKey key = null;
Object builder = null;
FlyweightSerializer<Object, Object> serializer = null;
long firstDescriptor = -1, lastDescriptor = -1;
int firstOffset = -1, lastOffset = -1;
try
{
KeyOrderReader<JournalKey> reader;
while ((reader = readers.poll()) != null)
{
if (key == null || !reader.key().equals(key))
{
maybeWritePartition(writer, key, builder, serializer, firstDescriptor, firstOffset);
key = reader.key();
serializer = (FlyweightSerializer<Object, Object>) key.type.serializer;
builder = serializer.mergerFor(key);
firstDescriptor = lastDescriptor = -1;
firstOffset = lastOffset = -1;
}
boolean advanced;
do
{
try (DataInputBuffer in = new DataInputBuffer(reader.record(), false))
{
if (lastDescriptor != -1)
{
Invariants.require(reader.descriptor.timestamp <= lastDescriptor,
"Descriptors were accessed out of order: %d was accessed after %d", reader.descriptor.timestamp, lastDescriptor);
Invariants.require(reader.descriptor.timestamp != lastDescriptor ||
reader.offset() < lastOffset,
"Offsets were accessed out of order: %d was accessed after %s", reader.offset(), lastOffset);
}
serializer.deserialize(key, builder, in, reader.descriptor.userVersion);
lastDescriptor = reader.descriptor.timestamp;
lastOffset = reader.offset();
if (firstDescriptor == -1)
{
firstDescriptor = lastDescriptor;
firstOffset = lastOffset;
}
}
}
while ((advanced = reader.advance()) && reader.key().equals(key));
if (advanced) readers.offer(reader); // there is more to this reader, but not with this key
else reader.close();
}
maybeWritePartition(writer, key, builder, serializer, firstDescriptor, firstOffset);
}
catch (Throwable t)
{
Throwable accumulate = writer.abort(t);
throw new RuntimeException(String.format("Caught exception while serializing. Last seen key: %s", key), accumulate);
}
cfs.addSSTables(writer.finish(true));
return Collections.emptyList();
}
this.writer = SSTableTxnWriter.create(cfs, descriptor, 0, 0, null, false, header);
}
private JournalKey prevKey;
private DecoratedKey prevDecoratedKey;
private void maybeWritePartition(SSTableTxnWriter writer, JournalKey key, Object builder, FlyweightSerializer<Object, Object> serializer, long descriptor, int offset) throws IOException
@Override
SSTableTxnWriter writer()
{
if (builder != null)
{
DecoratedKey decoratedKey = AccordKeyspace.JournalColumns.decorate(key);
return writer;
}
if (prevKey != null)
{
Invariants.requireArgument((decoratedKey.compareTo(prevDecoratedKey) >= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1),
String.format("Partition key and JournalKey didn't have matching order, which may imply a serialization issue.\n%s (%s)\n%s (%s)",
key, decoratedKey, prevKey, prevDecoratedKey));
}
prevKey = key;
prevDecoratedKey = decoratedKey;
@Override
void finishAndAddWriter()
{
cfs.addSSTables(writer.finish(true));
writer.close();
}
SimpleBuilder partitionBuilder = PartitionUpdate.simpleBuilder(cfs.metadata(), decoratedKey);
try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
{
serializer.reserialize(key, builder, out, userVersion);
partitionBuilder.row(descriptor, offset)
.add("record", out.asNewBuffer())
.add("user_version", userVersion);
}
writer.append(partitionBuilder.build().unfilteredIterator());
}
@Override
Throwable cleanupWriter(Throwable t)
{
return writer.abort(t);
}
}

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

@ -73,6 +73,7 @@ public class WaitingOnSerializer
@Override
public WaitingOn provide(TxnId txnId, PartialDeps deps, Timestamp executeAtLeast, long uniqueHlc)
{
Invariants.nonNull(deps);
RoutingKeys keys = deps.keyDeps.keys();
RangeDeps directRangeDeps = deps.rangeDeps;
int txnIdCount = directRangeDeps.txnIdCount();

154
test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
View File

@ -18,11 +18,15 @@
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.junit.Before;
@ -37,13 +41,27 @@ import accord.impl.basic.Cluster;
import accord.impl.basic.RandomDelayQueue;
import accord.local.CommandStores;
import accord.local.Node;
import accord.primitives.EpochSupplier;
import accord.utils.DefaultRandom;
import accord.utils.RandomSource;
import org.apache.cassandra.ServerTestUtils;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.compaction.CompactionController;
import org.apache.cassandra.db.compaction.CompactionIterator;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;
import org.apache.cassandra.db.compaction.writers.DefaultCompactionWriter;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.Params;
import org.apache.cassandra.journal.SegmentCompactor;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.journal.TestParams;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
@ -56,9 +74,9 @@ import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.ResultSerializers;
import org.apache.cassandra.service.accord.serializers.TopologySerializers;
import org.apache.cassandra.tools.FieldUtil;
import org.apache.cassandra.utils.concurrent.Condition;
import static accord.impl.PrefixedIntHashKey.ranges;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
public class AccordJournalBurnTest extends BurnTestBase
{
@ -116,11 +134,11 @@ public class AccordJournalBurnTest extends BurnTestBase
List<Node.Id> clients = generateIds(true, 1 + random.nextInt(4));
int rf;
float chance = random.nextFloat();
if (chance < 0.2f) { rf = random.nextInt(2, 9); }
else if (chance < 0.4f) { rf = 3; }
else if (chance < 0.7f) { rf = 5; }
else if (chance < 0.8f) { rf = 7; }
else { rf = 9; }
if (chance < 0.2f) rf = random.nextInt(2, 9);
else if (chance < 0.4f) rf = 3;
else if (chance < 0.7f) rf = 5;
else if (chance < 0.8f) rf = 7;
else rf = 9;
List<Node.Id> nodes = generateIds(false, random.nextInt(rf, rf * 3));
@ -135,6 +153,7 @@ public class AccordJournalBurnTest extends BurnTestBase
AccordKeyspace.TABLES = Tables.of(metadatas);
setUp();
}
Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord");
burn(random, new TopologyFactory(rf, ranges(0, HASH_RANGE_START, HASH_RANGE_END, random.nextInt(Math.max(nodes.size() + 1, rf), nodes.size() * 3))),
@ -145,7 +164,7 @@ public class AccordJournalBurnTest extends BurnTestBase
operations,
10 + random.nextInt(30),
new RandomDelayQueue.Factory(random).get(),
(node, agent) -> {
(node, agent, randomSource) -> {
try
{
File directory = new File(Files.createTempDirectory(Integer.toString(counter.incrementAndGet())));
@ -154,37 +173,18 @@ public class AccordJournalBurnTest extends BurnTestBase
cfs.disableAutoCompaction();
AccordJournal journal = new AccordJournal(new TestParams()
{
@Override
public FlushMode flushMode()
{
return FlushMode.PERIODIC;
}
@Override
public long flushPeriod(TimeUnit units)
{
return 1;
}
@Override
public int segmentSize()
{
return 32 * 1024 * 1024;
}
@Override
public boolean enableCompaction()
{
return false;
}
}, new AccordAgent(), directory, cfs)
{
@Override
public void saveCommand(int store, CommandUpdate update, @Nullable Runnable onFlush)
{
Condition condition = Condition.newOneTimeCondition();
super.saveCommand(store, update, condition::signal);
condition.awaitUninterruptibly();
// For the purpose of this test, we do not have to wait for flush, since we do not test durability and are using mmap
super.saveCommand(store, update, () -> {});
if (onFlush != null)
onFlush.run();
}
@ -192,21 +192,105 @@ public class AccordJournalBurnTest extends BurnTestBase
@Override
public void saveStoreState(int store, FieldUpdates fieldUpdates, @Nullable Runnable onFlush)
{
Condition condition = Condition.newOneTimeCondition();
super.saveStoreState(store, fieldUpdates, condition::signal);
super.saveStoreState(store, fieldUpdates, () -> {});
if (onFlush != null)
onFlush.run();
}
@Override
public void saveTopology(TopologyUpdate topologyUpdate, @Nullable Runnable onFlush)
public void saveTopology(TopologyUpdate topologyUpdate, Runnable onFlush)
{
Condition condition = Condition.newOneTimeCondition();
super.saveTopology(topologyUpdate, condition::signal);
super.saveTopology(topologyUpdate, () -> {});
if (onFlush != null)
onFlush.run();
}
@Override
protected SegmentCompactor<JournalKey, Object> compactor(ColumnFamilyStore cfs, Params params)
{
return new NemesisAccordSegmentCompactor<>(params.userVersion(), cfs, randomSource.fork())
{
@Nullable
@Override
public Collection<StaticSegment<JournalKey, Object>> compact(Collection<StaticSegment<JournalKey, Object>> staticSegments)
{
if (journalTable == null)
throw new IllegalStateException("Unsafe access to AccordJournal during <init>; journalTable was touched before it was published");
Collection<StaticSegment<JournalKey, Object>> result = super.compact(staticSegments);
journalTable.safeNotify(index -> index.remove(staticSegments));
return result;
}
};
}
private CompactionAwareWriter getCompactionAwareWriter(ColumnFamilyStore cfs,
Directories directories,
LifecycleTransaction transaction,
Set<SSTableReader> nonExpiredSSTables)
{
return new DefaultCompactionWriter(cfs, directories, transaction, nonExpiredSSTables, false, 0);
}
@Override
public void purge(CommandStores commandStores, EpochSupplier minEpoch)
{
this.journal.closeCurrentSegmentForTestingIfNonEmpty();
this.journal.runCompactorForTesting();
List<SSTableReader> all = new ArrayList<>(cfs.getLiveSSTables());
if (all.size() <= 1)
return;
Set<SSTableReader> sstables = new HashSet<>();
int min, max;
while (true)
{
int tmp1 = randomSource.nextInt(0, all.size());
int tmp2 = randomSource.nextInt(0, all.size());
if (tmp1 != tmp2 && Math.abs(tmp1 - tmp2) >= 1)
{
min = Math.min(tmp1, tmp2);
max = Math.max(tmp1, tmp2);
break;
}
}
// Random subset
for (int i = min; i < max; i++)
sstables.add(all.get(i));
List<ISSTableScanner> scanners = sstables.stream().map(SSTableReader::getScanner).collect(Collectors.toList());
Collection<SSTableReader> newSStables;
try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.COMPACTION);
CompactionController controller = new CompactionController(cfs, sstables, 0);
CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners, controller, 0, nextTimeUUID()))
{
CompactionManager.instance.active.beginCompaction(ci);
try (CompactionAwareWriter writer = getCompactionAwareWriter(cfs, cfs.getDirectories(), txn, sstables))
{
while (ci.hasNext())
{
writer.append(ci.next());
ci.setTargetDirectory(writer.getSStableDirectory().path());
}
// point of no return
newSStables = writer.finish();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
finally
{
CompactionManager.instance.active.finishCompaction(ci);
}
}
}
@Override
public void replay(CommandStores commandStores)
{

105
test/distributed/org/apache/cassandra/service/accord/NemesisAccordSegmentCompactor.java Normal file
View File

@ -0,0 +1,105 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import accord.utils.RandomSource;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTableTxnWriter;
/**
* Nemesis compactor: a compactor that will distribute your keys over a large(r) number of SSTables.
*
* For testing purposes only.
*/
public class NemesisAccordSegmentCompactor<V> extends AbstractAccordSegmentCompactor<V>
{
private final RandomSource randomSource;
private final SSTableTxnWriter[] writers;
private final Set<SSTableTxnWriter> written = new HashSet<>();
public NemesisAccordSegmentCompactor(int userVersion, ColumnFamilyStore cfs, RandomSource randomSource)
{
super(userVersion, cfs);
this.randomSource = randomSource;
this.writers = new SSTableTxnWriter[randomSource.nextInt(2, 10)];
}
@Override
boolean considerWritingKey()
{
if (written.size() == writers.length - 1)
return false;
return randomSource.nextBoolean();
}
@Override
void switchPartitions()
{
written.clear();
}
@Override
void initializeWriter()
{
for (int i = 0; i < writers.length; i++)
{
Descriptor descriptor = cfs.newSSTableDescriptor(cfs.getDirectories().getDirectoryForNewSSTables());
SerializationHeader header = new SerializationHeader(true, cfs.metadata(), cfs.metadata().regularAndStaticColumns(), EncodingStats.NO_STATS);
writers[i] = SSTableTxnWriter.create(cfs, descriptor, 0, 0, null, false, header);
}
}
@Override
SSTableTxnWriter writer()
{
for (int i = 0; i < 10_000; i++)
{
SSTableTxnWriter writer = writers[randomSource.nextInt(writers.length)];
if (written.add(writer))
return writer;
}
throw new IllegalStateException(String.format("Could not pick an sstable from %s. Written: %s", Arrays.asList(writers), written));
}
@Override
void finishAndAddWriter()
{
for (SSTableTxnWriter writer : writers)
{
cfs.addSSTables(writer.finish(true));
writer.close();
}
Arrays.fill(writers, null);
}
@Override
Throwable cleanupWriter(Throwable t)
{
for (SSTableTxnWriter writer : writers)
t = writer.abort(t);
return t;
}
}

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

@ -29,7 +29,6 @@ import java.util.function.BooleanSupplier;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.ToLongFunction;
import javax.annotation.Nullable;
import accord.api.Agent;
@ -235,7 +234,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
};
this.journal = new DefaultJournal(nodeId, agent);
this.journal = new DefaultJournal(nodeId, agent, rs.fork());
this.commandStore = new AccordCommandStore(0,
storeService,
agent,
@ -466,9 +465,9 @@ public class SimulatedAccordCommandStore implements AutoCloseable
private static class DefaultJournal extends InMemoryJournal implements RangeSearcher.Supplier
{
private final RouteInMemoryIndex<?> index = new RouteInMemoryIndex<>();
private DefaultJournal(Node.Id id, Agent agent)
private DefaultJournal(Node.Id id, Agent agent, RandomSource rs)
{
super(id, agent);
super(id, agent, rs);
}
@Override