diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 1db1def96e..c96cbb43d1 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -61,7 +61,6 @@ import org.apache.cassandra.db.marshal.AbstractType; import org.apache.commons.collections.IteratorUtils; -import org.cliffc.high_scale_lib.NonBlockingHashMap; import com.google.common.collect.Iterators; import com.google.common.base.Predicate; @@ -84,7 +83,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 NonBlockingHashMap> memtablesPendingFlush = new NonBlockingHashMap>(); private static ExecutorService flushSorter_ = new JMXEnabledThreadPoolExecutor(1, Runtime.getRuntime().availableProcessors(), @@ -103,6 +101,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean private static final int KEY_RANGE_FILE_BUFFER_SIZE = 256 * 1024; + private Set memtablesPendingFlush = new ConcurrentSkipListSet(); + private final String table_; public final String columnFamily_; private final boolean isSuper_; @@ -132,7 +132,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean columnFamily_ = columnFamilyName; isSuper_ = isSuper; fileIndexGenerator_.set(indexValue); - memtable_ = new Memtable(table_, columnFamily_); + memtable_ = new Memtable(this); binaryMemtable_ = new AtomicReference(new BinaryMemtable(this)); if (logger_.isDebugEnabled()) @@ -377,7 +377,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance().getContext() : null; logger_.info(columnFamily_ + " has reached its threshold; switching in a fresh Memtable at " + ctx); final Condition condition = submitFlush(oldMemtable); - memtable_ = new Memtable(table_, columnFamily_); + memtable_ = new Memtable(this); // a second executor that makes sure the onMemtableFlushes get called in the right order, // while keeping the wait-for-flush (future.get) out of anything latency-sensitive. return commitLogUpdater_.submit(new WrappedRunnable() @@ -616,22 +616,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean ssTables_.replace(sstables, replacements); } - public static List getUnflushedMemtables(String cfName) - { - return new ArrayList(getMemtablesPendingFlushNotNull(cfName)); - } - - static Set getMemtablesPendingFlushNotNull(String columnFamilyName) - { - Set memtables = memtablesPendingFlush.get(columnFamilyName); - if (memtables == null) - { - memtablesPendingFlush.putIfAbsent(columnFamilyName, new ConcurrentSkipListSet()); - memtables = memtablesPendingFlush.get(columnFamilyName); // might not be the object we just put, if there was a race! - } - return memtables; - } - /** * 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 @@ -854,8 +838,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean iterators.add(iter); /* add the memtables being flushed */ - List memtables = getUnflushedMemtables(filter.getColumnFamilyName()); - for (Memtable memtable:memtables) + for (Memtable memtable : getMemtablesPendingFlush()) { iter = filter.getMemColumnIterator(memtable, getComparator()); returnCF.delete(iter.getColumnFamily()); @@ -930,7 +913,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // current memtable keys. have to go through the CFS api for locking. iterators.add(Iterators.filter(memtableKeyIterator(startWith), p)); // historical memtables - for (Memtable memtable : ColumnFamilyStore.getUnflushedMemtables(columnFamily_)) + for (Memtable memtable : memtablesPendingFlush) { iterators.add(Iterators.filter(memtable.getKeyIterator(startWith), p)); } @@ -1179,4 +1162,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean memtable_.clearUnsafe(); ssTables_.clearUnsafe(); } + + + public Set getMemtablesPendingFlush() + { + return memtablesPendingFlush; + } } diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index f4d0499f96..cd4166bc8f 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -51,16 +51,15 @@ public class Memtable implements Comparable, IFlushable private final AtomicInteger currentThroughput = new AtomicInteger(0); private final AtomicInteger currentOperations = new AtomicInteger(0); - private final String table; - private final String columnfamilyName; private final long creationTime; private final ConcurrentNavigableMap columnFamilies = new ConcurrentSkipListMap(); private final IPartitioner partitioner = StorageService.getPartitioner(); + private final ColumnFamilyStore cfs; - Memtable(String table, String cfName) + public Memtable(ColumnFamilyStore cfs) { - this.table = table; - columnfamilyName = cfName; + + this.cfs = cfs; creationTime = System.currentTimeMillis(); } @@ -147,8 +146,7 @@ public class Memtable implements Comparable, IFlushable private SSTableReader writeSortedContents() throws IOException { logger.info("Writing " + this); - ColumnFamilyStore cfStore = Table.open(table).getColumnFamilyStore(columnfamilyName); - SSTableWriter writer = new SSTableWriter(cfStore.getFlushPath(), columnFamilies.size(), StorageService.getPartitioner()); + SSTableWriter writer = new SSTableWriter(cfs.getFlushPath(), columnFamilies.size(), StorageService.getPartitioner()); DataOutputBuffer buffer = new DataOutputBuffer(); for (Map.Entry entry : columnFamilies.entrySet()) @@ -160,21 +158,20 @@ public class Memtable implements Comparable, IFlushable writer.append(entry.getKey(), buffer); } - SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table, columnfamilyName)); + SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(getTableName(), cfs.getColumnFamilyName())); logger.info("Completed flushing " + ssTable.getFilename()); return ssTable; } public void flushAndSignal(final Condition condition, ExecutorService sorter, final ExecutorService writer) { - ColumnFamilyStore.getMemtablesPendingFlushNotNull(columnfamilyName).add(this); // it's ok for the MT to briefly be both active and pendingFlush + cfs.getMemtablesPendingFlush().add(this); // it's ok for the MT to briefly be both active and pendingFlush writer.submit(new WrappedRunnable() { public void runMayThrow() throws IOException { - ColumnFamilyStore cfs = Table.open(table).getColumnFamilyStore(columnfamilyName); cfs.addSSTable(writeSortedContents()); - ColumnFamilyStore.getMemtablesPendingFlushNotNull(columnfamilyName).remove(Memtable.this); + cfs.getMemtablesPendingFlush().remove(Memtable.this); condition.signalAll(); } }); @@ -182,7 +179,7 @@ public class Memtable implements Comparable, IFlushable public String toString() { - return "Memtable(" + columnfamilyName + ")@" + hashCode(); + return "Memtable(" + cfs.getColumnFamilyName() + ")@" + hashCode(); } public Iterator getKeyIterator(DecoratedKey startWith) @@ -195,19 +192,24 @@ public class Memtable implements Comparable, IFlushable return columnFamilies.isEmpty(); } + private String getTableName() + { + return cfs.getTable().name; + } + /** * obtain an iterator of columns in this memtable in the specified order starting from a given column. */ public ColumnIterator getSliceIterator(ColumnFamily cf, SliceQueryFilter filter, AbstractType typeComparator) { - final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(table, filter.getColumnFamilyName()) : cf.cloneMeShallow(); + final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(getTableName(), filter.getColumnFamilyName()) : cf.cloneMeShallow(); final IColumn columns[] = (cf == null ? columnFamily : cf).getSortedColumns().toArray(new IColumn[columnFamily.getSortedColumns().size()]); // TODO if we are dealing with supercolumns, we need to clone them while we have the read lock since they can be modified later if (filter.reversed) ArrayUtils.reverse(columns); IColumn startIColumn; - final boolean isStandard = DatabaseDescriptor.getColumnFamilyType(table, filter.getColumnFamilyName()).equals("Standard"); + final boolean isStandard = DatabaseDescriptor.getColumnFamilyType(getTableName(), filter.getColumnFamilyName()).equals("Standard"); if (isStandard) startIColumn = new Column(filter.start); else @@ -252,8 +254,8 @@ public class Memtable implements Comparable, IFlushable public ColumnIterator getNamesIterator(final ColumnFamily cf, final NamesQueryFilter filter) { - final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(table, filter.getColumnFamilyName()) : cf.cloneMeShallow(); - final boolean isStandard = DatabaseDescriptor.getColumnFamilyType(table, filter.getColumnFamilyName()).equals("Standard"); + final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(getTableName(), filter.getColumnFamilyName()) : cf.cloneMeShallow(); + final boolean isStandard = DatabaseDescriptor.getColumnFamilyType(getTableName(), filter.getColumnFamilyName()).equals("Standard"); return new SimpleAbstractColumnIterator() {