import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
View(List<Memtable> liveMemtables, List<Memtable> flushingMemtables, Map<SSTableReader, SSTableReader> sstables, Map<SSTableReader, SSTableReader> compacting, SSTableIntervalTree intervalTree)
{
    assert liveMemtables != null;
    assert flushingMemtables != null;
    assert sstables != null;
    assert compacting != null;
    assert intervalTree != null;

    this.liveMemtables = liveMemtables;
    this.flushingMemtables = flushingMemtables;

    this.sstablesMap = sstables;
    this.sstables = sstablesMap.keySet();
    this.compactingMap = compacting;
    this.compacting = compactingMap.keySet();
    this.intervalTree = intervalTree;
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
static Function<View, View> markFlushing(final Memtable toFlush)
{
    return new Function<View, View>()
    {
        public View apply(View view)
        {
            List<Memtable> live = view.liveMemtables, flushing = view.flushingMemtables;
            List<Memtable> newLive = copyOf(filter(live, not(equalTo(toFlush))));
            List<Memtable> newFlushing = copyOf(concat(filter(flushing, lessThan(toFlush)),
                                                       of(toFlush),
                                                       filter(flushing, not(lessThan(toFlush)))));
            assert newLive.size() == live.size() - 1;
            assert newFlushing.size() == flushing.size() + 1;
            return new View(newLive, newFlushing, view.sstablesMap, view.compactingMap, view.intervalTree);
        }
    };
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
static Function<View, View> replaceFlushed(final Memtable memtable, final Collection<SSTableReader> flushed)
{
    return new Function<View, View>()
    {
        public View apply(View view)
        {
            List<Memtable> flushingMemtables = copyOf(filter(view.flushingMemtables, not(equalTo(memtable))));
            assert flushingMemtables.size() == view.flushingMemtables.size() - 1;

            if (flushed == null || flushed.isEmpty())
                return new View(view.liveMemtables, flushingMemtables, view.sstablesMap,
                                view.compactingMap, view.intervalTree);

            Map<SSTableReader, SSTableReader> sstableMap = replace(view.sstablesMap, emptySet(), flushed);
            return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap,
                            SSTableIntervalTree.build(sstableMap.keySet()));
        }
    };
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * get the Memtable that the ordered writeOp should be directed to
 */
public Memtable getMemtableFor(OpOrder.Group opGroup, ReplayPosition replayPosition)
{
    // since any new memtables appended to the list after we fetch it will be for operations started
    // after us, we can safely assume that we will always find the memtable that 'accepts' us;
    // if the barrier for any memtable is set whilst we are reading the list, it must accept us.

    // there may be multiple memtables in the list that would 'accept' us, however we only ever choose
    // the oldest such memtable, as accepts() only prevents us falling behind (i.e. ensures we don't
    // assign operations to a memtable that was retired/queued before we started)
    for (Memtable memtable : view.get().liveMemtables)
    {
        if (memtable.accepts(opGroup, replayPosition))
            return memtable;
    }
    throw new AssertionError(view.get().liveMemtables.toString());
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * @return the largest timestamp before which it's okay to drop tombstones for the given partition;
 * i.e., after the maxPurgeableTimestamp there may exist newer data that still needs to be suppressed
 * in other sstables.  This returns the minimum timestamp for any SSTable that contains this partition and is not
 * participating in this compaction, or memtable that contains this partition,
 * or LONG.MAX_VALUE if no SSTable or memtable exist.
 */
public long maxPurgeableTimestamp(DecoratedKey key)
{
    if (!compactingRepaired() || NEVER_PURGE_TOMBSTONES)
        return Long.MIN_VALUE;

    long min = Long.MAX_VALUE;
    overlapIterator.update(key);
    for (SSTableReader sstable : overlapIterator.overlaps())
    {
        // if we don't have bloom filter(bf_fp_chance=1.0 or filter file is missing),
        // we check index file instead.
        if (sstable.getBloomFilter() instanceof AlwaysPresentFilter && sstable.getPosition(key, SSTableReader.Operator.EQ, false) != null)
            min = Math.min(min, sstable.getMinTimestamp());
        else if (sstable.getBloomFilter().isPresent(key))
            min = Math.min(min, sstable.getMinTimestamp());
    }

    for (Memtable memtable : cfs.getTracker().getView().getAllMemtables())
    {
        Partition partition = memtable.getPartition(key);
        if (partition != null)
            min = Math.min(min, partition.stats().minTimestamp);
    }
    return min;
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * Handle a flushed memtable.
 *
 * @param memtable the flushed memtable
 * @param sstable the written sstable. can be null if the memtable was clean.
 */
public void replaceFlushed(Memtable memtable, SSTableReader sstable)
{
    cfs.getDataTracker().replaceFlushed(memtable, sstable);
    if (sstable != null)
        CompactionManager.instance.submitBackground(cfs);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
static Function<View, View> switchMemtable(final Memtable newMemtable)
{
    return new Function<View, View>()
    {
        public View apply(View view)
        {
            List<Memtable> newLive = ImmutableList.<Memtable>builder().addAll(view.liveMemtables).add(newMemtable).build();
            assert newLive.size() == view.liveMemtables.size() + 1;
            return new View(newLive, view.flushingMemtables, view.sstablesMap, view.compactingMap, view.intervalTree);
        }
    };
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/** (Re)initializes the tracker, purging all references. */
@VisibleForTesting
public void reset()
{
    view.set(new View(
                     !isDummy() ? ImmutableList.of(new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), cfstore))
                                : ImmutableList.<Memtable>of(),
                     ImmutableList.<Memtable>of(),
                     Collections.<SSTableReader, SSTableReader>emptyMap(),
                     Collections.<SSTableReader, SSTableReader>emptyMap(),
                     SSTableIntervalTree.empty()));
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * Switch the current memtable. This atomically appends a new memtable to the end of the list of active memtables,
 * returning the previously last memtable. It leaves the previous Memtable in the list of live memtables until
 * discarding(memtable) is called. These two methods must be synchronized/paired, i.e. m = switchMemtable
 * must be followed by discarding(m), they cannot be interleaved.
 *
 * @return the previously active memtable
 */
public Memtable switchMemtable(boolean truncating, Memtable newMemtable)
{
    Pair<View, View> result = apply(View.switchMemtable(newMemtable));
    if (truncating)
        notifyRenewed(newMemtable);

    return result.left.getCurrentMemtable();
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public void replaceFlushed(Memtable memtable, Collection<SSTableReader> sstables)
{
    assert !isDummy();
    if (sstables.isEmpty())
    {
        // sstable may be null if we flushed batchlog and nothing needed to be retained
        // if it's null, we don't care what state the cfstore is in, we just replace it and continue
        apply(View.replaceFlushed(memtable, null));
        return;
    }

    sstables.forEach(SSTableReader::setupOnline);
    // back up before creating a new Snapshot (which makes the new one eligible for compaction)
    maybeIncrementallyBackup(sstables);

    apply(View.replaceFlushed(memtable, sstables));

    Throwable fail;
    fail = updateSizeTracking(emptySet(), sstables, null);
    // TODO: if we're invalidated, should we notifyadded AND removed, or just skip both?
    fail = notifyAdded(sstables, fail);

    if (!isDummy() && !cfstore.isValid())
        dropSSTables();

    maybeFail(fail);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * Handle a flushed memtable.
 *
 * @param memtable the flushed memtable
 * @param sstables the written sstables. can be null or empty if the memtable was clean.
 */
public void replaceFlushed(Memtable memtable, Collection<SSTableReader> sstables)
{
    cfs.getTracker().replaceFlushed(memtable, sstables);
    if (sstables != null && !sstables.isEmpty())
        CompactionManager.instance.submitBackground(cfs);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
static View fakeView(int memtableCount, int sstableCount, ColumnFamilyStore cfs)
{
    List<Memtable> memtables = new ArrayList<>();
    List<SSTableReader> sstables = new ArrayList<>();
    for (int i = 0 ; i < memtableCount ; i++)
        memtables.add(MockSchema.memtable(cfs));
    for (int i = 0 ; i < sstableCount ; i++)
        sstables.add(MockSchema.sstable(i, cfs));
    return new View(ImmutableList.copyOf(memtables), Collections.<Memtable>emptyList(), Helpers.identityMap(sstables),
                    Collections.<SSTableReader, SSTableReader>emptyMap(), SSTableIntervalTree.build(sstables));
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
@Test
public void testNotifications()
{
    ColumnFamilyStore cfs = MockSchema.newCFS();
    SSTableReader r1 = MockSchema.sstable(0, cfs), r2 = MockSchema.sstable(1, cfs);
    Tracker tracker = new Tracker(null, false);
    MockListener listener = new MockListener(false);
    tracker.subscribe(listener);
    tracker.notifyAdded(singleton(r1));
    Assert.assertEquals(singleton(r1), ((SSTableAddedNotification) listener.received.get(0)).added);
    listener.received.clear();
    tracker.notifyDeleting(r1);
    Assert.assertEquals(r1, ((SSTableDeletingNotification) listener.received.get(0)).deleting);
    listener.received.clear();
    Assert.assertNull(tracker.notifySSTablesChanged(singleton(r1), singleton(r2), OperationType.COMPACTION, null));
    Assert.assertEquals(singleton(r1), ((SSTableListChangedNotification) listener.received.get(0)).removed);
    Assert.assertEquals(singleton(r2), ((SSTableListChangedNotification) listener.received.get(0)).added);
    listener.received.clear();
    tracker.notifySSTableRepairedStatusChanged(singleton(r1));
    Assert.assertEquals(singleton(r1), ((SSTableRepairStatusChanged) listener.received.get(0)).sstable);
    listener.received.clear();
    Memtable memtable = MockSchema.memtable(cfs);
    tracker.notifyRenewed(memtable);
    Assert.assertEquals(memtable, ((MemtableRenewedNotification) listener.received.get(0)).renewed);
    listener.received.clear();
    tracker.unsubscribe(listener);
    MockListener failListener = new MockListener(true);
    tracker.subscribe(failListener);
    tracker.subscribe(listener);
    Assert.assertNotNull(tracker.notifyAdded(singleton(r1), null));
    Assert.assertEquals(singleton(r1), ((SSTableAddedNotification) listener.received.get(0)).added);
    listener.received.clear();
    Assert.assertNotNull(tracker.notifySSTablesChanged(singleton(r1), singleton(r2), OperationType.COMPACTION, null));
    Assert.assertEquals(singleton(r1), ((SSTableListChangedNotification) listener.received.get(0)).removed);
    Assert.assertEquals(singleton(r2), ((SSTableListChangedNotification) listener.received.get(0)).added);
    listener.received.clear();
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public MemtableRenewedNotification(Memtable renewed)
{
    this.renewed = renewed;
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public Memtable getCurrentMemtable()
{
    return liveMemtables.get(liveMemtables.size() - 1);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
/**
 * @return the active memtable and all the memtables that are pending flush.
 */
public Iterable<Memtable> getAllMemtables()
{
    return concat(flushingMemtables, liveMemtables);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public void markFlushing(Memtable memtable)
{
    apply(View.markFlushing(memtable));
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public void notifyRenewed(Memtable renewed)
{
    INotification notification = new MemtableRenewedNotification(renewed);
    for (INotificationConsumer subscriber : subscribers)
        subscriber.handleNotification(notification, this);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
public void replaceFlushed(Memtable memtable, Collection<SSTableReader> sstables)
{
    cfs.getTracker().replaceFlushed(memtable, sstables);
    if (sstables != null && !sstables.isEmpty())
        CompactionManager.instance.submitBackground(cfs);
}
 

import org.apache.cassandra.db.Memtable; //导入依赖的package包/类
@Test
public void testFlushing()
{
    ColumnFamilyStore cfs = MockSchema.newCFS();
    View initialView = fakeView(1, 0, cfs);
    View cur = initialView;
    Memtable memtable1 = initialView.getCurrentMemtable();
    Memtable memtable2 = MockSchema.memtable(cfs);

    cur = View.switchMemtable(memtable2).apply(cur);
    Assert.assertEquals(2, cur.liveMemtables.size());
    Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
    Assert.assertEquals(memtable2, cur.getCurrentMemtable());

    Memtable memtable3 = MockSchema.memtable(cfs);
    cur = View.switchMemtable(memtable3).apply(cur);
    Assert.assertEquals(3, cur.liveMemtables.size());
    Assert.assertEquals(0, cur.flushingMemtables.size());
    Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
    Assert.assertEquals(memtable2, cur.liveMemtables.get(1));
    Assert.assertEquals(memtable3, cur.getCurrentMemtable());

    testFailure(View.replaceFlushed(memtable2, null), cur);

    cur = View.markFlushing(memtable2).apply(cur);
    Assert.assertTrue(cur.flushingMemtables.contains(memtable2));
    Assert.assertEquals(2, cur.liveMemtables.size());
    Assert.assertEquals(1, cur.flushingMemtables.size());
    Assert.assertEquals(memtable2, cur.flushingMemtables.get(0));
    Assert.assertEquals(memtable1, cur.liveMemtables.get(0));
    Assert.assertEquals(memtable3, cur.getCurrentMemtable());

    cur = View.markFlushing(memtable1).apply(cur);
    Assert.assertEquals(1, cur.liveMemtables.size());
    Assert.assertEquals(2, cur.flushingMemtables.size());
    Assert.assertEquals(memtable1, cur.flushingMemtables.get(0));
    Assert.assertEquals(memtable2, cur.flushingMemtables.get(1));
    Assert.assertEquals(memtable3, cur.getCurrentMemtable());

    cur = View.replaceFlushed(memtable2, null).apply(cur);
    Assert.assertEquals(1, cur.liveMemtables.size());
    Assert.assertEquals(1, cur.flushingMemtables.size());
    Assert.assertEquals(memtable1, cur.flushingMemtables.get(0));
    Assert.assertEquals(memtable3, cur.getCurrentMemtable());

    SSTableReader sstable = MockSchema.sstable(1, cfs);
    cur = View.replaceFlushed(memtable1, Collections.singleton(sstable)).apply(cur);
    Assert.assertEquals(0, cur.flushingMemtables.size());
    Assert.assertEquals(1, cur.liveMemtables.size());
    Assert.assertEquals(memtable3, cur.getCurrentMemtable());
    Assert.assertEquals(1, cur.sstables.size());
    Assert.assertEquals(sstable, cur.sstablesMap.get(sstable));
}