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

Fix growing pending background compaction 

patch by yukim; reviewed by benedict for CASSANDRA-9662
This commit is contained in:
Yuki Morishita 2015-06-26 11:50:51 -05:00
parent 452d6a445a
commit f283ed2981
5 changed files with 20 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -1,5 +1,6 @@
2.0.18
* Scrub (recover) sstables even when -Index.db is missing, (CASSANDRA-9591)
* Scrub (recover) sstables even when -Index.db is missing, (CASSANDRA-9591)
* Fix growing pending background compaction (CASSANDRA-9662)
2.0.17

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

@ -180,20 +180,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
public void run()
{
List<ColumnFamilyStore> submitted = new ArrayList<>();
for (Keyspace keyspace : Keyspace.all())
for (ColumnFamilyStore cfs : keyspace.getColumnFamilyStores())
if (!CompactionManager.instance.submitBackground(cfs, false).isEmpty())
submitted.add(cfs);
while (!submitted.isEmpty() && CompactionManager.instance.getActiveCompactions() < CompactionManager.instance.getMaximumCompactorThreads())
{
List<ColumnFamilyStore> submitMore = ImmutableList.copyOf(submitted);
submitted.clear();
for (ColumnFamilyStore cfs : submitMore)
if (!CompactionManager.instance.submitBackground(cfs, false).isEmpty())
submitted.add(cfs);
}
CompactionManager.instance.submitBackground(cfs);
}
};
}

18
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -127,11 +127,6 @@ public class CompactionManager implements CompactionManagerMBean
* turn into a no-op in the bucketing/candidate-scan phase.
*/
public List<Future<?>> submitBackground(final ColumnFamilyStore cfs)
{
return submitBackground(cfs, true);
}
public List<Future<?>> submitBackground(final ColumnFamilyStore cfs, boolean autoFill)
{
if (cfs.isAutoCompactionDisabled())
{
@ -151,14 +146,11 @@ public class CompactionManager implements CompactionManagerMBean
cfs.keyspace.getName(),
cfs.name,
cfs.getCompactionStrategy().getClass().getSimpleName());
List<Future<?>> futures = new ArrayList<Future<?>>();
List<Future<?>> futures = new ArrayList<>();
// we must schedule it at least once, otherwise compaction will stop for a CF until next flush
do {
compactingCF.add(cfs);
futures.add(executor.submit(new BackgroundCompactionTask(cfs)));
// if we have room for more compactions, then fill up executor
} while (autoFill && executor.getActiveCount() + futures.size() < executor.getMaximumPoolSize());
compactingCF.add(cfs);
futures.add(executor.submit(new BackgroundCompactionCandidate(cfs)));
return futures;
}
@ -173,11 +165,11 @@ public class CompactionManager implements CompactionManagerMBean
// the actual sstables to compact are not determined until we run the BCT; that way, if new sstables
// are created between task submission and execution, we execute against the most up-to-date information
class BackgroundCompactionTask implements Runnable
class BackgroundCompactionCandidate implements Runnable
{
private final ColumnFamilyStore cfs;
BackgroundCompactionTask(ColumnFamilyStore cfs)
BackgroundCompactionCandidate(ColumnFamilyStore cfs)
{
this.cfs = cfs;
}

6
src/java/org/apache/cassandra/metrics/CompactionMetrics.java
View File

@ -58,14 +58,14 @@ public class CompactionMetrics implements CompactionManager.CompactionExecutorSt
public Integer value()
{
int n = 0;
// add estimate number of compactions need to be done
for (String keyspaceName : Schema.instance.getKeyspaces())
{
for (ColumnFamilyStore cfs : Keyspace.open(keyspaceName).getColumnFamilyStores())
n += cfs.getCompactionStrategy().getEstimatedRemainingTasks();
}
for (ThreadPoolExecutor collector : collectors)
n += collector.getTaskCount() - collector.getCompletedTaskCount();
return n;
// add number of currently running compactions
return n + compactions.size();
}
});
completedTasks = Metrics.newGauge(factory.createMetricName("CompletedTasks"), new Gauge<Long>()

12
test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
View File

@ -82,8 +82,10 @@ public class CompactionsTest extends SchemaLoader
// enable compaction, submit background and wait for it to complete
store.enableAutoCompaction();
FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(store));
while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0)
do
{
TimeUnit.SECONDS.sleep(1);
} while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0);
// and sstable with ttl should be compacted
assertEquals(1, store.getSSTables().size());
@ -202,8 +204,10 @@ public class CompactionsTest extends SchemaLoader
// enable compaction, submit background and wait for it to complete
store.enableAutoCompaction();
FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(store));
while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0)
do
{
TimeUnit.SECONDS.sleep(1);
} while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0);
// even though both sstables were candidate for tombstone compaction
// it was not executed because they have an overlapping token range
@ -222,8 +226,10 @@ public class CompactionsTest extends SchemaLoader
//submit background task again and wait for it to complete
FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(store));
while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0)
do
{
TimeUnit.SECONDS.sleep(1);
} while (CompactionManager.instance.getPendingTasks() > 0 || CompactionManager.instance.getActiveCompactions() > 0);
//we still have 2 sstables, since they were not compacted against each other
assertEquals(2, store.getSSTables().size());