diff --git a/examples/bmt/CassandraBulkLoader.java b/examples/bmt/CassandraBulkLoader.java new file mode 100644 index 0000000000..6fb1a91e1a --- /dev/null +++ b/examples/bmt/CassandraBulkLoader.java @@ -0,0 +1,293 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /** + * Cassandra has a back door called the Binary Memtable. The purpose of this backdoor is to + * mass import large amounts of data, without using the Thrift interface. + * + * Inserting data through the binary memtable, allows you to skip the commit log overhead, and an ack + * from Thrift on every insert. The example below utilizes Hadoop to generate all the data necessary + * to send to Cassandra, and sends it using the Binary Memtable interface. What Hadoop ends up doing is + * creating the actual data that gets put into an SSTable as if you were using Thrift. With enough Hadoop nodes + * inserting the data, the bottleneck at this point should become the network. + * + * We recommend adjusting the compaction threshold to 0, while the import is running. After the import, you need + * to run `nodeprobe -host flush_binary ` on every node, as this will flush the remaining data still left + * in memory to disk. Then it's recommended to adjust the compaction threshold to it's original value. + * + * The example below is a sample Hadoop job, and it inserts SuperColumns. It can be tweaked to work with normal Columns. + * + * You should construct your data you want to import as rows delimited by a new line. You end up grouping by + * in the mapper, so that the end result generates the data set into a column oriented subset. Once you get to the + * reduce aspect, you can generate the ColumnFamilies you want inserted, and send it to your nodes. + * + * For Cassandra 0.6.4, we modified this example to wait for acks from all Cassandra nodes for each row + * before proceeding to the next. This means to keep Cassandra similarly busy you can either + * 1) add more reducer tasks, + * 2) remove the "wait for acks" block of code, + * 3) parallelize the writing of rows to Cassandra, e.g. with an Executor. + * + * THIS CANNOT RUN ON THE SAME IP ADDRESS AS A CASSANDRA INSTANCE. + */ + +package org.apache.cassandra.bulkloader; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.google.common.base.Charsets; + +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Column; +import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.ColumnFamilyType; +import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.db.filter.QueryPath; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.net.IAsyncResult; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.hadoop.filecache.DistributedCache; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; +import org.apache.hadoop.mapred.*; + +public class CassandraBulkLoader { + public static class Map extends MapReduceBase implements Mapper { + public void map(Text key, Text value, OutputCollector output, Reporter reporter) throws IOException { + // This is a simple key/value mapper. + output.collect(key, value); + } + } + + public static class Reduce extends MapReduceBase implements Reducer { + private Path[] localFiles; + private JobConf jobconf; + + public void configure(JobConf job) { + this.jobconf = job; + String cassConfig; + + // Get the cached files + try + { + localFiles = DistributedCache.getLocalCacheFiles(job); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + cassConfig = localFiles[0].getParent().toString(); + + System.setProperty("storage-config",cassConfig); + + try + { + StorageService.instance.initClient(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + try + { + Thread.sleep(10*1000); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + } + + public void close() + { + try + { + // release the cache + DistributedCache.releaseCache(new URI("/cassandra/storage-conf.xml#storage-conf.xml"), this.jobconf); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + catch (URISyntaxException e) + { + throw new RuntimeException(e); + } + try + { + // Sleep just in case the number of keys we send over is small + Thread.sleep(3*1000); + } + catch (InterruptedException e) + { + throw new RuntimeException(e); + } + StorageService.instance.stopClient(); + } + + public void reduce(Text key, Iterator values, OutputCollector output, Reporter reporter) throws IOException + { + ColumnFamily columnFamily; + String keyspace = "Keyspace1"; + String cfName = "Super1"; + Message message; + List columnFamilies; + columnFamilies = new LinkedList(); + String line; + + /* Create a column family */ + columnFamily = ColumnFamily.create(keyspace, cfName); + while (values.hasNext()) { + // Split the value (line based on your own delimiter) + line = values.next().toString(); + String[] fields = line.split("\1"); + String SuperColumnName = fields[1]; + String ColumnName = fields[2]; + String ColumnValue = fields[3]; + int timestamp = 0; + columnFamily.addColumn(new QueryPath(cfName, + ByteBufferUtil.bytes(SuperColumnName), + ByteBufferUtil.bytes(ColumnName)), + ByteBufferUtil.bytes(ColumnValue), + timestamp); + } + + columnFamilies.add(columnFamily); + + /* Get serialized message to send to cluster */ + message = createMessage(keyspace, key.getBytes(), cfName, columnFamilies); + List results = new ArrayList(); + for (InetAddress endpoint: StorageService.instance.getNaturalEndpoints(keyspace, ByteBufferUtil.bytes(key))) + { + /* Send message to end point */ + results.add(MessagingService.instance().sendRR(message, endpoint)); + } + /* wait for acks */ + for (IAsyncResult result : results) + { + try + { + result.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + } + catch (TimeoutException e) + { + // you should probably add retry logic here + throw new RuntimeException(e); + } + } + + output.collect(key, new Text(" inserted into Cassandra node(s)")); + } + } + + public static void runJob(String[] args) + { + JobConf conf = new JobConf(CassandraBulkLoader.class); + + if(args.length >= 4) + { + conf.setNumReduceTasks(new Integer(args[3])); + } + + try + { + // We store the cassandra storage-conf.xml on the HDFS cluster + DistributedCache.addCacheFile(new URI("/cassandra/storage-conf.xml#storage-conf.xml"), conf); + } + catch (URISyntaxException e) + { + throw new RuntimeException(e); + } + conf.setInputFormat(KeyValueTextInputFormat.class); + conf.setJobName("CassandraBulkLoader_v2"); + conf.setMapperClass(Map.class); + conf.setReducerClass(Reduce.class); + + conf.setOutputKeyClass(Text.class); + conf.setOutputValueClass(Text.class); + + FileInputFormat.setInputPaths(conf, new Path(args[1])); + FileOutputFormat.setOutputPath(conf, new Path(args[2])); + try + { + JobClient.runJob(conf); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + public static Message createMessage(String keyspace, byte[] key, String columnFamily, List columnFamilies) + { + ColumnFamily baseColumnFamily; + DataOutputBuffer bufOut = new DataOutputBuffer(); + RowMutation rm; + Message message; + Column column; + + /* Get the first column family from list, this is just to get past validation */ + baseColumnFamily = new ColumnFamily(ColumnFamilyType.Standard, + DatabaseDescriptor.getComparator(keyspace, columnFamily), + DatabaseDescriptor.getSubComparator(keyspace, columnFamily), + CFMetaData.getId(keyspace, columnFamily)); + + for(ColumnFamily cf : columnFamilies) { + bufOut.reset(); + ColumnFamily.serializer().serializeWithIndexes(cf, bufOut); + byte[] data = new byte[bufOut.getLength()]; + System.arraycopy(bufOut.getData(), 0, data, 0, bufOut.getLength()); + + column = new Column(FBUtilities.toByteBuffer(cf.id()), ByteBuffer.wrap(data), 0); + baseColumnFamily.addColumn(column); + } + rm = new RowMutation(keyspace, ByteBuffer.wrap(key)); + rm.add(baseColumnFamily); + + try + { + /* Make message */ + message = rm.makeRowMutationMessage(StorageService.Verb.BINARY, MessagingService.version_); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + + return message; + } + public static void main(String[] args) throws Exception + { + runJob(args); + } +} diff --git a/examples/bmt/README.txt b/examples/bmt/README.txt new file mode 100644 index 0000000000..78ba11c09d --- /dev/null +++ b/examples/bmt/README.txt @@ -0,0 +1,34 @@ +This is an example for the deprecated BinaryMemtable bulk-load interface. + +Inserting data through the binary memtable, allows you to skip the +commit log overhead, and an ack from Thrift on every insert. The +example below utilizes Hadoop to generate all the data necessary to +send to Cassandra, and sends it using the Binary Memtable +interface. What Hadoop ends up doing is creating the actual data that +gets put into an SSTable as if you were using Thrift. With enough +Hadoop nodes inserting the data, the bottleneck at this point should +become the network. + +We recommend adjusting the compaction threshold to 0 while the import +is running. After the import, you need to run `nodeprobe -host +flush_binary ` on every node, as this will flush the +remaining data still left in memory to disk. Then it's recommended to +adjust the compaction threshold to it's original value. + +The example in CassandraBulkLoader.java is a sample Hadoop job that +inserts SuperColumns. It can be tweaked to work with normal Columns. + +You should construct your data you want to import as rows delimited by +a new line. You end up grouping by in the mapper, so that +the end result generates the data set into a column oriented +subset. Once you get to the reduce aspect, you can generate the +ColumnFamilies you want inserted, and send it to your nodes. + +For Cassandra 0.6.4, we modified this example to wait for acks from +all Cassandra nodes for each row before proceeding to the next. This +means to keep Cassandra similarly busy you can either +1) add more reducer tasks, +2) remove the "wait for acks" block of code, +3) parallelize the writing of rows to Cassandra, e.g. with an Executor. + +THIS CANNOT RUN ON THE SAME IP ADDRESS AS A CASSANDRA INSTANCE. diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 2d48c2272f..f76b6f22f0 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -77,6 +77,8 @@ public class Config public Boolean snapshot_before_compaction = false; public Integer compaction_thread_priority = Thread.MIN_PRIORITY; + public Integer binary_memtable_throughput_in_mb = 256; + /* if the size of columns or super-columns are more than this, indexing will kick in */ public Integer column_index_size_in_kb = 64; public Integer in_memory_compaction_limit_in_mb = 256; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 2d03df0e1c..24c9cbc4b2 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -937,6 +937,11 @@ public class DatabaseDescriptor return conf.sliced_buffer_size_in_kb; } + public static int getBMTThreshold() + { + return conf.binary_memtable_throughput_in_mb; + } + public static int getCompactionThreadPriority() { return conf.compaction_thread_priority; diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java new file mode 100644 index 0000000000..01fa10a396 --- /dev/null +++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java @@ -0,0 +1,160 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.commitlog.ReplayPosition; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.io.sstable.SSTableReader; +import org.apache.cassandra.io.sstable.SSTableWriter; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.WrappedRunnable; +import org.cliffc.high_scale_lib.NonBlockingHashMap; + +public class BinaryMemtable implements IFlushable +{ + private static final Logger logger = LoggerFactory.getLogger(BinaryMemtable.class); + private final int threshold = DatabaseDescriptor.getBMTThreshold() * 1024 * 1024; + private final AtomicInteger currentSize = new AtomicInteger(0); + + /* Table and ColumnFamily name are used to determine the ColumnFamilyStore */ + private boolean isFrozen = false; + private final Map columnFamilies = new NonBlockingHashMap(); + /* Lock and Condition for notifying new clients about Memtable switches */ + private final Lock lock = new ReentrantLock(); + Condition condition; + private final IPartitioner partitioner = StorageService.getPartitioner(); + private final ColumnFamilyStore cfs; + + public BinaryMemtable(ColumnFamilyStore cfs) + { + this.cfs = cfs; + condition = lock.newCondition(); + } + + boolean isThresholdViolated() + { + return currentSize.get() >= threshold; + } + + /* + * This version is used by the external clients to put data into + * the memtable. This version will respect the threshold and flush + * the memtable to disk when the size exceeds the threshold. + */ + void put(DecoratedKey key, ByteBuffer buffer) + { + if (isThresholdViolated()) + { + lock.lock(); + try + { + if (!isFrozen) + { + isFrozen = true; + cfs.submitFlush(this, new CountDownLatch(1), null); + cfs.switchBinaryMemtable(key, buffer); + } + else + { + cfs.applyBinary(key, buffer); + } + } + finally + { + lock.unlock(); + } + } + else + { + resolve(key, buffer); + } + } + + public boolean isClean() + { + return columnFamilies.isEmpty(); + } + + private void resolve(DecoratedKey key, ByteBuffer buffer) + { + columnFamilies.put(key, buffer); + currentSize.addAndGet(buffer.remaining() + key.key.remaining()); + } + + private List getSortedKeys() + { + assert !columnFamilies.isEmpty(); + logger.info("Sorting " + this); + List keys = new ArrayList(columnFamilies.keySet()); + Collections.sort(keys); + return keys; + } + + private SSTableReader writeSortedContents(List sortedKeys, ReplayPosition context) throws IOException + { + logger.info("Writing " + this); + SSTableWriter writer = cfs.createFlushWriter(sortedKeys.size(), DatabaseDescriptor.getBMTThreshold(), context); + + for (DecoratedKey key : sortedKeys) + { + ByteBuffer bytes = columnFamilies.get(key); + assert bytes.remaining() > 0; + writer.append(key, bytes); + } + SSTableReader sstable = writer.closeAndOpenReader(); + logger.info("Completed flushing " + writer.getFilename()); + return sstable; + } + + public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context) + { + sorter.execute(new Runnable() + { + public void run() + { + final List sortedKeys = getSortedKeys(); + writer.execute(new WrappedRunnable() + { + public void runMayThrow() throws IOException + { + cfs.addSSTable(writeSortedContents(sortedKeys, context)); + latch.countDown(); + } + }); + } + }); + } +} diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 6e5dba7972..9113bffd20 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -25,6 +25,7 @@ import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.regex.Pattern; @@ -71,10 +72,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean private static Logger logger = LoggerFactory.getLogger(ColumnFamilyStore.class); /* - * maybeSwitchMemtable puts Memtable.getSortedContents on the writer executor. When the write is complete, + * submitFlush first puts [Binary]Memtable.getSortedContents on the flushSorter executor, + * which then puts the sorted results on the writer executor. This is because sorting is CPU-bound, + * and writing is disk-bound; we want to be able to do both at once. When the write is complete, * we turn the writer into an SSTableReader and add it to ssTables_ where it is available for reads. * - * There are two other things that maybeSwitchMemtable does. + * For BinaryMemtable that's about all that happens. For live Memtables there are two other things + * that switchMemtable does (which should be the only caller of submitFlush in this case). * First, it puts the Memtable into memtablesPendingFlush, where it stays until the flush is complete * and it's been added as an SSTableReader to ssTables_. Second, it adds an entry to commitLogUpdater * that waits for the flush to complete, then calls onMemtableFlush. This allows multiple flushes @@ -82,6 +86,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * which is necessary for replay in case of a restart since CommitLog assumes that when onMF is * called, all data up to the given context has been persisted to SSTables. */ + private static final ExecutorService flushSorter + = new JMXEnabledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), + StageManager.KEEPALIVE, + TimeUnit.SECONDS, + new LinkedBlockingQueue(Runtime.getRuntime().availableProcessors()), + new NamedThreadFactory("FlushSorter"), + "internal"); private static final ExecutorService flushWriter = new JMXEnabledThreadPoolExecutor(DatabaseDescriptor.getFlushWriters(), StageManager.KEEPALIVE, @@ -123,6 +134,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean private final ConcurrentSkipListMap indexedColumns; + // TODO binarymemtable ops are not threadsafe (do they need to be?) + private AtomicReference binaryMemtable; + private LatencyTracker readStats = new LatencyTracker(); private LatencyTracker writeStats = new LatencyTracker(); @@ -243,6 +257,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean this.keyCacheSaveInSeconds = new DefaultInteger(metadata.getKeyCacheSavePeriodInSeconds()); this.partitioner = partitioner; fileIndexGenerator.set(generation); + binaryMemtable = new AtomicReference(new BinaryMemtable(this)); if (logger.isDebugEnabled()) logger.debug("Starting CFS {}", columnFamily); @@ -643,11 +658,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } final CountDownLatch latch = new CountDownLatch(icc.size()); for (ColumnFamilyStore cfs : icc) - { - Memtable memtable = cfs.data.switchMemtable(); - logger.info("Enqueuing flush of {}", memtable); - memtable.flushAndSignal(latch, flushWriter, ctx); - } + submitFlush(cfs.data.switchMemtable(), latch, ctx); // we marked our memtable as frozen as part of the concurrency control, // so even if there was nothing to flush we need to switch it out @@ -688,6 +699,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean : DatabaseDescriptor.getCFMetaData(metadata.cfId) == null; } + void switchBinaryMemtable(DecoratedKey key, ByteBuffer buffer) + { + binaryMemtable.set(new BinaryMemtable(this)); + binaryMemtable.get().put(key, buffer); + } + public void forceFlushIfExpired() { if (getMemtableThreadSafe().isExpired()) @@ -718,6 +735,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean future.get(); } + public void forceFlushBinary() + { + if (binaryMemtable.get().isClean()) + return; + + submitFlush(binaryMemtable.get(), new CountDownLatch(1), null); + } + public void updateRowCache(DecoratedKey key, ColumnFamily columnFamily) { if (rowCache.isPutCopying()) @@ -768,6 +793,20 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return flushRequested ? mt : null; } + /* + * Insert/Update the column family for this key. param @ lock - lock that + * Caller is responsible for acquiring Table.flusherLock! + * param @ lock - lock that needs to be used. + * needs to be used. param @ key - key for update/insert param @ + * columnFamily - columnFamily changes + */ + void applyBinary(DecoratedKey key, ByteBuffer buffer) + { + long start = System.nanoTime(); + binaryMemtable.get().put(key, buffer); + writeStats.addNano(System.nanoTime() - start); + } + public static ColumnFamily removeDeletedCF(ColumnFamily cf, int gcBefore) { // in case of a timestamp tie, tombstones get priority over non-tombstones. @@ -959,6 +998,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + /** + * submits flush sort on the flushSorter executor, which will in turn submit to flushWriter when sorted. + * TODO because our executors use CallerRunsPolicy, when flushSorter fills up, no writes will proceed + * because the next flush will start executing on the caller, mutation-stage thread that has the + * flush write lock held. (writes aquire this as a read lock before proceeding.) + * This is good, because it backpressures flushes, but bad, because we can't write until that last + * flushing thread finishes sorting, which will almost always be longer than any of the flushSorter threads proper + * (since, by definition, it started last). + */ + void submitFlush(IFlushable flushable, CountDownLatch latch, ReplayPosition context) + { + logger.info("Enqueuing flush of {}", flushable); + flushable.flushAndSignal(latch, flushSorter, flushWriter, context); + } + public long getMemtableColumnsCount() { return getMemtableThreadSafe().getOperations(); diff --git a/src/java/org/apache/cassandra/db/IFlushable.java b/src/java/org/apache/cassandra/db/IFlushable.java new file mode 100644 index 0000000000..1be11c4e95 --- /dev/null +++ b/src/java/org/apache/cassandra/db/IFlushable.java @@ -0,0 +1,32 @@ +package org.apache.cassandra.db; +/* + * + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + * + */ + + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; + +import org.apache.cassandra.db.commitlog.ReplayPosition; + +public interface IFlushable +{ + public void flushAndSignal(CountDownLatch condition, ExecutorService sorter, ExecutorService writer, ReplayPosition context); +} diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index a57563401c..a17acc2fa7 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -46,7 +46,7 @@ import org.apache.cassandra.io.sstable.SSTableWriter; import org.apache.cassandra.utils.WrappedRunnable; import org.github.jamm.MemoryMeter; -public class Memtable implements Comparable +public class Memtable implements Comparable, IFlushable { private static final Logger logger = LoggerFactory.getLogger(Memtable.class); @@ -256,7 +256,7 @@ public class Memtable implements Comparable return ssTable; } - public void flushAndSignal(final CountDownLatch latch, ExecutorService writer, final ReplayPosition context) + public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context) { writer.execute(new WrappedRunnable() {