remove on-heap row cache

Removed on-heap row cache
patch by jbellis; reviewed by jasobrown for CASSANDRA-5348

CHANGES.txt

@@ -1,4 +1,5 @@
 2.0
+ * Removed on-heap row cache (CASSANDRA-5348)
  * use nanotime consistently for node-local timeouts (CASSANDRA-5581)
  * Avoid unnecessary second pass on name-based queries (CASSANDRA-5577)
  * Experimental triggers (CASSANDRA-1311)

conf/cassandra.yaml

@@ -170,24 +170,6 @@ row_cache_save_period: 0
 # Disabled by default, meaning all keys are going to be saved
 # row_cache_keys_to_save: 100
 
-# The provider for the row cache to use.
-#
-# Supported values are: ConcurrentLinkedHashCacheProvider, SerializingCacheProvider
-#
-# SerializingCacheProvider serialises the contents of the row and stores
-# it in native memory, i.e., off the JVM Heap. Serialized rows take
-# significantly less memory than "live" rows in the JVM, so you can cache
-# more rows in a given memory footprint.  And storing the cache off-heap
-# means you can use smaller heap sizes, reducing the impact of GC pauses.
-# Note however that when a row is requested from the row cache, it must be
-# deserialized into the heap for use.
-#
-# It is also valid to specify the fully-qualified class name to a class
-# that implements org.apache.cassandra.cache.IRowCacheProvider.
-#
-# Defaults to SerializingCacheProvider
-row_cache_provider: SerializingCacheProvider
-
 # The off-heap memory allocator.  Affects storage engine metadata as
 # well as caches.  Experiments show that JEMAlloc saves some memory
 # than the native GCC allocator (i.e., JEMalloc is more

src/java/org/apache/cassandra/cache/ConcurrentLinkedHashCache.java

@@ -130,9 +130,4 @@ public class ConcurrentLinkedHashCache<K extends IMeasurableMemory, V extends IM
     {
         return map.containsKey(key);
     }
-
-    public boolean isPutCopying()
-    {
-        return false;
-    }
 }

src/java/org/apache/cassandra/cache/ConcurrentLinkedHashCacheProvider.java

@@ -1,26 +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.cache;
-
-public class ConcurrentLinkedHashCacheProvider implements IRowCacheProvider
-{
-    public ICache<RowCacheKey, IRowCacheEntry> create(long capacity)
-    {
-        return ConcurrentLinkedHashCache.create(capacity);
-    }
-}

src/java/org/apache/cassandra/cache/ICache.java

@@ -51,10 +51,4 @@ public interface ICache<K, V>
     public Set<K> hotKeySet(int n);
 
     public boolean containsKey(K key);
-
-    /**
-     * @return true if the cache implementation inherently copies the cached values; otherwise,
-     * the caller should copy manually before caching shared values like Thrift ByteBuffers.
-     */
-    public boolean isPutCopying();
 }

src/java/org/apache/cassandra/cache/IRowCacheProvider.java

@@ -1,26 +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.cache;
-
-/**
- * Provides cache objects with a requested capacity.
- */
-public interface IRowCacheProvider
-{
-    public ICache<RowCacheKey, IRowCacheEntry> create(long capacity);
-}

src/java/org/apache/cassandra/cache/InstrumentingCache.java

@@ -125,11 +125,6 @@ public class InstrumentingCache<K, V>
         return map.containsKey(key);
     }
 
-    public boolean isPutCopying()
-    {
-        return map.isPutCopying();
-    }
-
     public CacheMetrics getMetrics()
     {
         return metrics;

src/java/org/apache/cassandra/cache/SerializingCache.java

@@ -249,9 +249,4 @@ public class SerializingCache<K, V> implements ICache<K, V>
     {
         return map.containsKey(key);
     }
-
-    public boolean isPutCopying()
-    {
-        return true;
-    }
 }

src/java/org/apache/cassandra/cache/SerializingCacheProvider.java

@@ -26,7 +26,7 @@ import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.io.ISerializer;
 import org.apache.cassandra.net.MessagingService;
 
