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

CEP-45: Integrate Mutation Journal with Memtable flush tracking 

* Tracking of Memtable -> SSTable flushes in Mutation Journal segments
  * Replay of Mutation Journal segments that were not fully flushed to SSTables
  * Distinguish between CommitLog and MutationJournal offsets
  * Effectively adds support for node bounces

Patch by Alex Petrov; reviewed by Aleksey Yeschenko for CASSANDRA-20919
This commit is contained in:
Alex Petrov 2025-09-18 17:54:52 +02:00 committed by Blake Eggleston
parent 1e036cf620
commit a7b431d683
28 changed files with 1578 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -138,6 +138,7 @@ import org.apache.cassandra.repair.consistent.admin.CleanupSummary;
import org.apache.cassandra.repair.consistent.admin.PendingStat;
import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
import org.apache.cassandra.replication.MutationId;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.CompactionParams;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
@ -517,9 +518,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
logger.info("Initializing {}.{}", getKeyspaceName(), name);
Memtable initialMemtable = DatabaseDescriptor.isDaemonInitialized() ?
createMemtable(new AtomicReference<>(CommitLog.instance.getCurrentPosition())) :
null;
Memtable initialMemtable = null;
if (DatabaseDescriptor.isDaemonInitialized())
{
CommitLogPosition commitLogPosition;
if (metadata().replicationType().isTracked())
commitLogPosition = MutationJournal.instance.getCurrentPosition();
else
commitLogPosition = CommitLog.instance.getCurrentPosition();
initialMemtable = createMemtable(new AtomicReference<>(commitLogPosition));
}
memtableMetricsReleaser = memtableFactory.createMemtableMetricsReleaser(metadata);
data = new Tracker(this, initialMemtable, loadSSTables);
@ -1183,8 +1191,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
// If a flush errored out but the error was ignored, make sure we don't discard the commit log.
if (flushFailure == null && mainMemtable != null)
{
CommitLogPosition commitLogLowerBound = mainMemtable.getCommitLogLowerBound();
commitLogUpperBound = mainMemtable.getFinalCommitLogUpperBound();
CommitLog.instance.discardCompletedSegments(metadata.id, mainMemtable.getCommitLogLowerBound(), commitLogUpperBound);
TableMetadata metadata = metadata();
if (metadata.replicationType().isTracked())
MutationJournal.instance.notifyFlushed(metadata.id, commitLogLowerBound, commitLogUpperBound);
else
CommitLog.instance.discardCompletedSegments(metadata.id, commitLogLowerBound, commitLogUpperBound);
}
metric.pendingFlushes.dec();
@ -1251,7 +1264,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
// we then ensure an atomic decision is made about the upper bound of the continuous range of commit log
// records owned by this memtable
setCommitLogUpperBound(commitLogUpperBound);
setCommitLogUpperBound(commitLogUpperBound, metadata().replicationType().isTracked());
// we then issue the barrier; this lets us wait for all operations started prior to the barrier to complete;
// since this happens after wiring up the commitLogUpperBound, we also know all operations with earlier
@ -1452,7 +1465,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
}
// atomically set the upper bound for the commit log
private static void setCommitLogUpperBound(AtomicReference<CommitLogPosition> commitLogUpperBound)
private static void setCommitLogUpperBound(AtomicReference<CommitLogPosition> commitLogUpperBound, boolean useMutationJournal)
{
// we attempt to set the holder to the current commit log context. at the same time all writes to the memtables are
// also maintaining this value, so if somebody sneaks ahead of us somehow (should be rare) we simply retry,
@ -1460,7 +1473,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner
CommitLogPosition lastReplayPosition;
while (true)
{
lastReplayPosition = new Memtable.LastCommitLogPosition((CommitLog.instance.getCurrentPosition()));
CommitLogPosition commitLogPosition;
if (useMutationJournal)
commitLogPosition = MutationJournal.instance.getCurrentPosition();
else
commitLogPosition = CommitLog.instance.getCurrentPosition();
lastReplayPosition = new Memtable.LastCommitLogPosition(commitLogPosition);
CommitLogPosition currentLast = commitLogUpperBound.get();
if ((currentLast == null || currentLast.compareTo(lastReplayPosition) <= 0)
&& commitLogUpperBound.compareAndSet(currentLast, lastReplayPosition))
@ -3329,7 +3348,11 @@ public <T> T withAllSSTables(final OperationType operationType, Function<Lifecyc
data.notifyDropped(DatabaseDescriptor.getAutoSnapshotTtl());
CommitLog.instance.forceRecycleAllSegments(Collections.singleton(metadata.id));
// TODO (required): test mutation tracking + table dropping
if (metadata().replicationType().isTracked())
MutationJournal.instance.notifyFlushed(metadata.id, new CommitLogPosition(0, 0), MutationJournal.instance.getCurrentPosition());
else
CommitLog.instance.forceRecycleAllSegments(Collections.singleton(metadata.id));
compactionStrategyManager.shutdown();

5
src/java/org/apache/cassandra/db/Mutation.java
View File

@ -203,6 +203,11 @@ public class Mutation implements IMutation, Supplier<Mutation>
return key;
}
public ImmutableMap<TableId, PartitionUpdate> modifications()
{
return modifications;
}
public ImmutableCollection<PartitionUpdate> getPartitionUpdates()
{
return modifications.values();

6
src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
View File

@ -120,7 +120,7 @@ public class CommitLogReplayer implements CommitLogReadHandler
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// but, if we've truncated the cf in question, then we need to need to start replay after the truncation
// but, if we've truncated the cf in question, then we need to start replay after the truncation
CommitLogPosition truncatedAt = SystemKeyspace.getTruncatedPosition(cfs.metadata.id);
if (truncatedAt != null)
{
@ -322,7 +322,7 @@ public class CommitLogReplayer implements CommitLogReadHandler
{
assert !newPUCollector.isEmpty();
Keyspace.open(newPUCollector.getKeyspaceName()).apply(newPUCollector.build(), false, true, false);
keyspace.apply(newPUCollector.build(), false, true, false);
commitLogReplayer.keyspacesReplayed.add(keyspace);
}
}
@ -438,7 +438,7 @@ public class CommitLogReplayer implements CommitLogReadHandler
if (toReplay.isEmpty())
logger.info("All tables will be included in commit log replay.");
else
logger.info("Tables to be replayed: {}", toReplay.asMap().toString());
logger.info("Tables to be replayed: {}", toReplay.asMap());
return new CustomReplayFilter(toReplay);
}

6
src/java/org/apache/cassandra/db/tracked/TrackedKeyspaceWriteHandler.java
View File

@ -24,7 +24,6 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.WriteContext;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.journal.RecordPointer;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -40,10 +39,9 @@ public class TrackedKeyspaceWriteHandler implements KeyspaceWriteHandler
group = Keyspace.writeOrder.start();
Tracing.trace("Appending to mutation journal");
RecordPointer pointer = MutationJournal.instance.write(mutation.id(), mutation);
CommitLogPosition pointer = MutationJournal.instance.write(mutation.id(), mutation);
// TODO (preferred): update journal to return CommitLogPosition or otherwise remove requirement to allocate second object here
return new CassandraWriteContext(group, new CommitLogPosition(pointer.segment, pointer.position));
return new CassandraWriteContext(group, pointer);
}
catch (Throwable t)
{

38
src/java/org/apache/cassandra/journal/ActiveSegment.java
View File

@ -35,6 +35,7 @@ import org.slf4j.LoggerFactory;
import accord.utils.Invariants;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.SyncUtil;
import org.apache.cassandra.utils.concurrent.OpOrder;
@ -100,6 +101,11 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
public CommitLogPosition currentPosition()
{
return new CommitLogPosition(id(), (int) allocateOffset);
}
static <K, V> ActiveSegment<K, V> create(Descriptor descriptor, Params params, KeySupport<K> keySupport)
{
InMemoryIndex<K> index = InMemoryIndex.create(keySupport);
@ -136,6 +142,12 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
throw new UnsupportedOperationException();
}
@Override
public void persistMetadata()
{
throw new UnsupportedOperationException("Can not mutate active segment's metadata.");
}
/**
* Read the entry and specified offset into the entry holder.
* Expects the caller to acquire the ref to the segment and the record to exist.
@ -427,19 +439,21 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
final class Allocation
final class Allocation extends RecordPointer
{
private final OpOrder.Group appendOp;
private final ByteBuffer buffer;
private final int start;
private final int length;
Allocation(OpOrder.Group appendOp, ByteBuffer buffer, int length)
{
super(descriptor.timestamp, buffer.position(), length);
this.appendOp = appendOp;
this.buffer = buffer;
this.start = buffer.position();
this.length = length;
}
Segment<K, V> segment()
{
return ActiveSegment.this;
}
void write(K id, ByteBuffer record)
@ -448,7 +462,7 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
{
EntrySerializer.write(id, record, keySupport, buffer, descriptor.userVersion);
metadata.update();
index.update(id, start, length);
index.update(id, position, length);
}
catch (IOException e)
{
@ -474,14 +488,12 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
}
}
// Variant of write that does not allocate/return a record pointer
void writeInternal(K id, ByteBuffer record)
{
try
{
EntrySerializer.write(id, record, keySupport, buffer, descriptor.userVersion);
index.update(id, start, length);
index.update(id, position, length);
metadata.update();
}
catch (IOException e)
@ -498,13 +510,13 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
{
try (Timer.Context ignored = waitingOnFlush.time())
{
waitForFlush(start);
waitForFlush(position);
}
}
boolean isFsynced()
{
return fsyncedTo >= start + length;
return fsyncedTo >= position + length;
}
Descriptor descriptor()
@ -514,12 +526,12 @@ public final class ActiveSegment<K, V> extends Segment<K, V>
int start()
{
return start;
return position;
}
RecordPointer recordPointer()
{
return new RecordPointer(descriptor.timestamp, start, length);
return this;
}
}

51
src/java/org/apache/cassandra/journal/DeserializedRecordConsumer.java Normal file
View File

@ -0,0 +1,51 @@
/*
* 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.journal;
import org.apache.cassandra.io.util.DataInputBuffer;
import java.io.IOException;
import java.nio.ByteBuffer;
public abstract class DeserializedRecordConsumer<K, V> implements RecordConsumer<K>
{
final ValueSerializer<K, V> valueSerializer;
public DeserializedRecordConsumer(ValueSerializer<K, V> valueSerializer)
{
this.valueSerializer = valueSerializer;
}
@Override
public void accept(long segment, int position, K key, ByteBuffer buffer, int userVersion)
{
try (DataInputBuffer in = new DataInputBuffer(buffer, false))
{
V value = valueSerializer.deserialize(key, in, userVersion);
accept(segment, position, key, value);
}
catch (IOException e)
{
// can only throw if serializer is buggy
throw new RuntimeException(e);
}
}
protected abstract void accept(long segment, int position, K key, V value);
}

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

@ -33,6 +33,7 @@ import java.util.concurrent.locks.LockSupport;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.LongConsumer;
import java.util.function.Predicate;
import java.util.zip.CRC32;
@ -41,6 +42,7 @@ import javax.annotation.Nullable;
import com.codahale.metrics.Timer.Context;
import com.google.common.annotations.VisibleForTesting;
import org.agrona.collections.Long2ObjectHashMap;
import org.jctools.queues.MpscUnboundedArrayQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -117,7 +119,7 @@ public class Journal<K, V> implements Shutdownable
// segment that is ready to be used; allocator thread fills this and blocks until consumed
private volatile ActiveSegment<K, V> availableSegment = null;
private final AtomicReference<Segments<K, V>> segments = new AtomicReference<>();
private final AtomicReference<Segments<K, V>> segments = new AtomicReference<>(new Segments<>(new Long2ObjectHashMap<>()));
final AtomicReference<State> state = new AtomicReference<>(State.UNINITIALIZED);
@ -369,6 +371,32 @@ public class Journal<K, V> implements Shutdownable
return null;
}
public boolean readLast(K id, RecordConsumer<K> consumer)
{
EntrySerializer.EntryHolder<K> holder = new EntrySerializer.EntryHolder<>();
try (OpOrder.Group group = readOrder.start())
{
for (Segment<K, V> segment : segments.get().allSorted(true))
{
if (segment.readLast(id, consumer))
{
return true;
}
}
}
return false;
}
public void readLast(K id, long segmentId, DeserializedRecordConsumer<K, V> consumer)
{
Segment<K, V> segment = segments.get().get(segmentId);
try (OpOrder.Group group = readOrder.start())
{
segment.readLast(id, 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<>();
@ -510,16 +538,16 @@ public class Journal<K, V> implements Shutdownable
public int sizeOfRecord(RecordPointer pointer)
{
Descriptor descriptor = segments.get().descriptor(pointer.segment);
Descriptor descriptor = segments.get().descriptor(pointer.segmentId);
Invariants.nonNull(descriptor);
return pointer.size - EntrySerializer.overheadSize(keySupport, descriptor.userVersion);
return pointer.length - EntrySerializer.overheadSize(keySupport, descriptor.userVersion);
}
public boolean read(RecordPointer pointer, RecordConsumer<K> consumer)
{
try (OpOrder.Group group = readOrder.start())
{
Segment<K, V> segment = segments.get().get(pointer.segment);
Segment<K, V> segment = segments.get().get(pointer.segmentId);
return segment != null && segment.read(pointer, consumer);
}
}
@ -588,7 +616,6 @@ public class Journal<K, V> implements Shutdownable
private ActiveSegment<K, V>.Allocation allocate(int entrySize)
{
ActiveSegment<K, V> segment = currentSegment;
ActiveSegment<K, V>.Allocation alloc;
while (null == (alloc = segment.allocate(entrySize)))
@ -843,11 +870,23 @@ public class Journal<K, V> implements Shutdownable
return oldest;
}
public List<Segment<K, V>> getSegments(long lowerBound, long upperBound)
{
List<Segment<K, V>> res = new ArrayList<>();
segments().select(lowerBound, upperBound, res);
return res;
}
public ActiveSegment<K, V> currentActiveSegment()
{
return currentSegment;
}
@Nullable protected Segment<K, V> getSegment(long timestamp)
{
return segments().get(timestamp);
}
ActiveSegment<K, V> getActiveSegment(long timestamp)
{
// we can race with segment addition to the segments() collection, with a new segment appearing in currentSegment first
@ -889,10 +928,12 @@ public class Journal<K, V> implements Shutdownable
private class CloseActiveSegmentRunnable implements Runnable
{
private final ActiveSegment<K, V> activeSegment;
private final Runnable onDone;
CloseActiveSegmentRunnable(ActiveSegment<K, V> activeSegment)
CloseActiveSegmentRunnable(ActiveSegment<K, V> activeSegment, @Nullable Runnable onDone)
{
this.activeSegment = activeSegment;
this.onDone = onDone;
}
@Override
@ -904,10 +945,16 @@ public class Journal<K, V> implements Shutdownable
activeSegment.persistComponents();
replaceCompletedSegment(activeSegment, StaticSegment.open(activeSegment.descriptor, keySupport));
activeSegment.release(Journal.this);
if (onDone != null) onDone.run();
}
}
void closeActiveSegmentAndOpenAsStatic(ActiveSegment<K, V> activeSegment)
protected void closeActiveSegmentAndOpenAsStatic(ActiveSegment<K, V> activeSegment)
{
closeActiveSegmentAndOpenAsStatic(activeSegment, null);
}
protected void closeActiveSegmentAndOpenAsStatic(ActiveSegment<K, V> activeSegment, @Nullable Runnable onDone)
{
if (activeSegment.isEmpty())
{
@ -916,7 +963,7 @@ public class Journal<K, V> implements Shutdownable
return;
}
closer.execute(new CloseActiveSegmentRunnable(activeSegment));
closer.execute(new CloseActiveSegmentRunnable(activeSegment, onDone));
}
@VisibleForTesting
@ -1019,11 +1066,19 @@ public class Journal<K, V> implements Shutdownable
/**
* segment iterator iterates all keys in order.
*/
public SegmentKeyIterator segmentKeyIterator(K min, K max, Predicate<Segment<?, ?>> include)
public SegmentKeyIterator segmentKeyIterator(K min, K max, Predicate<Segment<K, V>> include)
{
return new SegmentKeyIterator(min, max, include);
}
/**
* Static segment iterator iterates all keys in selected segments in order.
*/
public SegmentKeyIterator staticSegmentKeyIterator(Predicate<Segment<K, V>> predicate)
{
return new SegmentKeyIterator(null, null, predicate.and(s -> s.isStatic()));
}
/**
* List of key and a list of segment descriptors referencing this key
*/
@ -1043,6 +1098,17 @@ public class Journal<K, V> implements Shutdownable
this.segments = new long[maxSize];
}
public void segments(LongConsumer consumer)
{
for (int i = 0; i < size; i++)
consumer.accept(segments[i]);
}
public long lastSegment()
{
return segments[segments.length - 1];
}
public long[] copyOfSegments()
{
return segments == null ? new long[0] : Arrays.copyOf(segments, size);
@ -1088,11 +1154,17 @@ public class Journal<K, V> implements Shutdownable
private final ReferencedSegments<K, V> segments;
private final MergeIterator<Head, KeyRefs<K>> iterator;
public SegmentKeyIterator(K min, K max, Predicate<Segment<?, ?>> include)
public SegmentKeyIterator(K min, K max)
{
this(min, max, s -> true);
}
public SegmentKeyIterator(K min, K max, Predicate<Segment<K, V>> include)
{
this.segments = selectAndReference(s -> include.test(s) && !s.isEmpty()
&& (min == null || keySupport.compare(s.index().lastId(), min) >= 0)
&& (max == null || keySupport.compare(s.index().firstId(), max) <= 0));
List<Iterator<Head>> iterators = new ArrayList<>(segments.count());
for (Segment<K, V> segment : segments.allSorted(true))

40
src/java/org/apache/cassandra/journal/Metadata.java
View File

@ -37,23 +37,25 @@ import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
* - total count of records in this segment file
* used for compaction prioritisation
*/
final class Metadata
public final class Metadata
{
private int fsyncLimit;
// Indicates whether a segment needs to be replayed or no.
private volatile boolean needsReplay;
private volatile int recordsCount;
private static final AtomicIntegerFieldUpdater<Metadata> recordsCountUpdater =
AtomicIntegerFieldUpdater.newUpdater(Metadata.class, "recordsCount");
static Metadata empty()
{
return new Metadata(0, 0);
return new Metadata(0, 0, true);
}
private Metadata(int recordsCount, int fsyncLimit)
private Metadata(int recordsCount, int fsyncLimit, boolean needsReplay)
{
this.recordsCount = recordsCount;
this.fsyncLimit = fsyncLimit;
this.needsReplay = needsReplay;
}
void update()
@ -66,17 +68,27 @@ final class Metadata
this.fsyncLimit = fsyncLimit;
}
public void clearNeedsReplay()
{
this.needsReplay = false;
}
int fsyncLimit()
{
return fsyncLimit;
}
public boolean needsReplay()
{
return needsReplay;
}
private void incrementRecordsCount()
{
recordsCountUpdater.incrementAndGet(this);
}
int totalCount()
public int totalCount()
{
return recordsCount;
}
@ -86,8 +98,10 @@ final class Metadata
CRC32 crc = Crc.crc32();
out.writeInt(recordsCount);
out.writeInt(fsyncLimit);
out.writeBoolean(needsReplay);
updateChecksumInt(crc, recordsCount);
updateChecksumInt(crc, fsyncLimit);
updateChecksumInt(crc, needsReplay ? 1 : 0);
out.writeInt((int) crc.getValue());
}
@ -96,10 +110,12 @@ final class Metadata
CRC32 crc = Crc.crc32();
int recordsCount = in.readInt();
int fsyncLimit = in.readInt();
boolean needsReplay = in.readBoolean();
updateChecksumInt(crc, recordsCount);
updateChecksumInt(crc, fsyncLimit);
updateChecksumInt(crc, needsReplay ? 1 : 0);
validateCRC(crc, in.readInt());
return new Metadata(recordsCount, fsyncLimit);
return new Metadata(recordsCount, fsyncLimit, needsReplay);
}
void persist(Descriptor descriptor)
@ -149,7 +165,7 @@ final class Metadata
throw e;
}
return new Metadata(recordsCount, fsyncLimit);
return new Metadata(recordsCount, fsyncLimit, true);
}
static <K> Metadata rebuildAndPersist(Descriptor descriptor, KeySupport<K> keySupport)
@ -160,11 +176,11 @@ final class Metadata
}
@Override
public String toString()
{
public String toString() {
return "Metadata{" +
"fsyncLimit=" + fsyncLimit +
", recordsCount=" + recordsCount +
'}';
"fsyncLimit=" + fsyncLimit +
", needsReplay=" + needsReplay +
", recordsCount=" + recordsCount +
'}';
}
}

