From a432e180776ae4efbdc7e1aba6dec411e6fbf854 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 20 May 2009 14:58:39 +0000 Subject: [PATCH] more cleanup of compaction code. patch by jbellis; reviewed by Jun Rao for CASSANDRA-184 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@776720 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/db/ColumnFamilyStore.java | 121 ++++++------------ .../cassandra/db/MinorCompactionManager.java | 9 +- src/java/org/apache/cassandra/io/SSTable.java | 14 +- 3 files changed, 55 insertions(+), 89 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index cb43e69b9e..536d8b750b 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -88,9 +88,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean /* Modification lock used for protecting reads from compactions. */ private ReentrantReadWriteLock lock_ = new ReentrantReadWriteLock(true); - /* Flag indicates if a compaction is in process */ - private AtomicBoolean isCompacting_ = new AtomicBoolean(false); - private TimedStatsDeque readStats_ = new TimedStatsDeque(60000); private TimedStatsDeque diskReadStats_ = new TimedStatsDeque(60000); @@ -756,25 +753,26 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean */ void storeLocation(String filename, BloomFilter bf) { - int ssTableSize = 0; + int ssTableCount; lock_.writeLock().lock(); try { ssTables_.add(filename); SSTable.storeBloomFilter(filename, bf); - ssTableSize = ssTables_.size(); + ssTableCount = ssTables_.size(); } finally { lock_.writeLock().unlock(); } - if ((ssTableSize >= MinorCompactionManager.COMPACTION_THRESHOLD && !isCompacting_.get()) - || (isCompacting_.get() && ssTableSize % MinorCompactionManager.COMPACTION_THRESHOLD == 0)) + /* it's ok if compaction gets submitted multiple times while one is already in process. + worst that happens is, compactor will count the sstable files and decide there are + not enough to bother with. */ + if (ssTableCount >= MinorCompactionManager.COMPACTION_THRESHOLD) { - logger_.debug("Submitting for compaction ..."); - MinorCompactionManager.instance().submit(ColumnFamilyStore.this); - logger_.debug("Submitted for compaction ..."); + logger_.debug("Submitting " + columnFamily_ + " for compaction"); + MinorCompactionManager.instance().submit(this); } } @@ -862,45 +860,37 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean /* * Break the files into buckets and then compact. */ - public int doCompaction(int threshold) throws IOException + int doCompaction(int threshold) throws IOException { - isCompacting_.set(true); List files = new ArrayList(ssTables_); int filesCompacted = 0; - try + Set> buckets = getCompactionBuckets(files, 50L * 1024L * 1024L); + for (List fileList : buckets) { - Set> buckets = getCompactionBuckets(files, 50L * 1024L * 1024L); - for (List fileList : buckets) + Collections.sort(fileList, new FileNameComparator(FileNameComparator.Ascending)); + if (fileList.size() < threshold) { - Collections.sort(fileList, new FileNameComparator(FileNameComparator.Ascending)); - if (fileList.size() < threshold) + continue; + } + // For each bucket if it has crossed the threshhold do the compaction + // In case of range compaction merge the counting bloom filters also. + files.clear(); + int count = 0; + for (String file : fileList) + { + files.add(file); + count++; + if (count == threshold) { - continue; - } - // For each bucket if it has crossed the threshhold do the compaction - // In case of range compaction merge the counting bloom filters also. - files.clear(); - int count = 0; - for (String file : fileList) - { - files.add(file); - count++; - if (count == threshold) - { - filesCompacted += doFileCompaction(files, BUFSIZE); - break; - } + filesCompacted += doFileCompaction(files, BUFSIZE); + break; } } } - finally - { - isCompacting_.set(false); - } return filesCompacted; } - void doMajorCompaction(long skip) + void doMajorCompaction(long skip) throws IOException { doMajorCompactionInternal(skip); } @@ -911,39 +901,27 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean * all files greater than skip GB are skipped for this compaction. * Except if skip is 0 , in that case this is ignored and all files are taken. */ - void doMajorCompactionInternal(long skip) + void doMajorCompactionInternal(long skip) throws IOException { - isCompacting_.set(true); List filesInternal = new ArrayList(ssTables_); List files; - try + if (skip > 0L) { - if (skip > 0L) + files = new ArrayList(); + for (String file : filesInternal) { - files = new ArrayList(); - for (String file : filesInternal) + File f = new File(file); + if (f.length() < skip * 1024L * 1024L * 1024L) { - File f = new File(file); - if (f.length() < skip * 1024L * 1024L * 1024L) - { - files.add(file); - } + files.add(file); } } - else - { - files = filesInternal; - } - doFileCompaction(files, BUFSIZE); } - catch (IOException ex) + else { - logger_.error(ex); - } - finally - { - isCompacting_.set(false); + files = filesInternal; } + doFileCompaction(files, BUFSIZE); } /* @@ -983,19 +961,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean boolean doAntiCompaction(List ranges, EndPoint target, List fileList) throws IOException { - isCompacting_.set(true); List files = new ArrayList(ssTables_); - boolean result = true; - try - { - result = doFileAntiCompaction(files, ranges, target, fileList, null); - } - finally - { - isCompacting_.set(false); - } - return result; - + return doFileAntiCompaction(files, ranges, target, fileList, null); } void forceCleanup() @@ -1011,18 +978,10 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean */ void doCleanupCompaction() throws IOException { - isCompacting_.set(true); List files = new ArrayList(ssTables_); - try + for (String file : files) { - for (String file : files) - { - doCleanup(file); - } - } - finally - { - isCompacting_.set(false); + doCleanup(file); } } diff --git a/src/java/org/apache/cassandra/db/MinorCompactionManager.java b/src/java/org/apache/cassandra/db/MinorCompactionManager.java index f3e0ea2fdf..a57b9c4462 100644 --- a/src/java/org/apache/cassandra/db/MinorCompactionManager.java +++ b/src/java/org/apache/cassandra/db/MinorCompactionManager.java @@ -113,7 +113,14 @@ class MinorCompactionManager implements IComponentShutdown public void run() { logger_.debug("Started Major compaction for " + columnFamilyStore_.columnFamily_); - columnFamilyStore_.doMajorCompaction(skip_); + try + { + columnFamilyStore_.doMajorCompaction(skip_); + } + catch (IOException e) + { + throw new RuntimeException(e); + } logger_.debug("Finished Major compaction for " + columnFamilyStore_.columnFamily_); } } diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index 947db77f21..61101f54c9 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -203,28 +203,28 @@ public class SSTable * This method deletes both the specified data file * and the associated index file * - * @param dataFile - data file associated with the SSTable + * @param dataFileName - data file associated with the SSTable */ - public static void delete(String dataFile) + public static void delete(String dataFileName) { /* remove the cached index table from memory */ - indexMetadataMap_.remove(dataFile); + indexMetadataMap_.remove(dataFileName); /* Delete the checksum file associated with this data file */ try { - ChecksumManager.onFileDelete(dataFile); + ChecksumManager.onFileDelete(dataFileName); } catch (IOException ex) { logger_.info(LogUtil.throwableToString(ex)); } - File file = new File(dataFile); - assert file.exists() : "attempted to delete non-existing file " + dataFile; + File file = new File(dataFileName); + assert file.exists() : "attempted to delete non-existing file " + dataFileName; /* delete the data file */ if (!file.delete()) { - logger_.error("Failed to delete " + file.getName()); + logger_.error("Failed to delete " + dataFileName); } }