import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * removes all sstables that are not busy compacting.
 */
public void unreferenceSSTables()
{
    Set<SSTableReader> notCompacting;

    View currentView, newView;
    do
    {
        currentView = view.get();
        notCompacting = currentView.nonCompactingSStables();
        newView = currentView.replace(notCompacting, Collections.<SSTableReader>emptySet());
    }
    while (!view.compareAndSet(currentView, newView));

    if (notCompacting.isEmpty())
    {
        // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
        return;
    }
    notifySSTablesChanged(notCompacting, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
    removeOldSSTablesSize(notCompacting);
    releaseReferences(notCompacting, true);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    Set<SSTableReader> remaining = new HashSet<>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    for (SSTableReader sstable : currentView.sstables)
        if (!remaining.contains(sstable))
            sstable.selfRef().release();
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * A special kind of replacement for SSTableReaders that were cloned with a new index summary sampling level (see
 * SSTableReader.cloneWithNewSummarySamplingLevel and CASSANDRA-5519).  This does not mark the old reader
 * as compacted.
 * @param oldSSTables replaced readers
 * @param newSSTables replacement readers
 */
private void replaceReaders(Collection<SSTableReader> oldSSTables, Collection<SSTableReader> newSSTables, boolean notify)
{
    View currentView, newView;
    do
    {
        currentView = view.get();
        newView = currentView.replace(oldSSTables, newSSTables);
    }
    while (!view.compareAndSet(currentView, newView));

    if (!oldSSTables.isEmpty() && notify)
        notifySSTablesChanged(oldSSTables, newSSTables, OperationType.UNKNOWN);

    for (SSTableReader sstable : newSSTables)
        sstable.setTrackedBy(this);

    Refs.release(Refs.selfRefs(oldSSTables));
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
protected Writer(int keysToSave)
{
    if (keysToSave >= getKeySet().size())
        keys = getKeySet();
    else
        keys = hotKeySet(keysToSave);

    OperationType type;
    if (cacheType == CacheService.CacheType.KEY_CACHE)
        type = OperationType.KEY_CACHE_SAVE;
    else if (cacheType == CacheService.CacheType.ROW_CACHE)
        type = OperationType.ROW_CACHE_SAVE;
    else if (cacheType == CacheService.CacheType.COUNTER_CACHE)
        type = OperationType.COUNTER_CACHE_SAVE;
    else
        type = OperationType.UNKNOWN;

    info = new CompactionInfo(CFMetaData.denseCFMetaData(Keyspace.SYSTEM_KS, cacheType.toString(), BytesType.instance),
                              type,
                              0,
                              keys.size(),
                              "keys");
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * removes all sstables that are not busy compacting.
 */
public void unreferenceSSTables()
{
    Set<SSTableReader> notCompacting;

    View currentView, newView;
    do
    {
        currentView = view.get();
        notCompacting = currentView.nonCompactingSStables();
        newView = currentView.replace(notCompacting, Collections.<SSTableReader>emptySet());
    }
    while (!view.compareAndSet(currentView, newView));

    if (notCompacting.isEmpty())
    {
        // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
        return;
    }
    notifySSTablesChanged(notCompacting, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
    postReplace(notCompacting, Collections.<SSTableReader>emptySet());
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    List<SSTableReader> remaining = new ArrayList<>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
protected Writer(int keysToSave)
{
    if (keysToSave >= getKeySet().size())
        keys = getKeySet();
    else
        keys = hotKeySet(keysToSave);

    OperationType type;
    if (cacheType == CacheService.CacheType.KEY_CACHE)
        type = OperationType.KEY_CACHE_SAVE;
    else if (cacheType == CacheService.CacheType.ROW_CACHE)
        type = OperationType.ROW_CACHE_SAVE;
    else
        type = OperationType.UNKNOWN;

    info = new CompactionInfo(new CFMetaData(Keyspace.SYSTEM_KS, cacheType.toString(), ColumnFamilyType.Standard, BytesType.instance, null),
                              type,
                              0,
                              keys.size(),
                              "keys");
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
public void printCompactionStats(PrintStream outs)
{
    int compactionThroughput = probe.getCompactionThroughput();
    CompactionManagerMBean cm = probe.getCompactionManagerProxy();
    outs.println("pending tasks: " + cm.getPendingTasks());
    if (cm.getCompactions().size() > 0)
        outs.printf("%25s%16s%16s%16s%16s%10s%10s%n", "compaction type", "keyspace", "table", "completed", "total", "unit", "progress");
    long remainingBytes = 0;
    for (Map<String, String> c : cm.getCompactions())
    {
        String percentComplete = new Long(c.get("total")) == 0
                               ? "n/a"
                               : new DecimalFormat("0.00").format((double) new Long(c.get("completed")) / new Long(c.get("total")) * 100) + "%";
        outs.printf("%25s%16s%16s%16s%16s%10s%10s%n", c.get("taskType"), c.get("keyspace"), c.get("columnfamily"), c.get("completed"), c.get("total"), c.get("unit"), percentComplete);
        if (c.get("taskType").equals(OperationType.COMPACTION.toString()))
            remainingBytes += (new Long(c.get("total")) - new Long(c.get("completed")));
    }
    long remainingTimeInSecs = compactionThroughput == 0 || remainingBytes == 0
                    ? -1
                    : (remainingBytes) / (long) (1024L * 1024L * compactionThroughput);
    String remainingTime = remainingTimeInSecs < 0
                    ? "n/a"
                    : String.format("%dh%02dm%02ds", remainingTimeInSecs / 3600, (remainingTimeInSecs % 3600) / 60, (remainingTimeInSecs % 60));

    outs.printf("%25s%10s%n", "Active compaction remaining time : ", remainingTime);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
Throwable notifySSTablesChanged(Collection<SSTableReader> removed, Collection<SSTableReader> added, OperationType compactionType, Throwable accumulate)
{
    INotification notification = new SSTableListChangedNotification(added, removed, compactionType);
    for (INotificationConsumer subscriber : subscribers)
    {
        try
        {
            subscriber.handleNotification(notification, this);
        }
        catch (Throwable t)
        {
            accumulate = merge(accumulate, t);
        }
    }
    return accumulate;
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
public CompactionInfo getCompactionInfo()
{
    long rangesLeft = 0, rangesTotal = 0;
    Token lastToken = prevToken;

    // This approximation is not very accurate, but since we do not have a method which allows us to calculate the
    // percentage of a range covered by a second range, this is the best approximation that we can calculate.
    // Instead, we just count the total number of ranges that haven't been seen by the node (we use the order of
    // the tokens to determine whether they have been seen yet or not), and the total number of ranges that a node
    // has.
    for (Range<Token> range : StorageService.instance.getLocalRanges(baseCfs.keyspace.getName()))
    {
        rangesLeft++;
        rangesTotal++;
        // This will reset rangesLeft, so that the number of ranges left will be less than the total ranges at the
        // end of the method.
        if (lastToken == null || range.contains(lastToken))
            rangesLeft = 0;
    }
    return new CompactionInfo(baseCfs.metadata, OperationType.VIEW_BUILD, rangesLeft, rangesTotal, "ranges", compactionId);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Returns a Pair of all compacting and non-compacting sstables.  Non-compacting sstables will be marked as
 * compacting.
 */
@SuppressWarnings("resource")
private Pair<List<SSTableReader>, Map<UUID, LifecycleTransaction>> getCompactingAndNonCompactingSSTables()
{
    List<SSTableReader> allCompacting = new ArrayList<>();
    Map<UUID, LifecycleTransaction> allNonCompacting = new HashMap<>();
    for (Keyspace ks : Keyspace.all())
    {
        for (ColumnFamilyStore cfStore: ks.getColumnFamilyStores())
        {
            Set<SSTableReader> nonCompacting, allSSTables;
            LifecycleTransaction txn = null;
            do
            {
                View view = cfStore.getTracker().getView();
                allSSTables = ImmutableSet.copyOf(view.select(SSTableSet.CANONICAL));
                nonCompacting = ImmutableSet.copyOf(view.getUncompacting(allSSTables));
            }
            while (null == (txn = cfStore.getTracker().tryModify(nonCompacting, OperationType.UNKNOWN)));

            allNonCompacting.put(cfStore.metadata.cfId, txn);
            allCompacting.addAll(Sets.difference(allSSTables, nonCompacting));
        }
    }
    return Pair.create(allCompacting, allNonCompacting);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
@Test
public void testMarkObsolete()
{
    ColumnFamilyStore cfs = MockSchema.newCFS();
    LogTransaction txnLogs = new LogTransaction(OperationType.UNKNOWN);
    Iterable<SSTableReader> readers = Lists.newArrayList(MockSchema.sstable(1, cfs), MockSchema.sstable(2, cfs));

    List<LogTransaction.Obsoletion> obsoletions = new ArrayList<>();
    Assert.assertNull(Helpers.prepareForObsoletion(readers, txnLogs, obsoletions, null));
    assertNotNull(obsoletions);
    assertEquals(2, obsoletions.size());

    Throwable accumulate = Helpers.markObsolete(obsoletions, null);
    Assert.assertNull(accumulate);
    for (SSTableReader reader : readers)
        Assert.assertTrue(reader.isMarkedCompacted());

    accumulate = Helpers.markObsolete(obsoletions, null);
    assertNotNull(accumulate);

    txnLogs.finish();
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
@Test
public void testSplit()
{
    ColumnFamilyStore cfs = MockSchema.newCFS();
    Tracker tracker = new Tracker(null, false);
    List<SSTableReader> readers = readers(0, 4, cfs);
    tracker.addInitialSSTables(readers);
    LifecycleTransaction txn = tracker.tryModify(readers, OperationType.UNKNOWN);
    txn.cancel(readers.get(3));
    LifecycleTransaction txn2 = txn.split(readers.subList(0, 1));
    Assert.assertEquals(2, txn.originals().size());
    Assert.assertTrue(all(readers.subList(1, 3), in(txn.originals())));
    Assert.assertEquals(1, txn2.originals().size());
    Assert.assertTrue(all(readers.subList(0, 1), in(txn2.originals())));
    txn.update(readers(1, 2, cfs).get(0), true);
    boolean failed = false;
    try
    {
        txn.split(readers.subList(2, 3));
    }
    catch (Throwable t)
    {
        failed = true;
    }
    Assert.assertTrue(failed);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
@Test
public void testTryModify()
{
    ColumnFamilyStore cfs = MockSchema.newCFS();
    Tracker tracker = new Tracker(cfs, false);
    List<SSTableReader> readers = ImmutableList.of(MockSchema.sstable(0, true, cfs), MockSchema.sstable(1, cfs), MockSchema.sstable(2, cfs));
    tracker.addInitialSSTables(copyOf(readers));
    Assert.assertNull(tracker.tryModify(ImmutableList.of(MockSchema.sstable(0, cfs)), OperationType.COMPACTION));
    try (LifecycleTransaction txn = tracker.tryModify(readers.get(0), OperationType.COMPACTION);)
    {
        Assert.assertNotNull(txn);
        Assert.assertNull(tracker.tryModify(readers.get(0), OperationType.COMPACTION));
        Assert.assertEquals(1, txn.originals().size());
        Assert.assertTrue(txn.originals().contains(readers.get(0)));
    }
    try (LifecycleTransaction txn = tracker.tryModify(Collections.<SSTableReader>emptyList(), OperationType.COMPACTION);)
    {
        Assert.assertNotNull(txn);
        Assert.assertEquals(0, txn.originals().size());
    }
    readers.get(0).selfRef().release();
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
@Test
public void testSSTableSplit() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    truncate(cfs);
    cfs.disableAutoCompaction();
    SSTableReader s = writeFile(cfs, 1000);
    try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.UNKNOWN, s))
    {
        SSTableSplitter splitter = new SSTableSplitter(cfs, txn, 10);
        splitter.split();

        assertFileCounts(s.descriptor.directory.list());
        LifecycleTransaction.waitForDeletions();

        for (File f : s.descriptor.directory.listFiles())
        {
            // we need to clear out the data dir, otherwise tests running after this breaks
            FileUtils.deleteRecursive(f);
        }
    }
    truncate(cfs);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * removes all sstables that are not busy compacting.
 */
public void unreferenceSSTables()
{
    Set<SSTableReader> notCompacting;

    View currentView, newView;
    do
    {
        currentView = view.get();
        notCompacting = currentView.nonCompactingSStables();
        newView = currentView.replace(notCompacting, Collections.<SSTableReader>emptySet());
    }
    while (!view.compareAndSet(currentView, newView));

    if (notCompacting.isEmpty())
    {
        // notifySSTablesChanged -> LeveledManifest.promote doesn't like a no-op "promotion"
        return;
    }
    notifySSTablesChanged(notCompacting, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
    postReplace(notCompacting, Collections.<SSTableReader>emptySet(), true);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    Set<SSTableReader> remaining = new HashSet<>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    for (SSTableReader sstable : currentView.sstables)
        if (!remaining.contains(sstable))
            sstable.releaseReference();
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Calculate expected file size of SSTable after compaction.
 *
 * If operation type is {@code CLEANUP} and we're not dealing with an index sstable,
 * then we calculate expected file size with checking token range to be eliminated.
 *
 * Otherwise, we just add up all the files' size, which is the worst case file
 * size for compaction of all the list of files given.
 *
 * @param sstables SSTables to calculate expected compacted file size
 * @param operation Operation type
 * @return Expected file size of SSTable after compaction
 */
public long getExpectedCompactedFileSize(Iterable<SSTableReader> sstables, OperationType operation)
{
    if (operation != OperationType.CLEANUP || isIndex())
    {
        return SSTable.getTotalBytes(sstables);
    }

    // cleanup size estimation only counts bytes for keys local to this node
    long expectedFileSize = 0;
    Collection<Range<Token>> ranges = StorageService.instance.getLocalRanges(table.name);
    for (SSTableReader sstable : sstables)
    {
        List<Pair<Long, Long>> positions = sstable.getPositionsForRanges(ranges);
        for (Pair<Long, Long> position : positions)
            expectedFileSize += position.right - position.left;
    }
    return expectedFileSize;
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Discard all SSTables that were created before given timestamp. Caller is responsible to obtain compactionLock.
 *
 * @param truncatedAt The timestamp of the truncation
 *                    (all SSTables before that timestamp are going be marked as compacted)
 *
 * @return the most recent replay position of the truncated data
 */
public ReplayPosition discardSSTables(long truncatedAt)
{
    List<SSTableReader> truncatedSSTables = new ArrayList<SSTableReader>();

    for (SSTableReader sstable : getSSTables())
    {
        if (!sstable.newSince(truncatedAt))
            truncatedSSTables.add(sstable);
    }

    if (truncatedSSTables.isEmpty())
        return ReplayPosition.NONE;

    markCompacted(truncatedSSTables, OperationType.UNKNOWN);
    return ReplayPosition.getReplayPosition(truncatedSSTables);
}
 

import org.apache.cassandra.db.compaction.OperationType; //导入依赖的package包/类
/**
 * Removes every SSTable in the directory from the DataTracker's view.
 * @param directory the unreadable directory, possibly with SSTables in it, but not necessarily.
 */
void removeUnreadableSSTables(File directory)
{
    View currentView, newView;
    List<SSTableReader> remaining = new ArrayList<SSTableReader>();
    do
    {
        currentView = view.get();
        for (SSTableReader r : currentView.nonCompactingSStables())
        {
            if (!r.descriptor.directory.equals(directory))
                remaining.add(r);
        }

        if (remaining.size() == currentView.nonCompactingSStables().size())
            return;

        newView = currentView.replace(currentView.sstables, remaining);
    }
    while (!view.compareAndSet(currentView, newView));
    notifySSTablesChanged(remaining, Collections.<SSTableReader>emptySet(), OperationType.UNKNOWN);
}