natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix slow keycache load which blocks startup for tables with many sstables. 

Patch by Venkata Harikrishna Nukala; reviewed by Marcus Eriksson and Joseph Lynch for CASSANDRA-14898
This commit is contained in:
Venkata Harikrishna Nukala 2021-10-23 00:03:45 +05:30 committed by Joseph Lynch
parent 44d81b5d09
commit 1911a887e8
10 changed files with 330 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

4
build.xml
View File

@ -375,8 +375,8 @@
<dependency groupId="net.bytebuddy" artifactId="byte-buddy" version="${bytebuddy.version}" />
<dependency groupId="net.bytebuddy" artifactId="byte-buddy-agent" version="${bytebuddy.version}" />
<dependency groupId="org.openjdk.jmh" artifactId="jmh-core" version="1.1.1" scope="test"/>
<dependency groupId="org.openjdk.jmh" artifactId="jmh-generator-annprocess" version="1.1.1" scope="test"/>
<dependency groupId="org.openjdk.jmh" artifactId="jmh-core" version="1.21" scope="test"/>
<dependency groupId="org.openjdk.jmh" artifactId="jmh-generator-annprocess" version="1.21" scope="test"/>
<dependency groupId="org.apache.cassandra" artifactId="cassandra-all" version="${version}" />
<!--dependency groupId="org.apache.cassandra" artifactId="cassandra-thrift" version="${version}" scope="provided"/-->

7
conf/cassandra.yaml
View File

@ -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

8
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -198,8 +198,9 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
+ " does not match current schema version "
+ Schema.instance.getVersion());
ArrayDeque<Future<Pair<K, V>>> futures = new ArrayDeque<Future<Pair<K, V>>>();
while (in.available() > 0)
ArrayDeque<Future<Pair<K, V>>> 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<K extends CacheKey, V> extends InstrumentingCache<K
finally
{
FileUtils.closeQuietly(in);
cacheLoader.cleanupAfterDeserialize();
}
}
if (logger.isTraceEnabled())
@ -433,5 +435,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
void serialize(K key, DataOutputPlus out, ColumnFamilyStore cfs) throws IOException;
Future<Pair<K, V>> deserialize(DataInputPlus in, ColumnFamilyStore cfs) throws IOException;
default void cleanupAfterDeserialize() { }
}
}

2
src/java/org/apache/cassandra/config/Config.java
View File

@ -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<Config> overrideLoadConfig = null;

11
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -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;

2
src/java/org/apache/cassandra/db/lifecycle/View.java
View File

@ -137,7 +137,7 @@ public class View
case NONCOMPACTING:
return filter(sstables, (s) -> !compacting.contains(s));
case CANONICAL:
Set<SSTableReader> canonicalSSTables = new HashSet<>();
Set<SSTableReader> canonicalSSTables = new HashSet<>(sstables.size() + compacting.size());
for (SSTableReader sstable : compacting)
if (sstable.openReason != SSTableReader.OpenReason.EARLY)
canonicalSSTables.add(sstable);

38
src/java/org/apache/cassandra/service/CacheService.java
View File

@ -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<KeyCacheKey, RowIndexEntry>
{
// 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<Pair<String, String>, Map<Integer, SSTableReader>> 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<Pair<KeyCacheKey, RowIndexEntry>> 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<String, String> qualifiedName = Pair.create(cfs.metadata.ksName, cfs.metadata.cfName);
Map<Integer, SSTableReader> 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<SSTableReader> collection)
public void cleanupAfterDeserialize()
{
for (SSTableReader sstable : collection)
{
if (sstable.descriptor.generation == generation)
return sstable;
}
return null;
cachedSSTableReaders.clear();
}
}
}

137
test/microbench/org/apache/cassandra/test/microbench/CacheLoaderBench.java Normal file
View File

@ -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<ColumnFamilyStore> 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<KeyCacheKey, RowIndexEntry> keyCache = CacheService.instance.keyCache;
// serialize to file
keyCache.submitWrite(keyCache.size()).get();
}
@Setup(Level.Invocation)
public void setupKeyCache()
{
AutoSavingCache<KeyCacheKey, RowIndexEntry> keyCache = CacheService.instance.keyCache;
keyCache.clear();
}
@TearDown(Level.Trial)
public void teardown()
{
CQLTester.cleanup();
CQLTester.tearDownClass();
}
@Benchmark
public void keyCacheLoadTest() throws Throwable
{
AutoSavingCache<KeyCacheKey, RowIndexEntry> keyCache = CacheService.instance.keyCache;
keyCache.loadSavedAsync().get();
}
}

135
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -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<Pair<String, String>> 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<Pair<String, String>> tables, int numberOfRows) throws ExecutionException, InterruptedException
{
CompactionManager.instance.disableAutoCompaction();
// empty the cache
CacheService.instance.invalidateKeyCache();
assertEquals(0, CacheService.instance.keyCache.size());
for(Pair<String, String> 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<String, String> 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);