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

Ensure that submitBackground enqueues at least one task 

patch by Oleg Anastasyev; reviewed by jbellis for CASSANDRA-5554
This commit is contained in:
Jonathan Ellis 2013-05-13 11:51:56 -05:00
parent 56927973fe
commit 3b41d21d9b
2 changed files with 13 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
1.2.5
* Ensure that submitBackground enqueues at least one task (CASSANDRA-5554)
* fix 2i updates with identical values and timestamps (CASSANDRA-5540)
* fix compaction throttling bursty-ness (CASSANDRA-4316)
* reduce memory consumption of IndexSummary (CASSANDRA-5506)

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

@ -149,8 +149,8 @@ public class CompactionManager implements CompactionManagerMBean
/**
* Call this whenever a compaction might be needed on the given columnfamily.
* It's okay to over-call (within reason) since the compactions are single-threaded,
* and if a call is unnecessary, it will just be no-oped in the bucketing phase.
* It's okay to over-call (within reason) if a call is unnecessary, it will
* turn into a no-op in the bucketing/candidate-scan phase.
*/
public List<Future<?>> submitBackground(final ColumnFamilyStore cfs)
{
@ -158,21 +158,23 @@ public class CompactionManager implements CompactionManagerMBean
if (count > 0 && executor.getActiveCount() >= executor.getMaximumPoolSize())
{
logger.debug("Background compaction is still running for {}.{} ({} remaining). Skipping",
new Object[] {cfs.table.name, cfs.columnFamily, count});
cfs.table.name, cfs.columnFamily, count);
return Collections.emptyList();
}
logger.debug("Scheduling a background task check for {}.{} with {}",
new Object[] {cfs.table.name,
cfs.columnFamily,
cfs.getCompactionStrategy().getClass().getSimpleName()});
cfs.table.name,
cfs.columnFamily,
cfs.getCompactionStrategy().getClass().getSimpleName());
List<Future<?>> futures = new ArrayList<Future<?>>();
// if we have room for more compactions, then fill up executor
while (executor.getActiveCount() + futures.size() < executor.getMaximumPoolSize())
{
// we must schedule it at least once, otherwise compaction will stop for a CF until next flush
do {
futures.add(executor.submit(new BackgroundCompactionTask(cfs)));
compactingCF.add(cfs);
}
// if we have room for more compactions, then fill up executor
} while (executor.getActiveCount() + futures.size() < executor.getMaximumPoolSize());
return futures;
}
@ -590,7 +592,6 @@ public class CompactionManager implements CompactionManagerMBean
throw new IOException("disk full");
SSTableScanner scanner = sstable.getDirectScanner(getRateLimiter());
long rowsRead = 0;
List<IColumn> indexedColumnsInRow = null;
CleanupInfo ci = new CleanupInfo(sstable, scanner);