-public class SerializingCacheProvider implements IRowCacheProvider
+public class SerializingCacheProvider
 {
     public ICache<RowCacheKey, IRowCacheEntry> create(long capacity)
     {

src/java/org/apache/cassandra/config/Config.java

@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.config;
 
-import org.apache.cassandra.cache.SerializingCacheProvider;
 import org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
 import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
 import org.apache.cassandra.io.util.NativeAllocator;
@@ -164,7 +163,6 @@ public class Config
     public long row_cache_size_in_mb = 0;
     public volatile int row_cache_save_period = 0;
     public int row_cache_keys_to_save = Integer.MAX_VALUE;
-    public String row_cache_provider = SerializingCacheProvider.class.getSimpleName();
     public String memory_allocator = NativeAllocator.class.getSimpleName();
     public boolean populate_io_cache_on_flush = false;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -30,7 +30,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.auth.*;
-import org.apache.cassandra.cache.IRowCacheProvider;
 import org.apache.cassandra.config.Config.RequestSchedulerId;
 import org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
 import org.apache.cassandra.config.EncryptionOptions.ServerEncryptionOptions;
@@ -79,7 +78,6 @@ public class DatabaseDescriptor
     private static RequestSchedulerOptions requestSchedulerOptions;
 
     private static long keyCacheSizeInMB;
-    private static IRowCacheProvider rowCacheProvider;
     private static IAllocator memoryAllocator;
 
     private static String localDC;
@@ -428,7 +426,6 @@ public class DatabaseDescriptor
                     + conf.key_cache_size_in_mb + "', supported values are <integer> >= 0.");
         }
 
-        rowCacheProvider = FBUtilities.newCacheProvider(conf.row_cache_provider);
         memoryAllocator = FBUtilities.newOffHeapAllocator(conf.memory_allocator);
 
         if(conf.encryption_options != null)
@@ -1166,11 +1163,6 @@ public class DatabaseDescriptor
         return conf.row_cache_keys_to_save;
     }
 