1
src/java/org/apache/cassandra/journal/RecordConsumer.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.journal;
import java.nio.ByteBuffer;
// TODO (desired): rename to SerializedRecordConsumer
@FunctionalInterface
public interface RecordConsumer<K>
{

35
src/java/org/apache/cassandra/journal/RecordPointer.java
View File

@ -18,33 +18,29 @@
package org.apache.cassandra.journal;
import com.google.common.primitives.Ints;
import com.google.common.primitives.Longs;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
// TODO: make this available in the accord table as an ID
public class RecordPointer implements Comparable<RecordPointer>
public class RecordPointer extends CommitLogPosition
{
public final long segment; // unique segment id
public final int position; // record start position within the segment
public final int size; // full size of the record
public final int length; // full size of the record
public final long writtenAt; // only set for periodic mode
public RecordPointer(long segment, int position, int size)
public RecordPointer(long segment, int position, int length)
{
this(segment, position, size, 0);
this(segment, position, length, 0);
}
public RecordPointer(long segment, int position, int size, long writtenAt)
public RecordPointer(long segment, int position, int length, long writtenAt)
{
this.segment = segment;
this.position = position;
this.size = size;
super(segment, position);
this.length = length;
this.writtenAt = writtenAt;
}
public RecordPointer(RecordPointer pointer)
{
this(pointer.segment, pointer.position, pointer.size, pointer.writtenAt);
this(pointer.segmentId, pointer.position, pointer.length, pointer.writtenAt);
}
@Override
@ -55,26 +51,19 @@ public class RecordPointer implements Comparable<RecordPointer>
if (!(other instanceof RecordPointer))
return false;
RecordPointer that = (RecordPointer) other;
return this.segment == that.segment
return this.segmentId == that.segmentId
&& this.position == that.position;
}
@Override
public int hashCode()
{
return Long.hashCode(segment) + position * 31;
return Long.hashCode(segmentId) + position * 31;
}
@Override
public String toString()
{
return "(" + segment + ", " + position + ')';
}
@Override
public int compareTo(RecordPointer that)
{
int cmp = Longs.compare(this.segment, that.segment);
return cmp != 0 ? cmp : Ints.compare(this.position, that.position);
return "(" + segmentId + ", " + position + ')';
}
}

