diff --git a/CHANGES.txt b/CHANGES.txt index cf6a956c90..085e1ffb68 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.26: + * Fix slow keycache load which blocks startup for tables with many sstables (CASSANDRA-14898) * Fix rare NPE caused by batchlog replay / node decomission races (CASSANDRA-17049) * Allow users to view permissions of the roles they created (CASSANDRA-16902) * Fix failure handling in inter-node communication (CASSANDRA-16334) diff --git a/build.xml b/build.xml index b42c8ea56d..41f530dc10 100644 --- a/build.xml +++ b/build.xml @@ -375,8 +375,8 @@ - - + + diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index cc2a3690f4..ec2157b4a0 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -286,7 +286,12 @@ counter_cache_save_period: 7200 # If not set, the default directory is $CASSANDRA_HOME/data/saved_caches. # saved_caches_directory: /var/lib/cassandra/saved_caches -# commitlog_sync may be either "periodic" or "batch." +# Number of seconds the server will wait for each cache (row, key, etc ...) to load while starting +# the Cassandra process. Setting this to a negative value is equivalent to disabling all cache loading on startup +# while still having the cache during runtime. +# cache_load_timeout_seconds: 30 + +# commitlog_sync may be either "periodic" or "batch." # # When in batch mode, Cassandra won't ack writes until the commit log # has been fsynced to disk. It will wait diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java index 3da6352b0d..3b7a6b8299 100644 --- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java +++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java @@ -198,8 +198,9 @@ public class AutoSavingCache extends InstrumentingCache>> futures = new ArrayDeque>>(); - while (in.available() > 0) + ArrayDeque>> futures = new ArrayDeque<>(); + long loadByNanos = start + TimeUnit.SECONDS.toNanos(DatabaseDescriptor.getCacheLoadTimeout()); + while (System.nanoTime() < loadByNanos && in.available() > 0) { //ksname and cfname are serialized by the serializers in CacheService //That is delegated there because there are serializer specific conditions @@ -257,6 +258,7 @@ public class AutoSavingCache extends InstrumentingCache extends InstrumentingCache> deserialize(DataInputPlus in, ColumnFamilyStore cfs) throws IOException; + + default void cleanupAfterDeserialize() { } } } diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index a835e5aa4c..9e30306d0a 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -254,6 +254,8 @@ public class Config public volatile int counter_cache_save_period = 7200; public volatile int counter_cache_keys_to_save = Integer.MAX_VALUE; + public int cache_load_timeout_seconds = 30; + private static boolean isClientMode = false; private static Supplier overrideLoadConfig = null; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index b7708cdf3a..1795c984c6 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1979,6 +1979,17 @@ public class DatabaseDescriptor conf.counter_cache_save_period = counterCacheSavePeriod; } + public static int getCacheLoadTimeout() + { + return conf.cache_load_timeout_seconds; + } + + @VisibleForTesting + public static void setCacheLoadTimeout(int seconds) + { + conf.cache_load_timeout_seconds = seconds; + } + public static int getCounterCacheKeysToSave() { return conf.counter_cache_keys_to_save; diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java index 4b3aae0907..c33f305bf5 100644 --- a/src/java/org/apache/cassandra/db/lifecycle/View.java +++ b/src/java/org/apache/cassandra/db/lifecycle/View.java @@ -137,7 +137,7 @@ public class View case NONCOMPACTING: return filter(sstables, (s) -> !compacting.contains(s)); case CANONICAL: - Set canonicalSSTables = new HashSet<>(); + Set canonicalSSTables = new HashSet<>(sstables.size() + compacting.size()); for (SSTableReader sstable : compacting) if (sstable.openReason != SSTableReader.OpenReason.EARLY) canonicalSSTables.add(sstable); diff --git a/src/java/org/apache/cassandra/service/CacheService.java b/src/java/org/apache/cassandra/service/CacheService.java index c4d1722d97..873a319eb4 100644 --- a/src/java/org/apache/cassandra/service/CacheService.java +++ b/src/java/org/apache/cassandra/service/CacheService.java @@ -20,9 +20,12 @@ package org.apache.cassandra.service; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; @@ -441,6 +444,11 @@ public class CacheService implements CacheServiceMBean public static class KeyCacheSerializer implements CacheSerializer { + // For column families with many SSTables the linear nature of getSSTables slowed down KeyCache loading + // by orders of magnitude. So we cache the sstables once and rely on cleanupAfterDeserialize to cleanup any + // cached state we may have accumulated during the load. + Map, Map> cachedSSTableReaders = new ConcurrentHashMap<>(); + public void serialize(KeyCacheKey key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException { //Don't serialize old format entries since we didn't bother to implement serialization of both for simplicity @@ -460,6 +468,8 @@ public class CacheService implements CacheServiceMBean public Future> deserialize(DataInputPlus input, ColumnFamilyStore cfs) throws IOException { + boolean skipEntry = cfs == null || !cfs.isKeyCacheEnabled(); + //Keyspace and CF name are deserialized by AutoSaving cache and used to fetch the CFS provided as a //parameter so they aren't deserialized here, even though they are serialized by this serializer int keyLength = input.readInt(); @@ -472,7 +482,24 @@ public class CacheService implements CacheServiceMBean int generation = input.readInt(); input.readBoolean(); // backwards compatibility for "promoted indexes" boolean SSTableReader reader = null; - if (cfs == null || !cfs.isKeyCacheEnabled() || (reader = findDesc(generation, cfs.getSSTables(SSTableSet.CANONICAL))) == null) + if (!skipEntry) + { + Pair qualifiedName = Pair.create(cfs.metadata.ksName, cfs.metadata.cfName); + Map generationToSSTableReader = cachedSSTableReaders.get(qualifiedName); + if (generationToSSTableReader == null) + { + generationToSSTableReader = new HashMap<>(cfs.getLiveSSTables().size()); + for (SSTableReader ssTableReader : cfs.getSSTables(SSTableSet.CANONICAL)) + { + generationToSSTableReader.put(ssTableReader.descriptor.generation, ssTableReader); + } + + cachedSSTableReaders.putIfAbsent(qualifiedName, generationToSSTableReader); + } + reader = generationToSSTableReader.get(generation); + } + + if (skipEntry || reader == null) { // The sstable doesn't exist anymore, so we can't be sure of the exact version and assume its the current version. The only case where we'll be // wrong is during upgrade, in which case we fail at deserialization. This is not a huge deal however since 1) this is unlikely enough that @@ -488,14 +515,9 @@ public class CacheService implements CacheServiceMBean return Futures.immediateFuture(Pair.create(new KeyCacheKey(cfs.metadata.ksAndCFName, reader.descriptor, key), entry)); } - private SSTableReader findDesc(int generation, Iterable collection) + public void cleanupAfterDeserialize() { - for (SSTableReader sstable : collection) - { - if (sstable.descriptor.generation == generation) - return sstable; - } - return null; + cachedSSTableReaders.clear(); } } } diff --git a/test/microbench/org/apache/cassandra/test/microbench/CacheLoaderBench.java b/test/microbench/org/apache/cassandra/test/microbench/CacheLoaderBench.java new file mode 100644 index 0000000000..b3a713698b --- /dev/null +++ b/test/microbench/org/apache/cassandra/test/microbench/CacheLoaderBench.java @@ -0,0 +1,137 @@ +/* + * 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.test.microbench; + +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +import org.junit.Assert; + +import org.apache.cassandra.Util; +import org.apache.cassandra.cache.AutoSavingCache; +import org.apache.cassandra.cache.KeyCacheKey; +import org.apache.cassandra.config.ColumnDefinition; +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.RowIndexEntry; +import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.service.CacheService; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +@SuppressWarnings("unused") +@BenchmarkMode(Mode.SampleTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Warmup(iterations = 1, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 2, time = 2, timeUnit = TimeUnit.SECONDS) +@Fork(value = 1) +@Threads(1) +@State(Scope.Benchmark) +public class CacheLoaderBench extends CQLTester +{ + private static final int numSSTables = 1000; + private final Random random = new Random(); + + @Setup(Level.Trial) + public void setup() throws Throwable + { + CQLTester.prepareServer(); + String keyspace = createKeyspace("CREATE KEYSPACE %s with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes = false"); + String table1 = createTable(keyspace, "CREATE TABLE %s (key text PRIMARY KEY, val text)"); + String table2 = createTable(keyspace, "CREATE TABLE %s (key text PRIMARY KEY, val text)"); + + + Keyspace.system().forEach(k -> k.getColumnFamilyStores().forEach(ColumnFamilyStore::disableAutoCompaction)); + + ColumnFamilyStore cfs1 = Keyspace.open(keyspace).getColumnFamilyStore(table1); + ColumnFamilyStore cfs2 = Keyspace.open(keyspace).getColumnFamilyStore(table2); + cfs1.disableAutoCompaction(); + cfs2.disableAutoCompaction(); + + // Write a bunch of sstables to both cfs1 and cfs2 + + List columnFamilyStores = new ArrayList<>(2); + columnFamilyStores.add(cfs1); + columnFamilyStores.add(cfs2); + + logger.info("Creating {} sstables", numSSTables); + for (ColumnFamilyStore cfs: columnFamilyStores) + { + cfs.truncateBlocking(); + for (int i = 0; i < numSSTables ; i++) + { + ColumnDefinition colDef = ColumnDefinition.regularDef(cfs.metadata, ByteBufferUtil.bytes("val"), AsciiType.instance); + RowUpdateBuilder rowBuilder = new RowUpdateBuilder(cfs.metadata, System.currentTimeMillis() + random.nextInt(), "key"); + rowBuilder.add(colDef, "val1"); + rowBuilder.build().apply(); + cfs.forceBlockingFlush(); + } + + Assert.assertEquals(numSSTables, cfs.getLiveSSTables().size()); + + // preheat key cache + for (SSTableReader sstable : cfs.getLiveSSTables()) + sstable.getPosition(Util.dk("key"), SSTableReader.Operator.EQ); + } + + AutoSavingCache keyCache = CacheService.instance.keyCache; + + // serialize to file + keyCache.submitWrite(keyCache.size()).get(); + } + + @Setup(Level.Invocation) + public void setupKeyCache() + { + AutoSavingCache keyCache = CacheService.instance.keyCache; + keyCache.clear(); + } + + @TearDown(Level.Trial) + public void teardown() + { + CQLTester.cleanup(); + CQLTester.tearDownClass(); + } + + @Benchmark + public void keyCacheLoadTest() throws Throwable + { + AutoSavingCache keyCache = CacheService.instance.keyCache; + + keyCache.loadSavedAsync().get(); + } +} diff --git a/test/unit/org/apache/cassandra/db/KeyCacheTest.java b/test/unit/org/apache/cassandra/db/KeyCacheTest.java index 515d30e90c..9cb06b94db 100644 --- a/test/unit/org/apache/cassandra/db/KeyCacheTest.java +++ b/test/unit/org/apache/cassandra/db/KeyCacheTest.java @@ -17,9 +17,12 @@ */ package org.apache.cassandra.db; +import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ExecutionException; @@ -33,6 +36,8 @@ import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.cache.AutoSavingCache; +import org.apache.cassandra.cache.ICache; import org.apache.cassandra.cache.KeyCacheKey; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; @@ -42,18 +47,32 @@ import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.CacheService; +import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.concurrent.Refs; +import org.mockito.Mockito; +import org.mockito.internal.stubbing.answers.AnswersWithDelay; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; public class KeyCacheTest { private static final String KEYSPACE1 = "KeyCacheTest1"; + private static final String KEYSPACE2 = "KeyCacheTest2"; private static final String COLUMN_FAMILY1 = "Standard1"; private static final String COLUMN_FAMILY2 = "Standard2"; private static final String COLUMN_FAMILY3 = "Standard3"; + private static final String COLUMN_FAMILY7 = "Standard7"; + private static final String COLUMN_FAMILY8 = "Standard8"; + private static final String COLUMN_FAMILY9 = "Standard9"; + + private static final String COLUMN_FAMILY_K2_1 = "Standard1"; @BeforeClass @@ -64,7 +83,15 @@ public class KeyCacheTest KeyspaceParams.simple(1), SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY1), SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY2), - SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY3)); + SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY3), + SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY7), + SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY8), + SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY9)); + + SchemaLoader.createKeyspace(KEYSPACE2, + KeyspaceParams.simple(1), + SchemaLoader.standardCFMD(KEYSPACE2, COLUMN_FAMILY_K2_1)); + } @AfterClass @@ -231,6 +258,112 @@ public class KeyCacheTest assertKeyCacheSize(noEarlyOpen ? 4 : 2, KEYSPACE1, COLUMN_FAMILY1); } + @Test + public void testKeyCacheLoadNegativeCacheLoadTime() throws Exception + { + DatabaseDescriptor.setCacheLoadTimeout(-1); + String cf = COLUMN_FAMILY7; + + createAndInvalidateCache(Collections.singletonList(Pair.create(KEYSPACE1, cf)), 100); + + CacheService.instance.keyCache.loadSaved(); + + // Here max time to load cache is negative which means no time left to load cache. So the keyCache size should + // be zero after loadSaved(). + assertKeyCacheSize(0, KEYSPACE1, cf); + assertEquals(0, CacheService.instance.keyCache.size()); + } + + @Test + public void testKeyCacheLoadTwoTablesTime() throws Exception + { + DatabaseDescriptor.setCacheLoadTimeout(60); + String columnFamily1 = COLUMN_FAMILY8; + String columnFamily2 = COLUMN_FAMILY_K2_1; + int numberOfRows = 100; + List> tables = new ArrayList<>(2); + tables.add(Pair.create(KEYSPACE1, columnFamily1)); + tables.add(Pair.create(KEYSPACE2, columnFamily2)); + + createAndInvalidateCache(tables, numberOfRows); + + CacheService.instance.keyCache.loadSaved(); + + // Here max time to load cache is negative which means no time left to load cache. So the keyCache size should + // be zero after load. + assertKeyCacheSize(numberOfRows, KEYSPACE1, columnFamily1); + assertKeyCacheSize(numberOfRows, KEYSPACE2, columnFamily2); + assertEquals(numberOfRows * tables.size(), CacheService.instance.keyCache.size()); + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Test + public void testKeyCacheLoadCacheLoadTimeExceedingLimit() throws Exception + { + DatabaseDescriptor.setCacheLoadTimeout(2); + int delayMillis = 1000; + int numberOfRows = 100; + + String cf = COLUMN_FAMILY9; + + createAndInvalidateCache(Collections.singletonList(Pair.create(KEYSPACE1, cf)), numberOfRows); + + // Testing cache load. Here using custom built AutoSavingCache instance as simulating delay is not possible with + // 'CacheService.instance.keyCache'. 'AutoSavingCache.loadSaved()' is returning no.of entries loaded so we don't need + // to instantiate ICache.class. + CacheService.KeyCacheSerializer keyCacheSerializer = new CacheService.KeyCacheSerializer(); + CacheService.KeyCacheSerializer keyCacheSerializerSpy = Mockito.spy(keyCacheSerializer); + AutoSavingCache autoSavingCache = new AutoSavingCache(mock(ICache.class), + CacheService.CacheType.KEY_CACHE, + keyCacheSerializerSpy); + + doAnswer(new AnswersWithDelay(delayMillis, answer -> keyCacheSerializer.deserialize(answer.getArgument(0), + answer.getArgument(1)) )) + .when(keyCacheSerializerSpy).deserialize(any(DataInputPlus.class), any(ColumnFamilyStore.class)); + + long maxExpectedKeyCache = Math.min(numberOfRows, + 1 + TimeUnit.SECONDS.toMillis(DatabaseDescriptor.getCacheLoadTimeout()) / delayMillis); + + long keysLoaded = autoSavingCache.loadSaved(); + assertTrue(keysLoaded < maxExpectedKeyCache); + assertTrue(0 != keysLoaded); + Mockito.verify(keyCacheSerializerSpy, Mockito.times(1)).cleanupAfterDeserialize(); + } + + private void createAndInvalidateCache(List> tables, int numberOfRows) throws ExecutionException, InterruptedException + { + CompactionManager.instance.disableAutoCompaction(); + + // empty the cache + CacheService.instance.invalidateKeyCache(); + assertEquals(0, CacheService.instance.keyCache.size()); + + for(Pair entry : tables) + { + String keyspace = entry.left; + String cf = entry.right; + ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(cf); + + // insert data and force to disk + SchemaLoader.insertData(keyspace, cf, 0, numberOfRows); + store.forceBlockingFlush(); + } + for(Pair entry : tables) + { + String keyspace = entry.left; + String cf = entry.right; + // populate the cache + readData(keyspace, cf, 0, numberOfRows); + assertKeyCacheSize(numberOfRows, keyspace, cf); + } + + // force the cache to disk + CacheService.instance.keyCache.submitWrite(CacheService.instance.keyCache.size()).get(); + + CacheService.instance.invalidateKeyCache(); + assertEquals(0, CacheService.instance.keyCache.size()); + } + private static void readData(String keyspace, String columnFamily, int startRow, int numberOfRows) { ColumnFamilyStore store = Keyspace.open(keyspace).getColumnFamilyStore(columnFamily);