-    public static IRowCacheProvider getRowCacheProvider()
-    {
-        return rowCacheProvider;
-    }
-
     public static IAllocator getoffHeapMemoryAllocator()
     {
         return memoryAllocator;

src/java/org/apache/cassandra/db/CollationController.java

@@ -41,20 +41,15 @@ public class CollationController
 
     private final ColumnFamilyStore cfs;
     private final QueryFilter filter;
-    private final ColumnFamily.Factory factory;
     private final int gcBefore;
 
     private int sstablesIterated = 0;
 
-    public CollationController(ColumnFamilyStore cfs, boolean mutableColumns, QueryFilter filter, int gcBefore)
+    public CollationController(ColumnFamilyStore cfs, QueryFilter filter, int gcBefore)
     {
         this.cfs = cfs;
         this.filter = filter;
         this.gcBefore = gcBefore;
-
-        this.factory = mutableColumns
-                     ? AtomicSortedColumns.factory
-                     : ArrayBackedSortedColumns.factory;
     }
 
     public ColumnFamily getTopLevelColumns()
@@ -73,7 +68,7 @@ public class CollationController
     private ColumnFamily collectTimeOrderedData()
     {
         logger.trace("collectTimeOrderedData");
-        final ColumnFamily container = factory.create(cfs.metadata, filter.filter.isReversed());
+        final ColumnFamily container = ArrayBackedSortedColumns.factory.create(cfs.metadata, filter.filter.isReversed());
         List<OnDiskAtomIterator> iterators = new ArrayList<OnDiskAtomIterator>();
         Tracing.trace("Acquiring sstable references");
         ColumnFamilyStore.ViewFragment view = cfs.markReferenced(filter.key);
@@ -223,7 +218,7 @@ public class CollationController
         Tracing.trace("Acquiring sstable references");
         ColumnFamilyStore.ViewFragment view = cfs.markReferenced(filter.key);
         List<OnDiskAtomIterator> iterators = new ArrayList<OnDiskAtomIterator>(Iterables.size(view.memtables) + view.sstables.size());
-        ColumnFamily returnCF = factory.create(cfs.metadata, filter.filter.isReversed());
+        ColumnFamily returnCF = ArrayBackedSortedColumns.factory.create(cfs.metadata, filter.filter.isReversed());
 
         try
         {

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -808,29 +808,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         FBUtilities.waitOnFuture(forceFlush());
     }
 
-    public void maybeUpdateRowCache(DecoratedKey key, ColumnFamily columnFamily)
+    public void maybeUpdateRowCache(DecoratedKey key)
     {
         if (!isRowCacheEnabled())
             return;
 
         RowCacheKey cacheKey = new RowCacheKey(metadata.cfId, key);
-
-        // always invalidate a copying cache value
-        if (CacheService.instance.rowCache.isPutCopying())
-        {
-            invalidateCachedRow(cacheKey);
-            return;
-        }
-
-        // invalidate a normal cache value if it's a sentinel, so the read will retry (and include the new update)
-        IRowCacheEntry cachedRow = CacheService.instance.rowCache.getInternal(cacheKey);
-        if (cachedRow != null)
-        {
-            if (cachedRow instanceof RowCacheSentinel)
-                invalidateCachedRow(cacheKey);
-            else
-                ((ColumnFamily) cachedRow).addAll(columnFamily, HeapAllocator.instance);
-        }
+        invalidateCachedRow(cacheKey);
     }
 
     /**
@@ -846,7 +830,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
         Memtable mt = getMemtableThreadSafe();
         mt.put(key, columnFamily, indexer);
-        maybeUpdateRowCache(key, columnFamily);
+        maybeUpdateRowCache(key);
         metric.writeLatency.addNano(System.nanoTime() - start);
 
         // recompute liveRatio, if we have doubled the number of ops since last calculated
@@ -1239,7 +1223,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             {
                 // Some other read is trying to cache the value, just do a normal non-caching read
                 Tracing.trace("Row cache miss (race)");
-                return getTopLevelColumns(filter, Integer.MIN_VALUE, false);
+                return getTopLevelColumns(filter, Integer.MIN_VALUE);
             }
             Tracing.trace("Row cache hit");
             return (ColumnFamily) cached;
@@ -1251,9 +1235,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
         try
         {
-            ColumnFamily data = getTopLevelColumns(QueryFilter.getIdentityFilter(filter.key, name),
-                                                   Integer.MIN_VALUE,
-                                                   true);
+            ColumnFamily data = getTopLevelColumns(QueryFilter.getIdentityFilter(filter.key, name), Integer.MIN_VALUE);
             if (sentinelSuccess && data != null)
                 CacheService.instance.rowCache.replace(key, sentinel, data);
 
@@ -1295,7 +1277,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             }
             else
             {
-                ColumnFamily cf = getTopLevelColumns(filter, gcBefore, false);
+                ColumnFamily cf = getTopLevelColumns(filter, gcBefore);
 
                 if (cf == null)
                     return null;
@@ -1462,13 +1444,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
     }
 
-    public ColumnFamily getTopLevelColumns(QueryFilter filter, int gcBefore, boolean forCache)
+    public ColumnFamily getTopLevelColumns(QueryFilter filter, int gcBefore)
     {
         Tracing.trace("Executing single-partition query on {}", name);
-        CollationController controller = new CollationController(this,
-                                                                 forCache && !CacheService.instance.rowCache.isPutCopying(),
-                                                                 filter,
-                                                                 gcBefore);
+        CollationController controller = new CollationController(this, filter, gcBefore);
         ColumnFamily columns = controller.getTopLevelColumns();
         metric.updateSSTableIterated(controller.getSstablesIterated());
         return columns;
@@ -1727,8 +1706,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
      */
     public ColumnFamily getRawCachedRow(DecoratedKey key)
     {
-        if (!isRowCacheEnabled() || metadata.cfId == null)
-            return null; // secondary index
+        if (!isRowCacheEnabled())
+            return null;
 
         IRowCacheEntry cached = CacheService.instance.rowCache.getInternal(new RowCacheKey(metadata.cfId, key));
         return cached == null || cached instanceof RowCacheSentinel ? null : (ColumnFamily) cached;

src/java/org/apache/cassandra/db/compaction/CompactionController.java

@@ -177,20 +177,6 @@ public class CompactionController
         cfs.invalidateCachedRow(key);
     }
 
-    public void removeDeletedInCache(DecoratedKey key)
-    {
-        // For the copying cache, we'd need to re-serialize the updated cachedRow, which would be racy
-        // vs other updates.  We'll just ignore it instead, since the next update to this row will invalidate it
-        // anyway, so the odds of a "tombstones consuming memory indefinitely" problem are minimal.
-        // See https://issues.apache.org/jira/browse/CASSANDRA-3921 for more discussion.
-        if (CacheService.instance.rowCache.isPutCopying())
-            return;
-
-        ColumnFamily cachedRow = cfs.getRawCachedRow(key);
-        if (cachedRow != null)
-            ColumnFamilyStore.removeDeleted(cachedRow, gcBefore);
-    }
-
     /**
      * @return an AbstractCompactedRow implementation to write the merged rows in question.
      *

src/java/org/apache/cassandra/db/compaction/CompactionTask.java

@@ -165,9 +165,6 @@ public class CompactionTask extends AbstractCompactionTask
                     continue;
                 }
 
-                // If the row is cached, we call removeDeleted on at read time it to have coherent query returns,
-                // but if the row is not pushed out of the cache, obsolete tombstones will persist indefinitely.
-                controller.removeDeletedInCache(row.key);
                 totalkeysWritten++;
 
                 if (DatabaseDescriptor.getPreheatKeyCache())

src/java/org/apache/cassandra/service/CacheService.java

@@ -130,14 +130,12 @@ public class CacheService implements CacheServiceMBean
      */
     private AutoSavingCache<RowCacheKey, IRowCacheEntry> initRowCache()
     {
-        logger.info("Initializing row cache with capacity of {} MBs and provider {}",
-                    DatabaseDescriptor.getRowCacheSizeInMB(),
-                    DatabaseDescriptor.getRowCacheProvider().getClass().getName());
+        logger.info("Initializing row cache with capacity of {} MBs", DatabaseDescriptor.getRowCacheSizeInMB());
 
         long rowCacheInMemoryCapacity = DatabaseDescriptor.getRowCacheSizeInMB() * 1024 * 1024;
 
         // cache object
-        ICache<RowCacheKey, IRowCacheEntry> rc = DatabaseDescriptor.getRowCacheProvider().create(rowCacheInMemoryCapacity);
+        ICache<RowCacheKey, IRowCacheEntry> rc = new SerializingCacheProvider().create(rowCacheInMemoryCapacity);
         AutoSavingCache<RowCacheKey, IRowCacheEntry> rowCache = new AutoSavingCache<RowCacheKey, IRowCacheEntry>(rc, CacheType.ROW_CACHE, new RowCacheSerializer());
 
         int rowCacheKeysToSave = DatabaseDescriptor.getRowCacheKeysToSave();
@@ -302,7 +300,7 @@ public class CacheService implements CacheServiceMBean
                 public Pair<RowCacheKey, IRowCacheEntry> call() throws Exception
                 {
                     DecoratedKey key = cfs.partitioner.decorateKey(buffer);
-                    ColumnFamily data = cfs.getTopLevelColumns(QueryFilter.getIdentityFilter(key, cfs.name), Integer.MIN_VALUE, true);
+                    ColumnFamily data = cfs.getTopLevelColumns(QueryFilter.getIdentityFilter(key, cfs.name), Integer.MIN_VALUE);
                     return Pair.create(new RowCacheKey(cfs.metadata.cfId, key), (IRowCacheEntry) data);
                 }
             });
@@ -313,7 +311,7 @@ public class CacheService implements CacheServiceMBean
             for (ByteBuffer key : buffers)
             {
                 DecoratedKey dk = cfs.partitioner.decorateKey(key);
-                ColumnFamily data = cfs.getTopLevelColumns(QueryFilter.getIdentityFilter(dk, cfs.name), Integer.MIN_VALUE, true);
+                ColumnFamily data = cfs.getTopLevelColumns(QueryFilter.getIdentityFilter(dk, cfs.name), Integer.MIN_VALUE);
                 if (data != null)
                     rowCache.put(new RowCacheKey(cfs.metadata.cfId, dk), data);
             }

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -41,7 +41,6 @@ import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.cache.IRowCacheProvider;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.DecoratedKey;
@@ -500,13 +499,6 @@ public class FBUtilities
         return field;
     }
 