9
src/java/org/apache/cassandra/journal/Segment.java
View File

@ -73,11 +73,18 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
abstract ActiveSegment<K, V> asActive();
abstract StaticSegment<K, V> asStatic();
public Metadata metadata()
{
return metadata;
}
public long id()
{
return descriptor.timestamp;
}
public abstract void persistMetadata();
/*
* Reading entries (by id, by offset, iterate)
*/
@ -112,7 +119,7 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
boolean read(RecordPointer pointer, RecordConsumer<K> consumer)
{
EntrySerializer.EntryHolder<K> into = new EntrySerializer.EntryHolder<>();
if (read(pointer.position, pointer.size, into))
if (read(pointer.position, pointer.length, into))
{
consumer.accept(descriptor.timestamp, pointer.position, into.key, into.value, descriptor.userVersion);
return true;

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

@ -129,6 +129,37 @@ public class Segments<K, V>
into.add(segment.asActive());
}
void select(long minTimestamp, long maxTimestamp, Collection<Segment<K, V>> into)
{
List<Segment<K, V>> sorted = allSorted(true);
int idx = minTimestamp == 0 ? 0 : findIdxFor(minTimestamp);
while (idx < sorted.size())
{
Segment<K, V> segment = sorted.get(idx++);
if (segment.descriptor.timestamp > maxTimestamp)
break;
into.add(segment);
}
}
int findIdxFor(long timestamp)
{
List<Segment<K, V>> sorted = allSorted(true);
int low = 0, mid = sorted.size(), high = mid - 1, res = -1;
while (low <= high)
{
mid = (low + high) >>> 1;
res = Long.compare(timestamp, sorted.get(mid).descriptor.timestamp);
if (res > 0)
low = mid + 1;
else if (res == 0)
return mid;
else
high = mid - 1;
}
throw new IllegalStateException(String.format("Could not find a segment with timestamp %d among %s", timestamp, sorted));
}
boolean isSwitched(ActiveSegment<K, V> active)
{
for (Segment<K, V> segment : segments.values())
@ -233,6 +264,7 @@ public class Segments<K, V>
@Override
public String toString()
{
List<Segment<K, V>> sorted = allSorted(true);
return sorted.toString();
}

8
src/java/org/apache/cassandra/journal/StaticSegment.java
View File

@ -33,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.SyncUtil;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.MemoryUtil;
@ -277,6 +278,13 @@ public final class StaticSegment<K, V> extends Segment<K, V>
return this;
}
@Override
public void persistMetadata()
{
metadata.persist(descriptor);
SyncUtil.trySyncDir(descriptor.directory);
}
/**
* Read the entry and specified offset into the entry holder.
* Expects the record to have been written at this offset, but potentially not flushed and lost.

6
src/java/org/apache/cassandra/replication/MutationId.java
View File

@ -46,6 +46,12 @@ public class MutationId extends ShortMutationId
*/
protected final int timestamp;
public MutationId(long logId, int offset, int timestamp)
{
super(logId, offset);
this.timestamp = timestamp;
}
public MutationId(long logId, long sequenceId)
{
super(logId, offset(sequenceId));

188
src/java/org/apache/cassandra/replication/MutationJournal.java
View File

@ -17,31 +17,55 @@
*/
package org.apache.cassandra.replication;
import accord.utils.Invariants;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
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.journal.*;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Semaphore;
import org.jctools.maps.NonBlockingHashMapLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.zip.Checksum;
import static org.apache.cassandra.utils.FBUtilities.getAvailableProcessors;
// TODO (required): handle table truncations
public class MutationJournal
{
public static final MutationJournal instance = new MutationJournal();
private static final Logger log = LoggerFactory.getLogger(MutationJournal.class);
private final Journal<ShortMutationId, Mutation> journal;
private final Map<Long, SegmentStateTracker> segmentStateTrackers;
// Most of the time during write, we will notify last known segment, so we optimistically cache last segment tracker,
// without imposing any visibility guarantees. If we do not see the right segment in this field, we will look it up
// in NBHM.
private SegmentStateTracker lastSegmentTracker;
private MutationJournal()
{
@ -51,7 +75,38 @@ public class MutationJournal
@VisibleForTesting
MutationJournal(File directory, Params params)
{
journal = new Journal<>("MutationJournal", directory, params, new MutationIdSupport(), new MutationSerializer(), SegmentCompactor.noop());
journal = new Journal<>("MutationJournal", directory, params, MutationIdSupport.INSTANCE, MutationSerializer.INSTANCE, SegmentCompactor.noop()) {
@Override
protected void closeActiveSegmentAndOpenAsStatic(ActiveSegment<ShortMutationId, Mutation> activeSegment, Runnable onDone)
{
super.closeActiveSegmentAndOpenAsStatic(activeSegment,
() -> {
maybeCleanupStaticSegment(Invariants.nonNull(getSegment(activeSegment.id())));
if (onDone != null) onDone.run();
});
}
};
segmentStateTrackers = new NonBlockingHashMapLong<>();
}
public CommitLogPosition getCurrentPosition()
{
return journal.currentActiveSegment().currentPosition();
}
// If all Memtables associated with given segment were flushed by the time we have closed active segment
// and opened it as static, mark its metadata to indicate it does not need replay. It may happen that we
// crash before persisting this metadata, in which case we will unnecessarily replay the segment, which
// has no correctness implications.
private void maybeCleanupStaticSegment(Segment<ShortMutationId, Mutation> segment)
{
Invariants.require(segment.isStatic());
SegmentStateTracker tracker = segmentStateTrackers.get(segment.id());
if (tracker != null && tracker.removeCleanFromDirty())
{
segment.metadata().clearNeedsReplay();
segment.persistMetadata();
}
}
public void start()
@ -66,7 +121,40 @@ public class MutationJournal
public RecordPointer write(ShortMutationId id, Mutation mutation)
{
return journal.blockingWrite(id, mutation);
// TODO (required): why are we using blocking write here? We can/should wait for completion on `close` of WriteContext.
RecordPointer ptr = journal.blockingWrite(id, mutation);
// IMPORTANT: there should be no way for mutation to be applied to memtable before we mark it as dirty here,
// since this will introduce a race between marking as dirty and marking as clean.
for (TableId tableId : mutation.getTableIds())
{
SegmentStateTracker tracker = lastSegmentTracker;
if (tracker == null || tracker.segmentId() != ptr.segmentId)
{
tracker = segmentStateTrackers.computeIfAbsent(ptr.segmentId, SegmentStateTracker::new);
lastSegmentTracker = tracker;
}
tracker.markDirty(tableId, ptr);
}
return ptr;
}
/**
* Called by post-flush callback, Memtable is fully flushed to SSTable.
*/
public void notifyFlushed(TableId tableId, CommitLogPosition lowerBound, CommitLogPosition upperBound)
{
for (Segment<ShortMutationId, Mutation> segment : journal.getSegments(lowerBound.segmentId, upperBound.segmentId))
{
Invariants.nonNull(segmentStateTrackers.get(segment.id())).markClean(tableId, lowerBound, upperBound);
// We can only safely mark static segments as non-replayable. Active segment can still be written to,
// so we only persist this metadata on flush.
if (segment.isStatic())
maybeCleanupStaticSegment(segment);
}
}
@Nullable
@ -109,6 +197,89 @@ public class MutationJournal
}
}
public void replayStaticSegments()
{
replay(new DeserializedRecordConsumer<>(MutationSerializer.INSTANCE)
{
@Override
protected void accept(long segmentId, int position, ShortMutationId key, Mutation value)
{
if (Schema.instance.getKeyspaceMetadata(value.getKeyspaceName()) == null)
return;
// TODO: if (commitLogReplayer.pointInTimeExceeded(mutation))
final Keyspace keyspace = Keyspace.open(value.getKeyspaceName());
Mutation.PartitionUpdateCollector newPUCollector = null;
// TODO (required): replayFilter
for (Map.Entry<TableId, PartitionUpdate> e : value.modifications().entrySet())
{
PartitionUpdate update = e.getValue();
update.validate();
if (Schema.instance.getTableMetadata(update.metadata().id) == null)
continue; // dropped
TableId tableId = e.getKey();
// Start segment state tracking
segmentStateTrackers.computeIfAbsent(segmentId, SegmentStateTracker::new)
.markDirty(tableId, segmentId, position);
// TODO (required): shouldReplay
if (newPUCollector == null)
newPUCollector = new Mutation.PartitionUpdateCollector(value.id(), value.getKeyspaceName(), value.key());
newPUCollector.add(update);
// TODO (required): replayedCount
}
if (newPUCollector != null)
{
assert !newPUCollector.isEmpty();
keyspace.apply(newPUCollector.build(), false, true, false);
}
}
}, getAvailableProcessors());
}
@VisibleForTesting
public void replay(DeserializedRecordConsumer<ShortMutationId, Mutation> replayOne, int parallelism)
{
try (Journal<ShortMutationId, Mutation>.StaticSegmentKeyIterator iter =
journal.staticSegmentKeyIterator(s -> s.isStatic()
&& s.metadata().totalCount() > 0
&& s.metadata().needsReplay()))
{
final Semaphore replayParallelism = Semaphore.newSemaphore(parallelism);
final AtomicBoolean abort = new AtomicBoolean();
while (iter.hasNext() && !abort.get())
{
Journal.KeyRefs<ShortMutationId> v = iter.next();
v = v; // Make sure it can not be used in async lambda by accident
ShortMutationId key = v.key();
long lastSegment = v.lastSegment();
// TODO: respect SystemKeyspace.getTruncatedPosition(cfs.metadata.id);
replayParallelism.acquireThrowUncheckedOnInterrupt(1);
Stage.MUTATION.submit(() -> journal.readLast(key, lastSegment, replayOne))
.addCallback(new BiConsumer<Object, Throwable>()
{
@Override
public void accept(Object o, Throwable fail)
{
if (fail != null && !journal.handleError("Could not replay mutation " + key, fail))
abort.set(true);
replayParallelism.release(1);
}
});
}
// Wait for all mutations to be applied before returning
replayParallelism.acquireThrowUncheckedOnInterrupt(parallelism);
}
}
@VisibleForTesting
public void closeCurrentSegmentForTestingIfNonEmpty()
{
journal.closeCurrentSegmentForTestingIfNonEmpty();
}
static class JournalParams implements Params
{
@Override
@ -140,6 +311,12 @@ public class MutationJournal
}
}
@Override
public int compactMaxSegments()
{
return 0;
}
@Override
public ReplayMode replayMode()
{
@ -179,6 +356,8 @@ public class MutationJournal
static class MutationIdSupport implements KeySupport<ShortMutationId>
{
static final MutationIdSupport INSTANCE = new MutationIdSupport();
static final int LOG_ID_OFFSET = 0;
static final int OFFSET_OFFSET = LOG_ID_OFFSET + TypeSizes.LONG_SIZE;
@ -249,17 +428,20 @@ public class MutationJournal
}
}
static class MutationSerializer implements ValueSerializer<ShortMutationId, Mutation>
public static class MutationSerializer implements ValueSerializer<ShortMutationId, Mutation>
{
public static MutationSerializer INSTANCE = new MutationSerializer();
@Override
public void serialize(ShortMutationId id, Mutation mutation, DataOutputPlus out, int userVersion) throws IOException
{
Invariants.require(id.hostId != Integer.MIN_VALUE);
Mutation.serializer.serialize(mutation, out, userVersion);
}
@Override
public Mutation deserialize(ShortMutationId id, DataInputPlus in, int userVersion) throws IOException
{
Invariants.require(id.hostId != Integer.MIN_VALUE);
return Mutation.serializer.deserialize(in, userVersion);
}
}

295
src/java/org/apache/cassandra/replication/SegmentStateTracker.java Normal file
View File

@ -0,0 +1,295 @@
/*
* 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.replication;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import accord.utils.Invariants;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.IntegerInterval;
/**
* Tracks offsets of clean (i.e. memtable->sstable flushed) and dirty (i.e. not yet durably persisted in sstable)
* allocations.
*
* Mutations in segments marked as clean do not need to be replayed.
*
* Tracks which parts of a commit log segment contain unflushed data for each table, and determines when all
* mutations associated with a segment are fully memtable->sstable flushed
*
* Maintains per-table states:
* - "dirty" high mark (bumped when new allocation is made in the segment)
* - "clean" intervals (min/max bounds reported via memtable flushes)
*
* A segment is considered clean when all dirty intervals are covered by clean intervals for every table.
*/
public class SegmentStateTracker
{
final long segmentId;
private final Map<TableId, IntervalState> states = new HashMap<>(32);
private final Lock lock = new ReentrantLock();
public SegmentStateTracker(long segmentId)
{
this.segmentId = segmentId;
}
public long segmentId()
{
return segmentId;
}
@VisibleForTesting
public boolean isClean()
{
removeCleanFromDirty();
lock.lock();
try
{
return states.isEmpty();
}
finally
{
lock.unlock();
}
}
/**
* Should be called _only_ for a static segment to ensure there can be no way for interval state
* to go back from clean to dirty.
*
* Removes all clean (i.e. memtable -> sstable flushed) from dirty interval. If metadata tracking for all intervals of all tables
* are clean, returns true. False otherwise.
*/
public boolean removeCleanFromDirty()
{
List<Map.Entry<TableId, IntervalState>> states;
// Take a "snapshot" of states, while holding a lock
lock.lock();
try
{
states = new ArrayList<>(this.states.entrySet());
}
finally
{
lock.unlock();
}
int[] remove = new int[states.size()];
int removeCount = 0;
// Check if any of the remaining items can be cleaned up, without holding a lock
for (int i = 0; i < states.size(); i++)
{
IntervalState state = states.get(i).getValue();
if (!state.isDirty())
remove[removeCount++] = i;
}
// Remove all fully covered items, while holding a lock
if (removeCount > 0)
{
lock.lock();
try
{
if (this.states.size() == removeCount)
{
this.states.clear();
return true;
}
for (int i = 0; i < removeCount; i++)
{
Map.Entry<TableId, IntervalState> e = states.get(remove[i]);
this.states.remove(e.getKey());
}
}
finally
{
lock.unlock();
}
}
return false;
}
public void markDirty(TableId tableId, CommitLogPosition ptr)
{
markDirty(tableId, ptr.segmentId, ptr.position);
}
public void markDirty(TableId tableId, long segmentId, int position)
{
Invariants.require(segmentId == this.segmentId);
IntervalState state;
lock.lock();
try
{
state = states.computeIfAbsent(tableId, (k) -> {
// Initialize with given position as both low and high bound to ensure we correctly set
// lower bound when marking as clean
return new IntervalState(position, position);
});
}
finally
{
lock.unlock();
}
state.markDirty(position);
}
public void markClean(TableId tableId, CommitLogPosition lowerBound, CommitLogPosition upperBound)
{
Invariants.require(lowerBound.compareTo(upperBound) <= 0, "%s should be smaller than %s", lowerBound, upperBound);
if (lowerBound.segmentId > segmentId || upperBound.segmentId < segmentId)
return;
IntervalState state;
lock.lock();
try
{
state = states.get(tableId);
}
finally
{
lock.unlock();
}
if (state != null)
{
// TODO (required): test this logic
// Only mark clean ranges for _this_ segment
int lower = lowerBound.segmentId == segmentId ? lowerBound.position : 0;
int upper = upperBound.segmentId == segmentId ? upperBound.position : Integer.MAX_VALUE;
state.markClean(lower, upper);
}
}
private static class IntervalState
{
static final long[] EMPTY = new long[0];
// dirty interval in this segment; if interval is not covered by the clean set, the log contains unflushed data
volatile long dirty;
// clean intervals; separate map from above to permit marking Cfs clean whilst the log is still in use
volatile long[] clean = EMPTY;
private static final AtomicLongFieldUpdater<IntervalState> dirtyUpdater = AtomicLongFieldUpdater.newUpdater (IntervalState.class, "dirty");
private static final AtomicReferenceFieldUpdater<IntervalState, long[]> cleanUpdater = AtomicReferenceFieldUpdater.newUpdater(IntervalState.class, long[].class, "clean");
public IntervalState(int lower, int upper)
{
this(make(lower, upper));
}
private IntervalState(long dirty)
{
this.dirty = dirty;
}
public void markClean(int start, int end)
{
long[] prev;
long[] next;
do
{
prev = this.clean;
next = IntegerInterval.Set.add(prev, start, end);
}
while (!cleanUpdater.compareAndSet(this, prev, next));
}
public boolean isDirty()
{
long[] clean = this.clean;
long dirty = this.dirty;
return !IntegerInterval.Set.covers(clean, lower(dirty), upper(dirty));
}
/**
* Expands the interval to cover the given value by extending one of its sides if necessary.
* Mutates this. Thread-safe.
*/
public void markDirty(int value)
{
long prev;
int lower;
int upper;
do
{
prev = dirty;
upper = upper(prev);
lower = lower(prev);
if (value > upper) // common case
upper = value;
else if (value < lower)
lower = value;
}
while (!dirtyUpdater.compareAndSet(this, prev, make(lower, upper)));
}
public String toString()
{
long dirty = this.dirty;
long[] clean = this.clean;
StringBuilder sb = new StringBuilder();
for (int i = 0; i < clean.length; i++)
{
long l = clean[i];
if (i > 0)
sb.append(',');
sb.append('[').append(lower(l)).append(',').append(upper(l)).append("]");
}
return "dirty:[" + lower(dirty) + ',' + upper(dirty) + "], clean:[" + sb + "]";
}
private static long make(int lower, int upper)
{
assert lower <= upper;
return ((lower & 0xFFFFFFFFL) << 32) | upper & 0xFFFFFFFFL;
}
private static int lower(long interval)
{
return (int) (interval >>> 32);
}
private static int upper(long interval)
{
return (int) interval;
}
}
@Override
public String toString() {
return "DefaultStateTracker{" +
"segmentId=" + segmentId +
", states=" + states +
'}';
}
}

1
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -361,6 +361,7 @@ public class CassandraDaemon
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
MutationJournal.instance.replayStaticSegments();
}
catch (IOException e)
{

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

@ -371,7 +371,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
&& diff.after.route() != null)
journal.onDurable(pointer, () ->
journalTable.safeNotify(index ->
index.update(pointer.segment, key.commandStoreId, key.id, diff.after.route())));
index.update(pointer.segmentId, key.commandStoreId, key.id, diff.after.route())));
if (onFlush != null)
journal.onDurable(pointer, onFlush);
}

