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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Joseph Lynch 2021-12-09 10:24:19 -05:00
parent cd73c14cec 1911a887e8
commit 1fce84f983
9 changed files with 332 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -11,6 +11,7 @@
* Reduce thread contention in CommitLogSegment and HintsBuffer (CASSANDRA-16072)
* Avoid sending CDC column if not enabled (CASSANDRA-16770)
Merged from 3.0:
* 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)

7
conf/cassandra.yaml
View File

@ -368,7 +368,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

@ -206,8 +206,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
@ -265,6 +266,7 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
finally
{
FileUtils.closeQuietly(in);
cacheLoader.cleanupAfterDeserialize();
}
}
if (logger.isTraceEnabled())
@ -441,5 +443,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

@ -269,6 +269,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

@ -2310,6 +2310,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

@ -136,7 +136,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);

40
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;
@ -440,6 +443,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
@ -461,6 +469,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,8 +482,25 @@ public class CacheService implements CacheServiceMBean
ByteBuffer key = ByteBufferUtil.read(input, keyLength);
int generation = input.readInt();
input.readBoolean(); // backwards compatibility for "promoted indexes" boolean
SSTableReader reader;
if (cfs == null || !cfs.isKeyCacheEnabled() || (reader = findDesc(generation, cfs.getSSTables(SSTableSet.CANONICAL))) == null)
SSTableReader reader = 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
@ -489,14 +516,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();
}
}

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

@ -18,12 +18,16 @@
package org.apache.cassandra.db;
import java.io.IOException;
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;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableList;
import org.junit.AfterClass;
@ -32,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.DatabaseDescriptor;
import org.apache.cassandra.db.compaction.OperationType;
@ -39,21 +45,37 @@ 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.hamcrest.Matchers;
import org.mockito.Mockito;
import org.mockito.internal.stubbing.answers.AnswersWithDelay;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThat;
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_FAMILY4 = "Standard4";
private static final String COLUMN_FAMILY5 = "Standard5";
private static final String COLUMN_FAMILY6 = "Standard6";
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
@ -67,7 +89,15 @@ public class KeyCacheTest
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY3),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY4),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY5),
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY6));
SchemaLoader.standardCFMD(KEYSPACE1, COLUMN_FAMILY6),
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
@ -308,6 +338,112 @@ public class KeyCacheTest
assertKeyCacheSize(noEarlyOpen ? 4 : 2, KEYSPACE1, cf);
}
@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();
assertThat(keysLoaded, Matchers.lessThanOrEqualTo(maxExpectedKeyCache));
assertNotEquals(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);