From a3e292648541aeac51be5df40238fde8abed6747 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 13 Jun 2011 19:46:18 +0000 Subject: [PATCH] r/m binarymemtable patch by jbellis; reviewed by brandonwilliams for CASSANDRA-2692 git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1135249 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 4 + NEWS.txt | 9 + examples/bmt/CassandraBulkLoader.java | 293 ------------------ examples/bmt/README.txt | 34 -- .../org/apache/cassandra/config/Config.java | 2 - .../cassandra/config/DatabaseDescriptor.java | 5 - .../apache/cassandra/db/BinaryMemtable.java | 160 ---------- .../cassandra/db/BinaryVerbHandler.java | 56 ---- .../cassandra/db/ColumnFamilyStore.java | 69 +---- .../org/apache/cassandra/db/IFlushable.java | 32 -- .../org/apache/cassandra/db/Memtable.java | 4 +- .../org/apache/cassandra/db/RowMutation.java | 9 - src/java/org/apache/cassandra/db/Table.java | 15 - .../cassandra/service/StorageService.java | 5 +- 14 files changed, 23 insertions(+), 674 deletions(-) delete mode 100644 examples/bmt/CassandraBulkLoader.java delete mode 100644 examples/bmt/README.txt delete mode 100644 src/java/org/apache/cassandra/db/BinaryMemtable.java delete mode 100644 src/java/org/apache/cassandra/db/BinaryVerbHandler.java delete mode 100644 src/java/org/apache/cassandra/db/IFlushable.java diff --git a/CHANGES.txt b/CHANGES.txt index 245b52ac07..499dd35147 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,7 @@ +1.0-dev + * removed binarymemtable (CASSANDRA-2692) + + 0.8.1 * CQL: - support for insert, delete in BATCH (CASSANDRA-2537) diff --git a/NEWS.txt b/NEWS.txt index 4eb8fbe897..c94da5d05d 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -1,3 +1,12 @@ +1.0 +=== + +Upgrading +--------- + - the BinaryMemtable bulk-load interface has been removed. Use the + sstableloader tool instead. + + 0.8 === diff --git a/examples/bmt/CassandraBulkLoader.java b/examples/bmt/CassandraBulkLoader.java deleted file mode 100644 index 6fb1a91e1a..0000000000 --- a/examples/bmt/CassandraBulkLoader.java +++ /dev/null @@ -1,293 +0,0 @@ -/* - * 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 deleted file mode 100644 index 78ba11c09d..0000000000 --- a/examples/bmt/README.txt +++ /dev/null @@ -1,34 +0,0 @@ -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 f76b6f22f0..2d48c2272f 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -77,8 +77,6 @@ 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 1decb8c8ec..2200e4a774 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -937,11 +937,6 @@ 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 deleted file mode 100644 index 01fa10a396..0000000000 --- a/src/java/org/apache/cassandra/db/BinaryMemtable.java +++ /dev/null @@ -1,160 +0,0 @@ -/** - * 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/BinaryVerbHandler.java b/src/java/org/apache/cassandra/db/BinaryVerbHandler.java deleted file mode 100644 index fcc317460e..0000000000 --- a/src/java/org/apache/cassandra/db/BinaryVerbHandler.java +++ /dev/null @@ -1,56 +0,0 @@ -/** - * 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.ByteArrayInputStream; -import java.io.DataInputStream; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.apache.cassandra.net.IVerbHandler; -import org.apache.cassandra.net.Message; -import org.apache.cassandra.net.MessagingService; - -public class BinaryVerbHandler implements IVerbHandler -{ - private static Logger logger_ = LoggerFactory.getLogger(BinaryVerbHandler.class); - - public void doVerb(Message message, String id) - { - byte[] bytes = message.getMessageBody(); - ByteArrayInputStream buffer = new ByteArrayInputStream(bytes); - - try - { - RowMutation rm = RowMutation.serializer().deserialize(new DataInputStream(buffer), message.getVersion()); - rm.applyBinary(); - - WriteResponse response = new WriteResponse(rm.getTable(), rm.key(), true); - Message responseMessage = WriteResponse.makeWriteResponseMessage(message, response); - if (logger_.isDebugEnabled()) - logger_.debug("binary " + rm + " applied. Sending response to " + id + "@" + message.getFrom()); - MessagingService.instance().sendReply(responseMessage, id, message.getFrom()); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } -} diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 0902997ed9..b536dbbd4e 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -25,7 +25,6 @@ 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; @@ -74,13 +73,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean private static Logger logger = LoggerFactory.getLogger(ColumnFamilyStore.class); /* - * 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, + * maybeSwitchMemtable puts Memtable.getSortedContents on the writer executor. When the write is complete, * we turn the writer into an SSTableReader and add it to ssTables_ where it is available for reads. * - * 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). + * There are two other things that maybeSwitchMemtable does. * 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 @@ -88,13 +84,6 @@ 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, @@ -136,9 +125,6 @@ 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(); @@ -256,7 +242,6 @@ 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); @@ -657,7 +642,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } final CountDownLatch latch = new CountDownLatch(icc.size()); for (ColumnFamilyStore cfs : icc) - submitFlush(cfs.data.switchMemtable(), latch, ctx); + { + Memtable memtable = cfs.data.switchMemtable(); + logger.info("Enqueuing flush of {}", memtable); + memtable.flushAndSignal(latch, flushWriter, 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 @@ -698,13 +687,6 @@ 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 Future forceFlush() { // during index build, 2ary index memtables can be dirty even if parent is not. if so, @@ -729,14 +711,6 @@ 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()) @@ -787,20 +761,6 @@ 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. @@ -992,21 +952,6 @@ 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 deleted file mode 100644 index 1be11c4e95..0000000000 --- a/src/java/org/apache/cassandra/db/IFlushable.java +++ /dev/null @@ -1,32 +0,0 @@ -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 54403e44eb..7eea278191 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, IFlushable +public class Memtable implements Comparable { private static final Logger logger = LoggerFactory.getLogger(Memtable.class); @@ -256,7 +256,7 @@ public class Memtable implements Comparable, IFlushable return ssTable; } - public void flushAndSignal(final CountDownLatch latch, ExecutorService sorter, final ExecutorService writer, final ReplayPosition context) + public void flushAndSignal(final CountDownLatch latch, ExecutorService writer, final ReplayPosition context) { writer.execute(new WrappedRunnable() { diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 328268b297..5bee2b6269 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -215,15 +215,6 @@ public class RowMutation implements IMutation, MessageProducer Table.open(table_).apply(this, false); } - /* - * This is equivalent to calling commit. Applies the changes to - * to the table that is obtained by calling Table.open(). - */ - void applyBinary() throws IOException, ExecutionException, InterruptedException - { - Table.open(table_).load(this); - } - public Message getMessage(Integer version) throws IOException { return makeRowMutationMessage(StorageService.Verb.MUTATION, version); diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 557ad698c6..f0dfb6cbea 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -669,21 +669,6 @@ public class Table return futures; } - // for binary load path. skips commitlog. - void load(RowMutation rowMutation) throws IOException - { - DecoratedKey key = StorageService.getPartitioner().decorateKey(rowMutation.key()); - for (ColumnFamily columnFamily : rowMutation.getColumnFamilies()) - { - Collection columns = columnFamily.getSortedColumns(); - for (IColumn column : columns) - { - ColumnFamilyStore cfStore = columnFamilyStores.get(ByteBufferUtil.toInt(column.name())); - cfStore.applyBinary(key, column.value()); - } - } - } - public String getDataFileLocation(long expectedSize) { String path = DatabaseDescriptor.getDataFileLocationForTable(name, expectedSize); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 6f304c7260..43099f84df 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -88,7 +88,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public enum Verb { MUTATION, - BINARY, + BINARY, // Deprecated READ_REPAIR, READ, REQUEST_RESPONSE, // client-initiated reads and writes @@ -235,7 +235,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe } /* register the verb handlers */ - MessagingService.instance().registerVerbHandlers(Verb.BINARY, new BinaryVerbHandler()); MessagingService.instance().registerVerbHandlers(Verb.MUTATION, new RowMutationVerbHandler()); MessagingService.instance().registerVerbHandlers(Verb.READ_REPAIR, new ReadRepairVerbHandler()); MessagingService.instance().registerVerbHandlers(Verb.READ, new ReadVerbHandler()); @@ -1460,8 +1459,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe { for (ColumnFamilyStore cfStore : getValidColumnFamilies(tableName, columnFamilies)) { - logger_.debug("Forcing binary flush on keyspace " + tableName + ", CF " + cfStore.getColumnFamilyName()); - cfStore.forceFlushBinary(); logger_.debug("Forcing flush on keyspace " + tableName + ", CF " + cfStore.getColumnFamilyName()); cfStore.forceBlockingFlush(); }