2
src/java/org/apache/cassandra/tcm/Startup.java
View File

@ -53,6 +53,7 @@ import org.apache.cassandra.gms.NewGossiper;
import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.KeyspaceMetadata;
@ -302,6 +303,7 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
try
{
CommitLog.instance.recoverSegmentsOnDisk();
MutationJournal.instance.replayStaticSegments();
}
catch (IOException e)
{

32
src/java/org/apache/cassandra/utils/IntegerInterval.java
View File

@ -23,9 +23,12 @@ import java.util.Collection;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.stream.Collectors;
import accord.utils.Invariants;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import net.nicoulaj.compilecommand.annotations.Inline;
/**
* Mutable integer interval class, thread-safe.
* Represents the interval [lower,upper].
@ -33,7 +36,7 @@ import com.google.common.primitives.Longs;
public class IntegerInterval
{
volatile long interval;
private static AtomicLongFieldUpdater<IntegerInterval> intervalUpdater =
private static final AtomicLongFieldUpdater<IntegerInterval> intervalUpdater =
AtomicLongFieldUpdater.newUpdater(IntegerInterval.class, "interval");
private IntegerInterval(long interval)
@ -127,7 +130,7 @@ public class IntegerInterval
*/
public static class Set
{
static long[] EMPTY = new long[0];
static final long[] EMPTY = new long[0];
private volatile long[] ranges = EMPTY;
@ -136,11 +139,15 @@ public class IntegerInterval
*/
public synchronized void add(int start, int end)
{
assert start <= end;
long[] ranges, newRanges;
{
ranges = this.ranges; // take local copy to avoid risk of it changing in the midst of operation
this.ranges = add(this.ranges, start, end);
}
@Inline
public static long[] add(long[] ranges, int start, int end)
{
Invariants.require(start <= end, "Start (%d) should be less than or equal to end (%d)", start, end);
long[] newRanges;
{
// extend ourselves to cover any ranges we overlap
// record directly preceding our end may extend past us, so take the max of our end and its
int rpos = Arrays.binarySearch(ranges, ((end & 0xFFFFFFFFL) << 32) | 0xFFFFFFFFL); // floor (i.e. greatest <=) of the end position
@ -175,7 +182,7 @@ public class IntegerInterval
for (int i = rpos + 1; i < ranges.length; ++i)
newRanges[dest++] = ranges[i];
}
this.ranges = newRanges;
return newRanges;
}
/**
@ -192,8 +199,13 @@ public class IntegerInterval
*/
public boolean covers(int start, int end)
{
long[] ranges = this.ranges; // take local copy to avoid risk of it changing in the midst of operation
int rpos = Arrays.binarySearch(ranges, ((start & 0xFFFFFFFFL) << 32) | 0xFFFFFFFFL); // floor (i.e. greatest <=) of the end position
return covers(this.ranges, start, end);
}
@Inline
public static boolean covers(long[] ranges, int start, int end)
{
int rpos = Arrays.binarySearch(ranges, ((start & 0xFFFFFFFFL) << 32) | 0xFFFFFFFFL); // floor (i.e. greatest <=) of the end position
if (rpos < 0)
rpos = (-1 - rpos) - 1;
if (rpos == -1)
@ -220,7 +232,7 @@ public class IntegerInterval
public Collection<IntegerInterval> intervals()
{
return Lists.transform(Longs.asList(ranges), iv -> new IntegerInterval(iv));
return Lists.transform(Longs.asList(ranges), IntegerInterval::new);
}
@Override

1
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -840,6 +840,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
NodeId self = ClusterMetadata.current().myNodeId();
if (self != null)
AccordService.localStartup(self);
MutationJournal.instance.replayStaticSegments();
}
catch (IOException e)
{

92
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounceTest.java Normal file
View File

@ -0,0 +1,92 @@
/*
* 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.distributed.test.tracking;
import java.util.ArrayList;
import java.util.List;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.shared.ClusterUtils;
import org.apache.cassandra.distributed.test.log.FuzzTestBase;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.dsl.HistoryBuilder;
import org.apache.cassandra.harry.dsl.ReplayingHistoryBuilder;
import org.apache.cassandra.harry.execution.InJvmDTestVisitExecutor;
import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.SchemaGenerators;
import org.apache.cassandra.replication.MutationJournal;
import org.junit.Test;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
public class MutationTrackingBounceTest extends FuzzTestBase
{
private static final int POPULATION = 1000;
@Test
public void bounceTest() throws Throwable
{
try (Cluster cluster = builder().withNodes(1).start())
{
int tables = 10;
int writesPerKey = 2;
int pks = 100;
withRandom(rng -> {
cluster.schemaChange(String.format("CREATE KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1} " +
"AND replication_type='tracked'",
KEYSPACE));
List<HistoryBuilder> builders = new ArrayList<>();
for (int i = 0; i < tables; i++)
{
Generator<SchemaSpec> schemaGen = SchemaGenerators.trivialSchema(KEYSPACE, () -> "mutation_tracking_bounce_" + (builders.size() + 1), POPULATION,
SchemaSpec.optionsBuilder());
SchemaSpec schema = schemaGen.generate(rng);
cluster.schemaChange(schema.compile());
builders.add(new ReplayingHistoryBuilder(schema.valueGenerators,
hb -> InJvmDTestVisitExecutor.builder()
.consistencyLevel(ConsistencyLevel.QUORUM)
.build(schema, hb, cluster)));
}
int counter = 0;
for (int pk = 0; pk < pks; pk++) {
for (HistoryBuilder history : builders)
for (int i = 0; i < writesPerKey; i++)
history.insert(pk);
if (++counter % 10 == 0)
cluster.get(1).runOnInstance(() -> MutationJournal.instance.closeCurrentSegmentForTestingIfNonEmpty());
}
ClusterUtils.stopUnchecked(cluster.get(1));
cluster.get(1).startup();
for (int pk = 0; pk < pks; pk++)
for (HistoryBuilder history : builders)
for (int i = 0; i < 10; i++)
history.selectPartition(pk);
cluster.get(1).runOnInstance(new MutationTrackingBounce_ValidateRunnable(tables * pks * writesPerKey));
});
}
}
}

69
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingBounce_ValidateRunnable.java Normal file
View File

@ -0,0 +1,69 @@
/*
* 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.distributed.test.tracking;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.journal.DeserializedRecordConsumer;
import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.replication.ShortMutationId;
import org.junit.Assert;
// Separate class for MutationTrackingBounceTest, since without it we were getting non-serializable class exceptions, likely due to static fields
public class MutationTrackingBounce_ValidateRunnable implements IIsolatedExecutor.SerializableRunnable
{
private final int count;
public MutationTrackingBounce_ValidateRunnable(int expectedMutations)
{
this.count = expectedMutations;
}
@Override
public void run()
{
AtomicInteger counter = new AtomicInteger();
MutationJournal.instance.replay(new DeserializedRecordConsumer<>(MutationJournal.MutationSerializer.INSTANCE)
{
Set<ShortMutationId> seen = new HashSet<>();
@Override
protected void accept(long segment, int position, ShortMutationId key, Mutation mutation)
{
if (!seen.add(key))
throw new AssertionError(String.format("Should have witnessed each key just once, but seen %s already", key));
for (PartitionUpdate partitionUpdate : mutation.getPartitionUpdates())
{
if (!MutationTrackingService.instance.createSummaryForKey(partitionUpdate.partitionKey(), partitionUpdate.metadata().id, false)
.contains(key))
{
throw new AssertionError(String.format("Mutation %s should have been witnessed (%s)", mutation, key));
}
}
counter.incrementAndGet();
}
}, 1);
Assert.assertEquals(count, counter.get());
}
}

15
test/harry/main/org/apache/cassandra/harry/checker/TestHelper.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.harry.checker;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -47,6 +48,20 @@ public class TestHelper
}
}
public static void withRandom(long seed, ModelChecker.ThrowingConsumer<EntropySource> rng, Consumer<Throwable> onError)
{
try
{
logger.info("Seed: {}", seed);
rng.accept(new JdkRandomEntropySource(seed));
}
catch (Throwable t)
{
onError.accept(t);
throw new AssertionError(String.format("Caught an exception at seed:%dL", seed), t);
}
}
public static void repeat(int num, ExecUtil.ThrowingSerializableRunnable<?> r)
{
for (int i = 0; i < num; i++)

141
test/unit/org/apache/cassandra/journal/SegmentsTest.java Normal file
View File

@ -0,0 +1,141 @@
/*
* 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.journal;
import java.io.File;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import org.apache.cassandra.utils.concurrent.Ref;
import org.junit.Test;
import static org.apache.cassandra.harry.checker.TestHelper.withRandom;
public class SegmentsTest
{
@Test
public void testSelect()
{
withRandom(0l, rng -> {
// Create mock segments with different timestamps
java.io.File file = File.createTempFile("segments", "test");
List<Segment<String, String>> segmentList = new ArrayList<>();
Set<Long> taken = new HashSet<>();
for (int i = 0; i < 100; i++)
{
while (true)
{
long id = rng.nextLong(0, 10_000);
if (taken.add(id))
{
segmentList.add(new TestSegment<>(file, id));
break;
}
}
}
segmentList.sort(Comparator.comparing(s -> s.descriptor.timestamp));
Segments<String, String> segments = Segments.of(segmentList);
for (int i = 0; i < 10_000; i++)
{
// Generate two distinct segment idxs
int i1 = rng.nextInt(segmentList.size());
int i2;
do
{
i2 = rng.nextInt(segmentList.size());
}
while (i2 == i1);
int min = Math.min(i1, i2);
int max = Math.max(i1, i2);
List<Segment<String, String>> selected = new ArrayList<>();
segments.select(segmentList.get(min).id(),
segmentList.get(max).id(),
selected);
List<Segment<String, String>> expected = segmentList.subList(min, max + 1);
if (!Objects.equals(expected, selected))
{
throw new AssertionError(String.format("\nExpected: %s\n" +
"Selected: %s",
expected,
selected));
}
}
});
}
private static class TestSegment<K, V> extends Segment<K, V>
{
TestSegment(File dir, long timestamp)
{
super(Descriptor.create(new org.apache.cassandra.io.util.File(dir), timestamp, 1), null, null);
}
@Override
void close(Journal<K, V> journal)
{
}
@Override
public boolean isActive()
{
return false;
}
@Override public boolean isStatic()
{
return false;
}
@Override
boolean isEmpty()
{
return false;
}
@Override Index<K> index() { throw new UnsupportedOperationException(); }
@Override boolean isFlushed(long position) { throw new UnsupportedOperationException(); }
@Override public void persistMetadata() { throw new UnsupportedOperationException(); }
@Override boolean read(int offset, int size, EntrySerializer.EntryHolder<K> into) { throw new UnsupportedOperationException(); }
@Override public ActiveSegment<K, V> asActive() { throw new UnsupportedOperationException(); }
@Override public StaticSegment<K, V> asStatic() { throw new UnsupportedOperationException(); }
@Override public Ref<Segment<K, V>> selfRef() { throw new UnsupportedOperationException(); }
@Override public Ref<Segment<K, V>> tryRef(){ throw new UnsupportedOperationException(); }
@Override public Ref<Segment<K, V>> ref(){ throw new UnsupportedOperationException(); }
@Override
public String toString()
{
return "TestSegment{" +
"id=" + descriptor.timestamp +
'}';
}
@Override
public boolean equals(Object obj)
{
TestSegment<K, V> other = (TestSegment<K, V>) obj;
return descriptor.equals(other.descriptor);
}
}
}

218
test/unit/org/apache/cassandra/replication/MutationJournalReplayTest.java Normal file
View File

@ -0,0 +1,218 @@
/*
* 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.replication;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.RowUpdateBuilder;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.harry.checker.TestHelper;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.DeserializedRecordConsumer;
import org.apache.cassandra.journal.TestParams;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.ReplicationType;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tools.FieldUtil;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import static org.apache.cassandra.replication.MutationJournalTest.assertMutationsEqual;
public class MutationJournalReplayTest
{
private static final String KEYSPACE = "ks";
private static final String TABLE_PREFIX = "tbl";
private static final TableMetadata[] TABLES = new TableMetadata[10];
private static MutationJournal journal;
@BeforeClass
public static void setUp() throws IOException
{
SchemaLoader.prepareServer();
File directory = new File(Files.createTempDirectory("mutation-journal-replay-test"));
directory.deleteRecursiveOnExit();
journal = new MutationJournal(directory, new TestParams(MessagingService.current_version)
{
@Override
public long flushPeriod(TimeUnit units)
{
return 1;
}
@Override
public FlushMode flushMode()
{
return FlushMode.PERIODIC;
}
});
FieldUtil.setInstanceUnsafe(MutationJournal.class, journal, "instance");
journal.start();
for (int i = 0; i < TABLES.length; i++)
{
TABLES[i] = TableMetadata.builder(KEYSPACE, String.format("%s_%d", TABLE_PREFIX, i))
.keyspaceReplicationType(ReplicationType.tracked)
.addPartitionKeyColumn("pk", UTF8Type.instance)
.addClusteringColumn("ck", UTF8Type.instance)
.addRegularColumn("value", UTF8Type.instance)
.build();
}
SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(1, ReplicationType.tracked), TABLES);
}
@AfterClass
public static void tearDown()
{
journal.shutdownBlocking();
}
@Test
public void testReplay() throws Throwable
{
long seed = 0l;
TestHelper.withRandom(seed,
rng -> {
List<Mutation> original = new ArrayList<>();
for (int i = 1; i <= 10_000; i++)
{
MutationId id = new MutationId(100L, i, i);
Mutation mutation = mutation(i % 10, i).withMutationId(id);
journal.write(id, mutation);
original.add(mutation);
if (i % rng.nextInt(1, 100) > 90)
journal.closeCurrentSegmentForTestingIfNonEmpty();
}
journal.closeCurrentSegmentForTestingIfNonEmpty();
List<Mutation> replayed = new ArrayList<>();
journal.replay(new DeserializedRecordConsumer<ShortMutationId, Mutation>(MutationJournal.MutationSerializer.INSTANCE)
{
@Override
protected void accept(long segment, int position, ShortMutationId key, Mutation mutation)
{
replayed.add(mutation);
}
}, 1);
assertMutationsEqual(original, replayed);
});
}
@Test
public void testReplayFlushed() throws Throwable
{
long seed = 0l;
class Bounds
{
final CommitLogPosition start;
final CommitLogPosition end;
final int count;
Bounds(CommitLogPosition start, CommitLogPosition end, int count)
{
this.start = start;
this.end = end;
this.count = count;
}
}
TestHelper.withRandom(seed,
rng -> {
List<Mutation> original = new ArrayList<>();
List<Bounds> testFlushBounds = new ArrayList<>();
CommitLogPosition prevPos = journal.getCurrentPosition();
int count = 0;
for (int i = 1; i <= 1000; i++)
{
MutationId id = new MutationId(100L, i, i);
Mutation mutation = mutation(i % TABLES.length, i).withMutationId(id);
journal.write(id, mutation);
count++;
original.add(mutation);
if (i % rng.nextInt(1, 100) > 90)
{
CommitLogPosition curPos = journal.getCurrentPosition();
journal.closeCurrentSegmentForTestingIfNonEmpty();
testFlushBounds.add(new Bounds(prevPos, curPos, count));
count = 0;
prevPos = curPos;
}
}
journal.closeCurrentSegmentForTestingIfNonEmpty();
int flushed = 0;
for (Bounds bounds : testFlushBounds)
{
if (rng.nextBoolean())
{
for (TableMetadata table : TABLES)
journal.notifyFlushed(table.id, bounds.start, bounds.end);
flushed += bounds.count;
}
}
List<Mutation> replayed = new ArrayList<>();
journal.replay(new DeserializedRecordConsumer<ShortMutationId, Mutation>(MutationJournal.MutationSerializer.INSTANCE)
{
@Override
protected void accept(long segment, int position, ShortMutationId key, Mutation mutation)
{
replayed.add(mutation);
}
}, 1);
Assert.assertEquals(original.size() - flushed,
replayed.size());
});
}
private static String CACHED_STRING = null;
private static Mutation mutation(int table, int value)
{
if (CACHED_STRING == null)
{
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 512; i++)
{
sb.append('.');
}
CACHED_STRING = sb.toString();
}
return new RowUpdateBuilder(TABLES[table], 0, "key_" + value)
.clustering("ck")
.add("value", CACHED_STRING)
.build();
}
}

9
test/unit/org/apache/cassandra/replication/MutationJournalTest.java
View File

@ -127,19 +127,20 @@ public class MutationJournalTest
assertMutationsEqual(expected, actual);
}
private static void assertMutationEquals(Mutation expected, Mutation actual)
public static void assertMutationEquals(Mutation expected, Mutation actual)
{
assertEquals(serialize(expected), serialize(actual));
if (!serialize(expected).equals(serialize(actual)))
throw new AssertionError(String.format("Expected %s but got %s", expected, actual));
}
private static void assertMutationsEqual(List<Mutation> expected, List<Mutation> actual)
public static void assertMutationsEqual(List<Mutation> expected, List<Mutation> actual)
{
assertEquals(expected.size(), actual.size());
for (int i = 0; i < expected.size(); i++)
assertMutationEquals(expected.get(i), actual.get(i));
}
private static ByteBuffer serialize(Mutation mutation)
public static ByteBuffer serialize(Mutation mutation)
{
try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get())
{

237
test/unit/org/apache/cassandra/replication/SegmentStateTrackerTest.java Normal file
View File

@ -0,0 +1,237 @@
/*
* 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.replication;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.cassandra.db.commitlog.CommitLogPosition;
import org.apache.cassandra.harry.checker.TestHelper;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.Pair;
import org.junit.Assert;
import org.junit.Test;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class SegmentStateTrackerTest
{
@Test
public void trivialTest()
{
long segment = 123L;
SegmentStateTracker tracker = new SegmentStateTracker(segment);
TableId tbl = TableId.generate();
assertTrue(tracker.isClean());
for (int i = 1; i <= 10; i++)
{
tracker.markDirty(tbl, segment, i);
assertFalse(tracker.isClean());
if (i % 5 == 0)
{
tracker.markClean(tbl, pos(segment, i - 5), pos(segment, i));
assertTrue(tracker.isClean());
}
}
assertTrue(tracker.isClean());
}
@Test
public void memtableSpannignMultipleSegmentsTest()
{
SegmentStateTracker segment1 = new SegmentStateTracker(1);
SegmentStateTracker segment2 = new SegmentStateTracker(2);
TableId tbl = TableId.fromLong(1);
segment1.markDirty(tbl, segment1.segmentId(), 10);
segment2.markDirty(tbl, segment2.segmentId(), 10);
segment2.markDirty(tbl, segment2.segmentId(), 20);
segment1.markClean(tbl, pos(segment1.segmentId(), 10), pos(segment2.segmentId(), 10)); // first flush
segment2.markClean(tbl, pos(segment1.segmentId(), 10), pos(segment2.segmentId(), 10)); // first flush
segment2.markClean(tbl, pos(segment2.segmentId(), 10), pos(segment2.segmentId(), 30)); // second flush
Assert.assertTrue(segment1.isClean());
Assert.assertTrue(segment2.isClean());
}
@Test
public void fuzzTest()
{
int allocationCount = 10_000;
long seed = 0;
List<String> ops = null; // new ArrayList<>(); // to enable operation logging
AtomicLong tableIdGen = new AtomicLong();
TestHelper.withRandom(seed,
rng -> {
// This test tracks state from the perspective of state tracker:
// We make allocations by bumping the pointer, reporting allocations to different tables.
// Table can flush subsequent allocations, and report information back to the segment.
class Table
{
final TableId tableId = TableId.fromLong(tableIdGen.getAndIncrement());
// Track all allocations to a cfid in a Table, and perform contiguous "flushes"
// that perform position bounds to tracker
ArrayList<CommitLogPosition> unflushedAllocations = new ArrayList<>();
// Memtables _always_ report contiguous chunks of metadata, see CFS$Flush, new memtable
// will always be created with commitLogUpperBound of the previous one.
CommitLogPosition lastFlushMax = null;
void addAllocation(CommitLogPosition pos)
{
if (ops != null) ops.add("Allocate " + pos + " in " + tableId);
unflushedAllocations.add(pos);
}
void flush(Collection<SegmentStateTracker> trackers)
{
if (unflushedAllocations.isEmpty())
return;
CommitLogPosition min = lastFlushMax == null ? unflushedAllocations.get(0) : lastFlushMax;
CommitLogPosition max = unflushedAllocations.get(rng.nextInt(unflushedAllocations.size()));
lastFlushMax = max;
if (ops != null) ops.add(String.format("Flush %s [%s, %s]", tableId, min, max));
for (SegmentStateTracker tracker : trackers)
reportFlushed(tracker, min, max);
// TODO (required): use array an copying instead
unflushedAllocations.removeIf(alloc -> alloc.compareTo(min) >= 0 && alloc.compareTo(max) <= 0);
}
boolean hasUnflushed()
{
return !unflushedAllocations.isEmpty();
}
boolean hasUnflushedFor(long segment)
{
for (CommitLogPosition alloc : unflushedAllocations)
{
if (alloc.segmentId == segment)
return true;
}
return false;
}
List<CommitLogPosition> getUnflushedFor(long segment)
{
List<CommitLogPosition> unflushed = new ArrayList<>();
for (CommitLogPosition alloc : unflushedAllocations)
{
if (alloc.segmentId == segment)
unflushed.add(alloc);
}
return unflushed;
}
void reportFlushed(SegmentStateTracker tracker, CommitLogPosition minBound, CommitLogPosition maxBound)
{
if (tracker.segmentId() >= minBound.segmentId && tracker.segmentId() <= maxBound.segmentId)
tracker.markClean(tableId, minBound, maxBound);
}
}
int tableCount = 10;
List<Table> tables = new ArrayList<>(tableCount);
for (int i = 0; i < tableCount; i++)
tables.add(new Table());
Map<Long, SegmentStateTracker> segments = new HashMap<>();
Runnable validateAllSegments = () -> {
for (SegmentStateTracker segment : segments.values())
{
boolean segmentIsClean = segment.isClean();
boolean allTablesFlushed = tables.stream().noneMatch(t -> t.hasUnflushedFor(segment.segmentId()));
if (segmentIsClean != allTablesFlushed)
throw new IllegalArgumentException(String.format("Segment is %sclean, but table has %sunflushed allocations:\n%s\n%s",
segmentIsClean ? "" : "not ",
allTablesFlushed ? "" : "no ",
segment,
tables.stream().map(t -> Pair.create(t.tableId, t.getUnflushedFor(segment.segmentId())))
.filter(t -> !t.right.isEmpty())
.collect(Collectors.toList())
));
}
};
SegmentStateTracker currentSegment = null;
int currentSegmentOffset = 0;
for (int i = 0; i < allocationCount; i++)
{
if (i > 0 && i % 50 == 0)
{
for (int j = 0; j < 3; j++)
{
Table table = tables.get(rng.nextInt(tableCount));
table.flush(segments.values());
validateAllSegments.run();
}
}
if (i % 100 == 0)
{
currentSegment = new SegmentStateTracker(segments.size());
currentSegmentOffset = 0;
segments.put(currentSegment.segmentId(), currentSegment);
}
int size = rng.nextInt(100);
Table table = tables.get(rng.nextInt(tableCount));
CommitLogPosition pos = pos(currentSegment.segmentId(), currentSegmentOffset);
table.addAllocation(pos);
currentSegment.markDirty(table.tableId, pos);
currentSegmentOffset += size;
validateAllSegments.run();
}
while (tables.stream().anyMatch(Table::hasUnflushed))
{
Table table = tables.get(rng.nextInt(tableCount));
if (table.hasUnflushed())
table.flush(segments.values());
validateAllSegments.run();
}
},
e -> {
if (ops != null)
{
System.out.println("History: ");
for (String op : ops)
System.out.println(op);
}
});
}
public static CommitLogPosition pos(long segmentId, int pos)
{
return new CommitLogPosition(segmentId, pos);
}
}