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

merge from 0.3 branch 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@776716 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-05-20 14:44:41 +00:00
parent 0ae07f299e
commit d03d80b6ed
6 changed files with 48 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
build.xml
View File

@ -178,7 +178,7 @@
</copy>
<tar compression="gzip" longfile="gnu"
destfile="${build.dir}/${final.name}.tgz">
destfile="${build.dir}/${final.name}-bin.tar.gz">
<tarfileset dir="${dist.dir}"
prefix="${final.name}">
@ -191,6 +191,15 @@
<include name="bin/cassandra"/>
</tarfileset>
</tar>
<tar compression="gzip" longfile="gnu"
destfile="${build.dir}/${final.name}-src.tar.gz">
<tarfileset dir="${basedir}"
prefix="${final.name}-src">
<include name="**"/>
<exclude name="build/**" />
</tarfileset>
</tar>
</target>
<target name="build-test" depends="build" description="Build the Cassandra classes">

BIN
lib/activation.jar
View File

Binary file not shown.

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

@ -56,7 +56,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
private static Logger logger_ = Logger.getLogger(ColumnFamilyStore.class);
private static int COMPACTION_THRESHOLD = 4; // compact this many sstables at a time
private static final int BUFSIZE = 128 * 1024 * 1024;
private static final int COMPACTION_MEMORY_THRESHOLD = 1 << 30;
@ -770,8 +769,8 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
lock_.writeLock().unlock();
}
if ((ssTableSize >= COMPACTION_THRESHOLD && !isCompacting_.get())
|| (isCompacting_.get() && ssTableSize % COMPACTION_THRESHOLD == 0))
if ((ssTableSize >= MinorCompactionManager.COMPACTION_THRESHOLD && !isCompacting_.get())
|| (isCompacting_.get() && ssTableSize % MinorCompactionManager.COMPACTION_THRESHOLD == 0))
{
logger_.debug("Submitting for compaction ...");
MinorCompactionManager.instance().submit(ColumnFamilyStore.this);
@ -860,11 +859,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
return buckets.keySet();
}
public int doCompaction() throws IOException
{
return doCompaction(COMPACTION_THRESHOLD);
}
/*
* Break the files into buckets and then compact.
*/
@ -1294,6 +1288,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
private int doFileCompaction(List<String> files, int minBufferSize) throws IOException
{
logger_.info("Compacting [" + StringUtils.join(files, ",") + "]");
String compactionFileLocation = DatabaseDescriptor.getCompactionFileLocation(getExpectedCompactedFileSize(files));
// If the compaction file path is null that means we have no space left for this compaction.
// try again w/o the largest one.
@ -1314,6 +1309,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (pq.isEmpty())
{
logger_.warn("Nothing to compact (all files empty or corrupt)");
// TODO clean out bad files, if any
return 0;
}

63
src/java/org/apache/cassandra/db/MinorCompactionManager.java
View File

@ -46,7 +46,8 @@ class MinorCompactionManager implements IComponentShutdown
private static MinorCompactionManager instance_;
private static Lock lock_ = new ReentrantLock();
private static Logger logger_ = Logger.getLogger(MinorCompactionManager.class);
final static long intervalInMins_ = 5;
private static final long intervalInMins_ = 5;
static final int COMPACTION_THRESHOLD = 4; // compact this many sstables at a time
public static MinorCompactionManager instance()
{
@ -66,33 +67,6 @@ class MinorCompactionManager implements IComponentShutdown
return instance_;
}
class FileCompactor implements Callable<Integer>
{
private ColumnFamilyStore columnFamilyStore_;
FileCompactor(ColumnFamilyStore columnFamilyStore)
{
columnFamilyStore_ = columnFamilyStore;
}
public Integer call()
{
logger_.debug("Started compaction ..." + columnFamilyStore_.columnFamily_);
try
{
return columnFamilyStore_.doCompaction();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
finally
{
logger_.debug("Finished compaction ..." + columnFamilyStore_.columnFamily_);
}
}
}
class FileCompactor2 implements Callable<Boolean>
{
private ColumnFamilyStore columnFamilyStore_;
@ -138,9 +112,9 @@ class MinorCompactionManager implements IComponentShutdown
public void run()
{
logger_.debug("Started Major compaction ..."+columnFamilyStore_.columnFamily_);
logger_.debug("Started Major compaction for " + columnFamilyStore_.columnFamily_);
columnFamilyStore_.doMajorCompaction(skip_);
logger_.debug("Finished Major compaction ..."+columnFamilyStore_.columnFamily_);
logger_.debug("Finished Major compaction for " + columnFamilyStore_.columnFamily_);
}
}
@ -183,22 +157,41 @@ class MinorCompactionManager implements IComponentShutdown
public void submitPeriodicCompaction(final ColumnFamilyStore columnFamilyStore)
{
Runnable runnable = new Runnable() // having to wrap Callable in Runnable is retarded but that's what the API insists on.
Runnable runnable = new Runnable()
{
public void run()
{
new FileCompactor(columnFamilyStore).call();
try
{
columnFamilyStore.doCompaction(COMPACTION_THRESHOLD);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
};
compactor_.scheduleWithFixedDelay(runnable, MinorCompactionManager.intervalInMins_,
MinorCompactionManager.intervalInMins_, TimeUnit.MINUTES);
}
public Future<Integer> submit(ColumnFamilyStore columnFamilyStore)
public Future<Integer> submit(final ColumnFamilyStore columnFamilyStore)
{
return compactor_.submit(new FileCompactor(columnFamilyStore));
return submit(columnFamilyStore, COMPACTION_THRESHOLD);
}
Future<Integer> submit(final ColumnFamilyStore columnFamilyStore, final int threshold)
{
Callable<Integer> callable = new Callable<Integer>()
{
public Integer call() throws IOException
{
return columnFamilyStore.doCompaction(threshold);
}
};
return compactor_.submit(callable);
}
public void submitCleanup(ColumnFamilyStore columnFamilyStore)
{
compactor_.submit(new CleanupCompactor(columnFamilyStore));

2
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -220,7 +220,7 @@ public class SSTable
}
File file = new File(dataFile);
assert file.exists();
assert file.exists() : "attempted to delete non-existing file " + dataFile;
/* delete the data file */
if (!file.delete())
{

7
test/unit/org/apache/cassandra/db/OneCompactionTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.Set;
import java.util.HashSet;
@ -44,7 +45,9 @@ public class OneCompactionTest
store.forceBlockingFlush();
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
}
store.doCompaction(2);
Future<Integer> ft = MinorCompactionManager.instance().submit(store, 2);
ft.get();
assertEquals(1, store.getSSTableFilenames().size());
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
}
@ -57,6 +60,6 @@ public class OneCompactionTest
@Test
public void testCompaction2() throws IOException, ExecutionException, InterruptedException
{
testCompaction("Standard2", 500);
testCompaction("Standard2", 5);
}
}