-    public static IRowCacheProvider newCacheProvider(String cache_provider) throws ConfigurationException
-    {
-        if (!cache_provider.contains("."))
-            cache_provider = "org.apache.cassandra.cache." + cache_provider;
-        return FBUtilities.construct(cache_provider, "row cache provider");
-    }
-
     public static <T> CloseableIterator<T> closeableIterator(Iterator<T> iterator)
     {
         return new WrappedCloseableIterator<T>(iterator);

src/java/org/apache/cassandra/utils/StatusLogger.java

@@ -26,10 +26,8 @@ import javax.management.ObjectName;
 
 import com.google.common.collect.Iterables;
 
-import org.apache.cassandra.cache.AutoSavingCache;
-import org.apache.cassandra.cache.IRowCacheEntry;
-import org.apache.cassandra.cache.KeyCacheKey;
-import org.apache.cassandra.cache.RowCacheKey;
+import org.apache.cassandra.cache.*;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -92,18 +90,19 @@ public class StatusLogger
         int keyCacheKeysToSave = DatabaseDescriptor.getKeyCacheKeysToSave();
         int rowCacheKeysToSave = DatabaseDescriptor.getRowCacheKeysToSave();
 
-        logger.info(String.format("%-25s%10s%25s%25s%65s", "Cache Type", "Size", "Capacity", "KeysToSave", "Provider"));
-        logger.info(String.format("%-25s%10s%25s%25s%65s", "KeyCache",
-                                                           keyCache.weightedSize(),
-                                                           keyCache.getCapacity(),
-                                                           keyCacheKeysToSave == Integer.MAX_VALUE ? "all" : keyCacheKeysToSave,
-                                                           ""));
+        logger.info(String.format("%-25s%10s%25s%25s",
+                                  "Cache Type", "Size", "Capacity", "KeysToSave"));
+        logger.info(String.format("%-25s%10s%25s%25s",
+                                  "KeyCache",
+                                  keyCache.weightedSize(),
+                                  keyCache.getCapacity(),
+                                  keyCacheKeysToSave == Integer.MAX_VALUE ? "all" : keyCacheKeysToSave));
 
-        logger.info(String.format("%-25s%10s%25s%25s%65s", "RowCache",
-                                                           rowCache.weightedSize(),
-                                                           rowCache.getCapacity(),
-                                                           rowCacheKeysToSave == Integer.MAX_VALUE ? "all" : rowCacheKeysToSave,
-                                                           DatabaseDescriptor.getRowCacheProvider().getClass().getName()));
+        logger.info(String.format("%-25s%10s%25s%25s",
+                                  "RowCache",
+                                  rowCache.weightedSize(),
+                                  rowCache.getCapacity(),
+                                  rowCacheKeysToSave == Integer.MAX_VALUE ? "all" : rowCacheKeysToSave));
 
         // per-CF stats
         logger.info(String.format("%-25s%20s", "ColumnFamily", "Memtable ops,data"));

test/unit/org/apache/cassandra/db/CollationControllerTest.java

@@ -70,7 +70,7 @@ public class CollationControllerTest extends SchemaLoader
         // A NamesQueryFilter goes down one code path (through collectTimeOrderedData())
         // It should only iterate the last flushed sstable, since it probably contains the most recent value for Column1
         QueryFilter filter = QueryFilter.getNamesFilter(dk, "Standard1", ByteBufferUtil.bytes("Column1"));
-        CollationController controller = new CollationController(store, false, filter, Integer.MIN_VALUE);
+        CollationController controller = new CollationController(store, filter, Integer.MIN_VALUE);
         controller.getTopLevelColumns();
         assertEquals(1, controller.getSstablesIterated());
 
@@ -78,7 +78,7 @@ public class CollationControllerTest extends SchemaLoader
         // We will read "only" the last sstable in that case, but because the 2nd sstable has a tombstone that is more
         // recent than the maxTimestamp of the very first sstable we flushed, we should only read the 2 first sstables.
         filter = QueryFilter.getIdentityFilter(dk, "Standard1");
-        controller = new CollationController(store, false, filter, Integer.MIN_VALUE);
+        controller = new CollationController(store, filter, Integer.MIN_VALUE);
         controller.getTopLevelColumns();
         assertEquals(2, controller.getSstablesIterated());
     }