This is an automated email from the ASF dual-hosted git repository. benedict pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit fe9cff663b48fecdb964caaded2004e83a0c89f4 Author: Benedict Elliott Smith <[email protected]> AuthorDate: Wed Jul 28 20:03:09 2021 +0100 [CASSANDRA-16930] CEP-10 Phase 2: Improved Configuration For Controlling Determinism Co-authored-by: Benedict Elliott Smith <[email protected]> Co-authored-by: Sam Tunnicliffe <[email protected]> Co-authored-by: Caleb Rackliffe <[email protected]> --- .../org/apache/cassandra/auth/AuthKeyspace.java | 5 +- .../cassandra/concurrent/NamedThreadFactory.java | 1 + .../concurrent/SingleThreadExecutorPlus.java | 6 + .../cassandra/concurrent/SyncFutureTask.java | 17 ++- .../concurrent/ThreadPoolExecutorBuilder.java | 2 +- .../config/CassandraRelevantProperties.java | 79 ++++++++++- src/java/org/apache/cassandra/config/Config.java | 8 ++ .../cassandra/config/DatabaseDescriptor.java | 36 +++++ .../org/apache/cassandra/db/ColumnFamilyStore.java | 6 + src/java/org/apache/cassandra/db/Memtable.java | 22 ++- .../db/commitlog/BatchCommitLogService.java | 4 +- .../apache/cassandra/db/lifecycle/LogReplica.java | 10 +- .../org/apache/cassandra/gms/FailureDetector.java | 3 +- src/java/org/apache/cassandra/gms/Gossiper.java | 147 ++++++++++++++++----- .../apache/cassandra/io/sstable/Descriptor.java | 3 +- .../cassandra/io/sstable/format/SSTableReader.java | 12 +- .../org/apache/cassandra/io/util/PathUtils.java | 2 +- .../locator/AbstractReplicaCollection.java | 18 +++ .../org/apache/cassandra/net/RequestCallbacks.java | 8 +- .../apache/cassandra/schema/CompressionParams.java | 13 +- .../cassandra/schema/MigrationCoordinator.java | 3 +- src/java/org/apache/cassandra/schema/TableId.java | 5 + .../org/apache/cassandra/schema/TableMetadata.java | 5 +- .../org/apache/cassandra/service/ClientState.java | 7 +- .../apache/cassandra/service/StorageService.java | 33 ++++- .../org/apache/cassandra/utils/ByteBufferUtil.java | 2 + src/java/org/apache/cassandra/utils/Clock.java | 4 +- .../org/apache/cassandra/utils/FBUtilities.java | 12 +- src/java/org/apache/cassandra/utils/Hex.java | 17 +++ .../org/apache/cassandra/utils/LazyToString.java | 36 +++++ .../org/apache/cassandra/utils/MonotonicClock.java | 6 +- .../org/apache/cassandra/utils/NativeLibrary.java | 43 ++++-- .../org/apache/cassandra/utils/SigarLibrary.java | 4 +- src/java/org/apache/cassandra/utils/UUIDGen.java | 60 ++++++++- .../apache/cassandra/utils/concurrent/Threads.java | 135 +++++++++++++++++++ .../apache/cassandra/utils/memory/HeapPool.java | 76 ++++++++++- .../cassandra/utils/memory/MemtableAllocator.java | 2 +- .../cassandra/utils/memory/MemtablePool.java | 3 +- .../apache/cassandra/utils/memory/NativePool.java | 4 +- .../apache/cassandra/utils/memory/SlabPool.java | 4 +- .../config/DatabaseDescriptorRefTest.java | 1 + .../unit/org/apache/cassandra/db/CellSpecTest.java | 2 +- .../cassandra/db/ClusteringHeapSizeTest.java | 2 +- .../org/apache/cassandra/db/NativeCellTest.java | 2 +- .../cassandra/db/filter/ColumnFilterTest.java | 1 + .../apache/cassandra/gms/ArrivalWindowTest.java | 8 ++ 46 files changed, 780 insertions(+), 99 deletions(-) diff --git a/src/java/org/apache/cassandra/auth/AuthKeyspace.java b/src/java/org/apache/cassandra/auth/AuthKeyspace.java index a57257c..2271eff 100644 --- a/src/java/org/apache/cassandra/auth/AuthKeyspace.java +++ b/src/java/org/apache/cassandra/auth/AuthKeyspace.java @@ -19,6 +19,7 @@ package org.apache.cassandra.auth; import java.util.concurrent.TimeUnit; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; @@ -35,6 +36,8 @@ public final class AuthKeyspace { } + private static final int DEFAULT_RF = CassandraRelevantProperties.SYSTEM_AUTH_DEFAULT_RF.getInt(); + /** * Generation is used as a timestamp for automatic table creation on startup. * If you make any changes to the tables below, make sure to increment the @@ -109,7 +112,7 @@ public final class AuthKeyspace public static KeyspaceMetadata metadata() { return KeyspaceMetadata.create(SchemaConstants.AUTH_KEYSPACE_NAME, - KeyspaceParams.simple(1), + KeyspaceParams.simple(DEFAULT_RF), Tables.of(Roles, RoleMembers, RolePermissions, ResourceRoleIndex, NetworkPermissions)); } } diff --git a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java index 32df3f3..88e0d10 100644 --- a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java +++ b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java @@ -96,6 +96,7 @@ public class NamedThreadFactory implements ThreadFactory { this(id, priority, contextClassLoader, threadGroup, JVMStabilityInspector::uncaughtException); } + public NamedThreadFactory(String id, int priority, ClassLoader contextClassLoader, ThreadGroup threadGroup, Thread.UncaughtExceptionHandler uncaughtExceptionHandler) { this.id = id; diff --git a/src/java/org/apache/cassandra/concurrent/SingleThreadExecutorPlus.java b/src/java/org/apache/cassandra/concurrent/SingleThreadExecutorPlus.java index e72a6a4..eb28277 100644 --- a/src/java/org/apache/cassandra/concurrent/SingleThreadExecutorPlus.java +++ b/src/java/org/apache/cassandra/concurrent/SingleThreadExecutorPlus.java @@ -59,6 +59,12 @@ public class SingleThreadExecutorPlus extends ThreadPoolExecutorPlus implements set(false); run.run(); } + + @Override + public String toString() + { + return run.toString(); + } } SingleThreadExecutorPlus(ThreadPoolExecutorBuilder<? extends SingleThreadExecutorPlus> builder) diff --git a/src/java/org/apache/cassandra/concurrent/SyncFutureTask.java b/src/java/org/apache/cassandra/concurrent/SyncFutureTask.java index 19ca27f..4f4aa67 100644 --- a/src/java/org/apache/cassandra/concurrent/SyncFutureTask.java +++ b/src/java/org/apache/cassandra/concurrent/SyncFutureTask.java @@ -36,10 +36,21 @@ public class SyncFutureTask<T> extends SyncFuture<T> implements RunnableFuture<T public SyncFutureTask(WithResources withResources, Callable<T> call) { - this.call = () -> { - try (Closeable close = withResources.get()) + this.call = new Callable<T>() + { + @Override + public T call() throws Exception + { + try (Closeable close = withResources.get()) + { + return call.call(); + } + } + + @Override + public String toString() { - return call.call(); + return call.toString(); } }; } diff --git a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBuilder.java b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBuilder.java index 7c8dd93..5e578a5 100644 --- a/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBuilder.java +++ b/src/java/org/apache/cassandra/concurrent/ThreadPoolExecutorBuilder.java @@ -40,7 +40,7 @@ import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQu * <li>The default {@link UncaughtExceptionHandler} is inherited from {@link MetaFactory}, which in turn receives it * from the {@link ExecutorBuilderFactory} */ -class ThreadPoolExecutorBuilder<E extends ExecutorPlus> extends MetaFactory implements ExecutorBuilder<E> +public class ThreadPoolExecutorBuilder<E extends ExecutorPlus> extends MetaFactory implements ExecutorBuilder<E> { static <E extends SequentialExecutorPlus> ExecutorBuilder<E> sequential(Function<ThreadPoolExecutorBuilder<E>, E> constructor, ClassLoader contextClassLoader, ThreadGroup threadGroup, UncaughtExceptionHandler uncaughtExceptionHandler, String name) { diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 3eb5df1..b4a20da 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -136,6 +136,8 @@ public enum CassandraRelevantProperties /** mx4jport */ MX4JPORT ("mx4jport"), + RING_DELAY("cassandra.ring_delay_ms"), + /** * When bootstraping we wait for all schema versions found in gossip to be seen, and if not seen in time we fail * the bootstrap; this property will avoid failing and allow bootstrap to continue if set to true. @@ -152,6 +154,10 @@ public enum CassandraRelevantProperties */ GOSSIPER_QUARANTINE_DELAY("cassandra.gossip_quarantine_delay_ms"), + GOSSIPER_SKIP_WAITING_TO_SETTLE("cassandra.skip_wait_for_gossip_to_settle", "-1"), + + SHUTDOWN_ANNOUNCE_DELAY_IN_MS("cassandra.shutdown_announce_in_ms", "2000"), + /** * When doing a host replacement its possible that the gossip state is "empty" meaning that the endpoint is known * but the current state isn't known. If the host replacement is needed to repair this state, this property must @@ -171,6 +177,9 @@ public enum CassandraRelevantProperties LOG_DIR_AUDIT("cassandra.logdir.audit"), CONSISTENT_DIRECTORY_LISTINGS("cassandra.consistent_directory_listings", "false"), + CLOCK_GLOBAL("cassandra.clock", null), + CLOCK_MONOTONIC_APPROX("cassandra.monotonic_clock.approx", null), + CLOCK_MONOTONIC_PRECISE("cassandra.monotonic_clock.precise", null), //cassandra properties (without the "cassandra." prefix) @@ -183,9 +192,6 @@ public enum CassandraRelevantProperties DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES ("default.provide.overlapping.tombstones"), ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION ("org.apache.cassandra.disable_mbean_registration"), - //only for testing - ORG_APACHE_CASSANDRA_CONF_CASSANDRA_RELEVANT_PROPERTIES_TEST("org.apache.cassandra.conf.CassandraRelevantPropertiesTest"), - ORG_APACHE_CASSANDRA_DB_VIRTUAL_SYSTEM_PROPERTIES_TABLE_TEST("org.apache.cassandra.db.virtual.SystemPropertiesTableTest"), /** This property indicates whether disable_mbean_registration is true */ IS_DISABLED_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"), @@ -200,7 +206,39 @@ public enum CassandraRelevantProperties SNAPSHOT_MIN_ALLOWED_TTL_SECONDS("cassandra.snapshot.min_allowed_ttl_seconds", "60"), /** what class to use for mbean registeration */ - MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class"); + MBEAN_REGISTRATION_CLASS("org.apache.cassandra.mbean_registration_class"), + + BATCH_COMMIT_LOG_SYNC_INTERVAL("cassandra.batch_commitlog_sync_interval_millis", "1000"), + + SYSTEM_AUTH_DEFAULT_RF("cassandra.system_auth.default_rf", "1"), + + MEMTABLE_OVERHEAD_SIZE("cassandra.memtable.row_overhead_size", "-1"), + MEMTABLE_OVERHEAD_COMPUTE_STEPS("cassandra.memtable_row_overhead_computation_step", "100000"), + MIGRATION_DELAY("cassandra.migration_delay_ms", "60000"), + + PAXOS_REPAIR_RETRY_TIMEOUT_IN_MS("cassandra.paxos_repair_retry_timeout_millis", "60000"), + + // properties for debugging simulator ASM output + TEST_SIMULATOR_PRINT_ASM("cassandra.test.simulator.print_asm", "none"), + TEST_SIMULATOR_PRINT_ASM_TYPES("cassandra.test.simulator.print_asm_types", ""), + + // determinism properties for testing + DETERMINISM_SSTABLE_COMPRESSION_DEFAULT("cassandra.sstable_compression_default", "true"), + DETERMINISM_CONSISTENT_DIRECTORY_LISTINGS("cassandra.consistent_directory_listings", "false"), + DETERMINISM_UNSAFE_UUID_NODE("cassandra.unsafe.deterministicuuidnode", "false"), + + // properties to disable certain behaviours for testing + DISABLE_GOSSIP_ENDPOINT_REMOVAL("cassandra.gossip.disable_endpoint_removal"), + IGNORE_MISSING_NATIVE_FILE_HINTS("cassandra.require_native_file_hints", "false"), + DISABLE_SSTABLE_ACTIVITY_TRACKING("cassandra.sstable_activity_tracking", "true"), + TEST_IGNORE_SIGAR("cassandra.test.ignore_sigar", "false"), + PAXOS_EXECUTE_ON_SELF("cassandra.paxos.use_self_execution", "true"), + + // for specific tests + ORG_APACHE_CASSANDRA_CONF_CASSANDRA_RELEVANT_PROPERTIES_TEST("org.apache.cassandra.conf.CassandraRelevantPropertiesTest"), + ORG_APACHE_CASSANDRA_DB_VIRTUAL_SYSTEM_PROPERTIES_TABLE_TEST("org.apache.cassandra.db.virtual.SystemPropertiesTableTest"), + + ; CassandraRelevantProperties(String key, String defaultVal) @@ -302,6 +340,17 @@ public enum CassandraRelevantProperties /** * Gets the value of a system property as a int. + * @return system property int value if it exists, defaultValue otherwise. + */ + public long getLong() + { + String value = System.getProperty(key); + + return LONG_CONVERTER.convert(value == null ? defaultVal : value); + } + + /** + * Gets the value of a system property as a int. * @return system property int value if it exists, overrideDefaultValue otherwise. */ public int getInt(int overrideDefaultValue) @@ -322,6 +371,15 @@ public enum CassandraRelevantProperties System.setProperty(key, Integer.toString(value)); } + /** + * Sets the value into system properties. + * @param value to set + */ + public void setLong(long value) + { + System.setProperty(key, Long.toString(value)); + } + private interface PropertyConverter<T> { T convert(String value); @@ -344,6 +402,19 @@ public enum CassandraRelevantProperties } }; + private static final PropertyConverter<Long> LONG_CONVERTER = value -> + { + try + { + return Long.decode(value); + } + catch (NumberFormatException e) + { + throw new ConfigurationException(String.format("Invalid value for system property: " + + "expected integer value but got '%s'", value)); + } + }; + /** * @return whether a system property is present or not. */ diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index caa4965..8c498e4 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -81,6 +81,8 @@ public class Config public DiskFailurePolicy disk_failure_policy = DiskFailurePolicy.ignore; public CommitFailurePolicy commit_failure_policy = CommitFailurePolicy.stop; + public volatile boolean use_deterministic_table_id = false; + /* initial token in the ring */ public String initial_token; public Integer num_tokens; @@ -118,6 +120,7 @@ public class Config public int concurrent_writes = 32; public int concurrent_counter_writes = 32; public int concurrent_materialized_view_writes = 32; + public int available_processors = -1; @Deprecated public Integer concurrent_replicates = null; @@ -280,6 +283,8 @@ public class Config public int dynamic_snitch_reset_interval_in_ms = 600000; public double dynamic_snitch_badness_threshold = 1.0; + public String failure_detector = "FailureDetector"; + public EncryptionOptions.ServerEncryptionOptions server_encryption_options = new EncryptionOptions.ServerEncryptionOptions(); public EncryptionOptions client_encryption_options = new EncryptionOptions(); @@ -313,6 +318,8 @@ public class Config public volatile int counter_cache_save_period = 7200; public volatile int counter_cache_keys_to_save = Integer.MAX_VALUE; + public Long paxos_cache_size_in_mb = null; + private static boolean isClientMode = false; private static Supplier<Config> overrideLoadConfig = null; @@ -620,6 +627,7 @@ public class Config public enum MemtableAllocationType { unslabbed_heap_buffers, + unslabbed_heap_buffers_logged, heap_buffers, offheap_buffers, offheap_objects diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 878c9ec..9d08a62 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -32,6 +32,7 @@ import com.google.common.primitives.Ints; import com.google.common.primitives.Longs; import com.google.common.util.concurrent.RateLimiter; +import org.apache.cassandra.gms.IFailureDetector; import org.apache.cassandra.io.util.File; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -103,6 +104,7 @@ public class DatabaseDescriptor */ static final long LOWEST_ACCEPTED_TIMEOUT = 10L; + private static Supplier<IFailureDetector> newFailureDetector; private static IEndpointSnitch snitch; private static InetAddress listenAddress; // leave null so we can fall through to getLocalHost private static InetAddress broadcastAddress; @@ -1131,6 +1133,7 @@ public class DatabaseDescriptor return 1; return 0; }; + newFailureDetector = () -> createFailureDetector(conf.failure_detector); } // definitely not safe for tools + clients - implicitly instantiates schema @@ -1187,6 +1190,14 @@ public class DatabaseDescriptor return dynamic ? new DynamicEndpointSnitch(snitch) : snitch; } + private static IFailureDetector createFailureDetector(String detectorClassName) throws ConfigurationException + { + if (!detectorClassName.contains(".")) + detectorClassName = "org.apache.cassandra.gms." + detectorClassName; + IFailureDetector detector = FBUtilities.construct(detectorClassName, "failure detector"); + return detector; + } + public static IAuthenticator getAuthenticator() { return authenticator; @@ -1405,6 +1416,16 @@ public class DatabaseDescriptor snitch = eps; } + public static IFailureDetector newFailureDetector() + { + return newFailureDetector.get(); + } + + public static void setDefaultFailureDetector() + { + newFailureDetector = () -> createFailureDetector("FailureDetector"); + } + public static int getColumnIndexSize() { return (int) ByteUnit.KIBI_BYTES.toBytes(conf.column_index_size_in_kb); @@ -1734,6 +1755,11 @@ public class DatabaseDescriptor return conf.memtable_flush_writers; } + public static int getAvailableProcessors() + { + return conf == null ? -1 : conf.available_processors; + } + public static int getConcurrentCompactors() { return conf.concurrent_compactors; @@ -2437,6 +2463,16 @@ public class DatabaseDescriptor return conf.hinted_handoff_disabled_datacenters; } + public static boolean useDeterministicTableID() + { + return conf != null && conf.use_deterministic_table_id; + } + + public static void useDeterministicTableID(boolean value) + { + conf.use_deterministic_table_id = value; + } + public static void enableHintsForDC(String dc) { conf.hinted_handoff_disabled_datacenters.remove(dc); diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 8a60bff..5eabda7 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1228,6 +1228,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } }, reclaimExecutor); } + + @Override + public String toString() + { + return "Flush " + keyspace + '.' + name; + } } // atomically set the upper bound for the commit log diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 6e23cec..c755591 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -76,6 +76,8 @@ import org.apache.cassandra.utils.memory.NativePool; import org.apache.cassandra.utils.memory.SlabPool; import static org.apache.cassandra.utils.Clock.Global.nanoTime; +import static org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_OVERHEAD_COMPUTE_STEPS; +import static org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_OVERHEAD_SIZE; public class Memtable implements Comparable<Memtable> { @@ -91,6 +93,8 @@ public class Memtable implements Comparable<Memtable> final MemtableCleaner cleaner = ColumnFamilyStore::flushLargestMemtable; switch (DatabaseDescriptor.getMemtableAllocationType()) { + case unslabbed_heap_buffers_logged: + return new HeapPool.Logged(heapLimit, cleaningThreshold, cleaner); case unslabbed_heap_buffers: return new HeapPool(heapLimit, cleaningThreshold, cleaner); case heap_buffers: @@ -104,7 +108,13 @@ public class Memtable implements Comparable<Memtable> } } - private static final int ROW_OVERHEAD_HEAP_SIZE = estimateRowOverhead(Integer.parseInt(System.getProperty("cassandra.memtable_row_overhead_computation_step", "100000"))); + private static final int ROW_OVERHEAD_HEAP_SIZE; + static + { + int userDefinedOverhead = MEMTABLE_OVERHEAD_SIZE.getInt(-1); + if (userDefinedOverhead > 0) ROW_OVERHEAD_HEAP_SIZE = userDefinedOverhead; + else ROW_OVERHEAD_HEAP_SIZE = estimateRowOverhead(MEMTABLE_OVERHEAD_COMPUTE_STEPS.getInt()); + } private final MemtableAllocator allocator; private final AtomicLong liveDataSize = new AtomicLong(0); @@ -157,7 +167,7 @@ public class Memtable implements Comparable<Memtable> { this.cfs = cfs; this.commitLogLowerBound = commitLogLowerBound; - this.allocator = MEMORY_POOL.newAllocator(); + this.allocator = MEMORY_POOL.newAllocator(cfs); this.initialComparator = cfs.metadata().comparator; this.cfs.scheduleFlush(); this.columnsCollector = new ColumnsCollector(cfs.metadata().regularAndStaticColumns()); @@ -531,6 +541,12 @@ public class Memtable implements Comparable<Memtable> writeSortedContents(); return writer; } + + @Override + public String toString() + { + return "Flush " + cfs.keyspace + '.' + cfs.name; + } } private static int estimateRowOverhead(final int count) @@ -539,7 +555,7 @@ public class Memtable implements Comparable<Memtable> try (final OpOrder.Group group = new OpOrder().start()) { int rowOverhead; - MemtableAllocator allocator = MEMORY_POOL.newAllocator(); + MemtableAllocator allocator = MEMORY_POOL.newAllocator(null); ConcurrentNavigableMap<PartitionPosition, Object> partitions = new ConcurrentSkipListMap<>(); final Object val = new Object(); for (int i = 0 ; i < count ; i++) diff --git a/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java index 78bf30c..e913e67 100644 --- a/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/BatchCommitLogService.java @@ -17,6 +17,8 @@ */ package org.apache.cassandra.db.commitlog; +import static org.apache.cassandra.config.CassandraRelevantProperties.BATCH_COMMIT_LOG_SYNC_INTERVAL; + class BatchCommitLogService extends AbstractCommitLogService { /** @@ -24,7 +26,7 @@ class BatchCommitLogService extends AbstractCommitLogService * the disk sync. Instead we trigger it explicitly in {@link #maybeWaitForSync(CommitLogSegment.Allocation)}. * This value here is largely irrelevant, but should high enough so the sync thread is not continually waking up. */ - private static final int POLL_TIME_MILLIS = 1000; + private static final int POLL_TIME_MILLIS = BATCH_COMMIT_LOG_SYNC_INTERVAL.getInt(); public BatchCommitLogService(CommitLog commitLog) { diff --git a/src/java/org/apache/cassandra/db/lifecycle/LogReplica.java b/src/java/org/apache/cassandra/db/lifecycle/LogReplica.java index efd56d8..8be5031 100644 --- a/src/java/org/apache/cassandra/db/lifecycle/LogReplica.java +++ b/src/java/org/apache/cassandra/db/lifecycle/LogReplica.java @@ -33,6 +33,8 @@ import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.NativeLibrary; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORE_MISSING_NATIVE_FILE_HINTS; + /** * Because a column family may have sstables on different disks and disks can * be removed, we duplicate log files into many replicas so as to have a file @@ -47,6 +49,7 @@ import org.apache.cassandra.utils.NativeLibrary; final class LogReplica implements AutoCloseable { private static final Logger logger = LoggerFactory.getLogger(LogReplica.class); + private static final boolean REQUIRE_FD = !FBUtilities.isWindows && !IGNORE_MISSING_NATIVE_FILE_HINTS.getBoolean(); private final File file; private int directoryDescriptor; @@ -55,7 +58,7 @@ final class LogReplica implements AutoCloseable static LogReplica create(File directory, String fileName) { int folderFD = NativeLibrary.tryOpenDirectory(directory.path()); - if (folderFD == -1 && !FBUtilities.isWindows) + if (folderFD == -1 && REQUIRE_FD) throw new FSReadError(new IOException(String.format("Invalid folder descriptor trying to create log replica %s", directory.path())), directory.path()); return new LogReplica(new File(fileName), folderFD); @@ -181,4 +184,9 @@ final class LogReplica implements AutoCloseable str.append(System.lineSeparator()); } } + + public int hashCode() + { + return file.hashCode(); + } } diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index 522e082..40a2de5 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -40,6 +40,7 @@ import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; import static org.apache.cassandra.config.CassandraRelevantProperties.LINE_SEPARATOR; +import static org.apache.cassandra.config.DatabaseDescriptor.newFailureDetector; import static org.apache.cassandra.utils.MonotonicClock.preciseTime; /** @@ -71,7 +72,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean return DEFAULT_MAX_PAUSE; } - public static final IFailureDetector instance = new FailureDetector(); + public static final IFailureDetector instance = newFailureDetector(); public static final Predicate<InetAddressAndPort> isEndpointAlive = instance::isAlive; public static final Predicate<Replica> isReplicaAlive = r -> isEndpointAlive.test(r.endpoint()); diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index e2ebedc..5630b5b 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -59,7 +59,6 @@ import org.apache.cassandra.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import io.netty.util.concurrent.FastThreadLocal; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; @@ -68,21 +67,23 @@ import org.apache.cassandra.net.RequestCallback; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.utils.CassandraVersion; -import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; -import org.apache.cassandra.utils.MBeanWrapper; -import org.apache.cassandra.utils.NoSpamLogger; -import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.RecomputingSupplier; import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; +import org.apache.cassandra.utils.concurrent.NotScheduledFuture; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; +import static org.apache.cassandra.config.CassandraRelevantProperties.DISABLE_GOSSIP_ENDPOINT_REMOVAL; import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY; +import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_SKIP_WAITING_TO_SETTLE; +import static org.apache.cassandra.config.CassandraRelevantProperties.SHUTDOWN_ANNOUNCE_DELAY_IN_MS; +import static org.apache.cassandra.config.DatabaseDescriptor.getClusterName; +import static org.apache.cassandra.config.DatabaseDescriptor.getPartitionerName; import static org.apache.cassandra.net.NoPayload.noPayload; import static org.apache.cassandra.net.Verb.ECHO_REQ; import static org.apache.cassandra.net.Verb.GOSSIP_DIGEST_SYN; +import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; import static org.apache.cassandra.utils.Clock.Global.nanoTime; @@ -253,6 +254,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } private static final boolean disableThreadValidation = Boolean.getBoolean(Props.DISABLE_THREAD_VALIDATION); + private static volatile boolean disableEndpointRemoval = DISABLE_GOSSIP_ENDPOINT_REMOVAL.getBoolean(); private static long getVeryLongTime() { @@ -298,16 +300,17 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean taskLock.lock(); /* Update the local heartbeat counter. */ - endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat(); + endpointStateMap.get(getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat(); if (logger.isTraceEnabled()) logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion()); final List<GossipDigest> gDigests = new ArrayList<>(); + Gossiper.instance.makeRandomGossipDigest(gDigests); if (gDigests.size() > 0) { - GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(), - DatabaseDescriptor.getPartitionerName(), + GossipDigestSyn digestSynMessage = new GossipDigestSyn(getClusterName(), + getPartitionerName(), gDigests); Message<GossipDigestSyn> message = Message.out(GOSSIP_DIGEST_SYN, digestSynMessage); /* Gossip to some random live member */ @@ -453,8 +456,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public Set<InetAddressAndPort> getLiveMembers() { Set<InetAddressAndPort> liveMembers = new HashSet<>(liveEndpoints); - if (!liveMembers.contains(FBUtilities.getBroadcastAddressAndPort())) - liveMembers.add(FBUtilities.getBroadcastAddressAndPort()); + if (!liveMembers.contains(getBroadcastAddressAndPort())) + liveMembers.add(getBroadcastAddressAndPort()); return liveMembers; } @@ -532,7 +535,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return; } - FutureTask task = new FutureTask<>(runnable); + FutureTask<?> task = new FutureTask<>(runnable); Stage.GOSSIP.execute(task); try { @@ -652,6 +655,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.warn("Seeds list is now empty!"); } + if (disableEndpointRemoval) + return; + liveEndpoints.remove(endpoint); unreachableEndpoints.remove(endpoint); MessagingService.instance().versions.reset(endpoint); @@ -690,6 +696,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean */ private void quarantineEndpoint(InetAddressAndPort endpoint, long quarantineExpiration) { + if (disableEndpointRemoval) + return; justRemovedEndpoints.put(endpoint, quarantineExpiration); GossiperDiagnostics.quarantinedEndpoint(this, endpoint, quarantineExpiration); } @@ -701,7 +709,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public void replacementQuarantine(InetAddressAndPort endpoint) { // remember, quarantineEndpoint will effectively already add QUARANTINE_DELAY, so this is 2x - logger.debug(""); quarantineEndpoint(endpoint, currentTimeMillis() + QUARANTINE_DELAY); GossiperDiagnostics.replacementQuarantine(this, endpoint); } @@ -951,7 +958,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean int size = seeds.size(); if (size > 0) { - if (size == 1 && seeds.contains(FBUtilities.getBroadcastAddressAndPort())) + if (size == 1 && seeds.contains(getBroadcastAddressAndPort())) { return; } @@ -1064,7 +1071,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean Set<InetAddressAndPort> eps = endpointStateMap.keySet(); for (InetAddressAndPort endpoint : eps) { - if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort())) + if (endpoint.equals(getBroadcastAddressAndPort())) continue; FailureDetector.instance.interpret(endpoint); @@ -1264,6 +1271,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean void notifyFailureDetector(InetAddressAndPort endpoint, EndpointState remoteEndpointState) { + if (remoteEndpointState == null) + return; + EndpointState localEndpointState = endpointStateMap.get(endpoint); /* * If the local endpoint state exists then report to the FD only @@ -1365,8 +1375,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private void silentlyMarkDead(InetAddressAndPort addr, EndpointState localState) { localState.markDead(); - liveEndpoints.remove(addr); - unreachableEndpoints.put(addr, nanoTime()); + if (!disableEndpointRemoval) + { + liveEndpoints.remove(addr); + unreachableEndpoints.put(addr, nanoTime()); + } } /** @@ -1485,7 +1498,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean for (Entry<InetAddressAndPort, EndpointState> entry : epStateMap.entrySet()) { InetAddressAndPort ep = entry.getKey(); - if ( ep.equals(FBUtilities.getBroadcastAddressAndPort()) && !isInShadowRound()) + if (ep.equals(getBroadcastAddressAndPort()) && !isInShadowRound()) continue; if (justRemovedEndpoints.containsKey(ep)) { @@ -1792,7 +1805,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean buildSeedsList(); /* initialize the heartbeat state for this localEndpoint */ maybeInitializeLocalState(generationNbr); - EndpointState localState = endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()); + EndpointState localState = endpointStateMap.get(getBroadcastAddressAndPort()); localState.addApplicationStates(preloadLocalStates); minVersionSupplier.recompute(); @@ -1839,7 +1852,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean if (seeds.isEmpty() && peers.isEmpty()) return endpointShadowStateMap; - boolean isSeed = DatabaseDescriptor.getSeeds().contains(FBUtilities.getBroadcastAddressAndPort()); + boolean isSeed = DatabaseDescriptor.getSeeds().contains(getBroadcastAddressAndPort()); // We double RING_DELAY if we're not a seed to increase chance of successful startup during a full cluster bounce, // giving the seeds a chance to startup before we fail the shadow round int shadowRoundDelay = isSeed ? StorageService.RING_DELAY : StorageService.RING_DELAY * 2; @@ -1847,9 +1860,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean endpointShadowStateMap.clear(); // send a completely empty syn List<GossipDigest> gDigests = new ArrayList<>(); - GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(), - DatabaseDescriptor.getPartitionerName(), - gDigests); + GossipDigestSyn digestSynMessage = new GossipDigestSyn(getClusterName(), getPartitionerName(), gDigests); Message<GossipDigestSyn> message = Message.out(GOSSIP_DIGEST_SYN, digestSynMessage); inShadowRound = true; @@ -1905,7 +1916,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { for (InetAddressAndPort seed : DatabaseDescriptor.getSeeds()) { - if (seed.equals(FBUtilities.getBroadcastAddressAndPort())) + if (seed.equals(getBroadcastAddressAndPort())) continue; seeds.add(seed); } @@ -1924,7 +1935,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { for (InetAddressAndPort seed : DatabaseDescriptor.getSeeds()) { - if (seed.equals(FBUtilities.getBroadcastAddressAndPort())) + if (seed.equals(getBroadcastAddressAndPort())) continue; tmp.add(seed); } @@ -1978,12 +1989,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean HeartBeatState hbState = new HeartBeatState(generationNbr); EndpointState localState = new EndpointState(hbState); localState.markAlive(); - endpointStateMap.putIfAbsent(FBUtilities.getBroadcastAddressAndPort(), localState); + endpointStateMap.putIfAbsent(getBroadcastAddressAndPort(), localState); } public void forceNewerGeneration() { - EndpointState epstate = endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()); + EndpointState epstate = endpointStateMap.get(getBroadcastAddressAndPort()); epstate.getHeartBeatState().forceNewerGenerationUnsafe(); } @@ -1994,7 +2005,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public void addSavedEndpoint(InetAddressAndPort ep) { checkProperThreadForStateMutation(); - if (ep.equals(FBUtilities.getBroadcastAddressAndPort())) + if (ep.equals(getBroadcastAddressAndPort())) { logger.debug("Attempt to add self as saved endpoint"); return; @@ -2023,7 +2034,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private void addLocalApplicationStateInternal(ApplicationState state, VersionedValue value) { assert taskLock.isHeldByCurrentThread(); - InetAddressAndPort epAddr = FBUtilities.getBroadcastAddressAndPort(); + InetAddressAndPort epAddr = getBroadcastAddressAndPort(); EndpointState epState = endpointStateMap.get(epAddr); assert epState != null : "Can't find endpoint state for " + epAddr; // Fire "before change" notifications: @@ -2061,7 +2072,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public void stop() { - EndpointState mystate = endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()); + EndpointState mystate = endpointStateMap.get(getBroadcastAddressAndPort()); if (mystate != null && !isSilentShutdownState(mystate) && StorageService.instance.isJoined()) { logger.info("Announcing shutdown"); @@ -2070,7 +2081,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean Message message = Message.out(Verb.GOSSIP_SHUTDOWN, noPayload); for (InetAddressAndPort ep : liveEndpoints) MessagingService.instance().send(message, ep); - Uninterruptibles.sleepUninterruptibly(Integer.getInteger("cassandra.shutdown_announce_in_ms", 2000), TimeUnit.MILLISECONDS); + Uninterruptibles.sleepUninterruptibly(SHUTDOWN_ANNOUNCE_DELAY_IN_MS.getInt(), TimeUnit.MILLISECONDS); } else logger.warn("No local state, state is in silent shutdown, or node hasn't joined, not announcing shutdown"); @@ -2175,6 +2186,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean Map<ApplicationState, VersionedValue> states = new EnumMap<>(ApplicationState.class); states.put(ApplicationState.NET_VERSION, StorageService.instance.valueFactory.networkVersion(netVersion)); states.put(ApplicationState.HOST_ID, StorageService.instance.valueFactory.hostId(uuid)); + states.put(ApplicationState.RPC_ADDRESS, StorageService.instance.valueFactory.rpcaddress(addr.getAddress())); + states.put(ApplicationState.INTERNAL_ADDRESS_AND_PORT, StorageService.instance.valueFactory.internalAddressAndPort(addr)); + states.put(ApplicationState.RELEASE_VERSION, StorageService.instance.valueFactory.releaseVersion()); localState.addApplicationStates(states); } @@ -2246,7 +2260,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean public static void waitToSettle() { - int forceAfter = Integer.getInteger("cassandra.skip_wait_for_gossip_to_settle", -1); + int forceAfter = GOSSIPER_SKIP_WAITING_TO_SETTLE.getInt(); if (forceAfter == 0) { return; @@ -2403,7 +2417,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean CassandraVersion minVersion = null; for (InetAddressAndPort addr : Iterables.concat(Gossiper.instance.getLiveMembers(), - Gossiper.instance.getUnreachableMembers())) + Gossiper.instance.getUnreachableMembers())) { String versionString = getReleaseVersionString(addr); // Raced with changes to gossip state, wait until next iteration @@ -2432,4 +2446,71 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return minVersion; } + + public void unsafeSetEnabled() + { + scheduledGossipTask = new NotScheduledFuture<>(); + firstSynSendAt = 1; + } + + public Collection<InetAddressAndPort> unsafeClearRemoteState() + { + List<InetAddressAndPort> removed = new ArrayList<>(); + for (InetAddressAndPort ep : endpointStateMap.keySet()) + { + if (ep.equals(getBroadcastAddressAndPort())) + continue; + + for (IEndpointStateChangeSubscriber subscriber : subscribers) + subscriber.onRemove(ep); + + removed.add(ep); + } + this.endpointStateMap.keySet().retainAll(Collections.singleton(getBroadcastAddressAndPort())); + this.endpointShadowStateMap.keySet().retainAll(Collections.singleton(getBroadcastAddressAndPort())); + this.expireTimeEndpointMap.keySet().retainAll(Collections.singleton(getBroadcastAddressAndPort())); + this.justRemovedEndpoints.keySet().retainAll(Collections.singleton(getBroadcastAddressAndPort())); + this.unreachableEndpoints.keySet().retainAll(Collections.singleton(getBroadcastAddressAndPort())); + return removed; + } + + public void unsafeGossipWith(InetAddressAndPort ep) + { + /* Update the local heartbeat counter. */ + EndpointState epState = endpointStateMap.get(getBroadcastAddressAndPort()); + if (epState != null) + { + epState.getHeartBeatState().updateHeartBeat(); + if (logger.isTraceEnabled()) + logger.trace("My heartbeat is now {}", epState.getHeartBeatState().getHeartBeatVersion()); + } + + final List<GossipDigest> gDigests = new ArrayList<GossipDigest>(); + Gossiper.instance.makeRandomGossipDigest(gDigests); + + GossipDigestSyn digestSynMessage = new GossipDigestSyn(getClusterName(), + getPartitionerName(), + gDigests); + Message<GossipDigestSyn> message = Message.out(GOSSIP_DIGEST_SYN, digestSynMessage); + + MessagingService.instance().send(message, ep); + } + + public void unsafeSendShutdown(InetAddressAndPort to) + { + Message<?> message = Message.out(Verb.GOSSIP_SHUTDOWN, noPayload); + MessagingService.instance().send(message, to); + } + + public void unsafeSendLocalEndpointStateTo(InetAddressAndPort ep) + { + /* Update the local heartbeat counter. */ + EndpointState epState = endpointStateMap.get(getBroadcastAddressAndPort()); + if (epState == null) + throw new IllegalStateException(); + + GossipDigestAck2 digestAck2Message = new GossipDigestAck2(Collections.singletonMap(getBroadcastAddressAndPort(), epState)); + Message<GossipDigestAck2> message = Message.out(Verb.GOSSIP_DIGEST_ACK2, digestAck2Message); + MessagingService.instance().send(message, ep); + } } diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java index faca487..f4e7f00 100644 --- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java +++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java @@ -96,7 +96,8 @@ public class Descriptor this.generation = generation; this.formatType = formatType; - hashCode = Objects.hashCode(version, this.directory, generation, ksname, cfname, formatType); + // directory is unnecessary for hashCode, and for simulator consistency we do not include it + hashCode = Objects.hashCode(version, generation, ksname, cfname, formatType); } public Descriptor withGeneration(int newGeneration) diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index ebdb2ef..4af5743 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -144,6 +144,8 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS { private static final Logger logger = LoggerFactory.getLogger(SSTableReader.class); + private static final boolean TRACK_ACTIVITY = CassandraRelevantProperties.DISABLE_SSTABLE_ACTIVITY_TRACKING.getBoolean(); + private static final ScheduledExecutorPlus syncExecutor = initSyncExecutor(); private static ScheduledExecutorPlus initSyncExecutor() { @@ -1981,7 +1983,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS void setup(boolean trackHotness) { - tidy.setup(this, trackHotness); + tidy.setup(this, TRACK_ACTIVITY && trackHotness); this.readMeter = tidy.global.readMeter; } @@ -2097,6 +2099,12 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS if (logger.isTraceEnabled()) logger.trace("Async instance tidier for {}, completed", descriptor); } + + @Override + public String toString() + { + return "Tidy " + descriptor.ksname + '.' + descriptor.cfname + '-' + descriptor.generation; + } }); } @@ -2150,7 +2158,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS // Don't track read rates for tables in the system keyspace and don't bother trying to load or persist // the read meter when in client mode. // Also, do not track read rates when running in client or tools mode (syncExecuter isn't available in these modes) - if (SchemaConstants.isLocalSystemKeyspace(desc.ksname) || DatabaseDescriptor.isClientOrToolInitialized()) + if (!TRACK_ACTIVITY || SchemaConstants.isLocalSystemKeyspace(desc.ksname) || DatabaseDescriptor.isClientOrToolInitialized()) { readMeter = null; readMeterSyncFuture = NULL; diff --git a/src/java/org/apache/cassandra/io/util/PathUtils.java b/src/java/org/apache/cassandra/io/util/PathUtils.java index 6df5be1..28ce020 100644 --- a/src/java/org/apache/cassandra/io/util/PathUtils.java +++ b/src/java/org/apache/cassandra/io/util/PathUtils.java @@ -56,7 +56,7 @@ import static org.apache.cassandra.utils.Throwables.merge; */ public final class PathUtils { - private static final boolean consistentDirectoryListings = CassandraRelevantProperties.CONSISTENT_DIRECTORY_LISTINGS.getBoolean(); + private static final boolean consistentDirectoryListings = CassandraRelevantProperties.DETERMINISM_CONSISTENT_DIRECTORY_LISTINGS.getBoolean(); private static final Set<StandardOpenOption> READ_OPTIONS = unmodifiableSet(EnumSet.of(READ)); private static final Set<StandardOpenOption> WRITE_OPTIONS = unmodifiableSet(EnumSet.of(WRITE, CREATE)); diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java index 2ec555c..16db492 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicaCollection.java @@ -23,12 +23,14 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import com.google.common.collect.Iterables; +import java.util.AbstractList; import java.util.AbstractMap; import java.util.AbstractSet; import java.util.Arrays; import java.util.Collection; import java.util.Comparator; import java.util.Iterator; +import java.util.List; import java.util.Objects; import java.util.Set; import java.util.function.BiConsumer; @@ -396,6 +398,22 @@ public abstract class AbstractReplicaCollection<C extends AbstractReplicaCollect return snapshot(subList); } + public final <T> List<T> asList(Function<Replica, T> viewTransform) + { + return new AbstractList<T>() + { + public T get(int index) + { + return viewTransform.apply(list.get(index)); + } + + public int size() + { + return list.size; + } + }; + } + /** see {@link ReplicaCollection#count(Predicate)}*/ public int count(Predicate<Replica> predicate) { diff --git a/src/java/org/apache/cassandra/net/RequestCallbacks.java b/src/java/org/apache/cassandra/net/RequestCallbacks.java index 8cec096..6275c15 100644 --- a/src/java/org/apache/cassandra/net/RequestCallbacks.java +++ b/src/java/org/apache/cassandra/net/RequestCallbacks.java @@ -89,7 +89,8 @@ public class RequestCallbacks implements OutboundMessageCallbacks * Remove and return the {@link CallbackInfo} associated with given id and peer, if known. */ @Nullable - CallbackInfo remove(long id, InetAddressAndPort peer) + @VisibleForTesting + public CallbackInfo remove(long id, InetAddressAndPort peer) { return callbacks.remove(key(id, peer)); } @@ -250,13 +251,14 @@ public class RequestCallbacks implements OutboundMessageCallbacks } } - static class CallbackInfo + @VisibleForTesting + public static class CallbackInfo { final long createdAtNanos; final long expiresAtNanos; final InetAddressAndPort peer; - final RequestCallback callback; + public final RequestCallback callback; @Deprecated // for 3.0 compatibility purposes only public final Verb responseVerb; diff --git a/src/java/org/apache/cassandra/schema/CompressionParams.java b/src/java/org/apache/cassandra/schema/CompressionParams.java index c68bdf1d..fca7e4c 100644 --- a/src/java/org/apache/cassandra/schema/CompressionParams.java +++ b/src/java/org/apache/cassandra/schema/CompressionParams.java @@ -34,6 +34,7 @@ import org.apache.commons.lang3.builder.HashCodeBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.ParameterizedClass; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.exceptions.ConfigurationException; @@ -65,11 +66,13 @@ public final class CompressionParams public static final String ENABLED = "enabled"; public static final String MIN_COMPRESS_RATIO = "min_compress_ratio"; - public static final CompressionParams DEFAULT = new CompressionParams(LZ4Compressor.create(Collections.<String, String>emptyMap()), - DEFAULT_CHUNK_LENGTH, - calcMaxCompressedLength(DEFAULT_CHUNK_LENGTH, DEFAULT_MIN_COMPRESS_RATIO), - DEFAULT_MIN_COMPRESS_RATIO, - Collections.emptyMap()); + public static final CompressionParams DEFAULT = !CassandraRelevantProperties.DETERMINISM_SSTABLE_COMPRESSION_DEFAULT.getBoolean() + ? noCompression() + : new CompressionParams(LZ4Compressor.create(Collections.emptyMap()), + DEFAULT_CHUNK_LENGTH, + calcMaxCompressedLength(DEFAULT_CHUNK_LENGTH, DEFAULT_MIN_COMPRESS_RATIO), + DEFAULT_MIN_COMPRESS_RATIO, + Collections.emptyMap()); public static final CompressionParams NOOP = new CompressionParams(NoopCompressor.create(Collections.emptyMap()), // 4 KiB is often the underlying disk block size diff --git a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java index f0f700c..3050886 100644 --- a/src/java/org/apache/cassandra/schema/MigrationCoordinator.java +++ b/src/java/org/apache/cassandra/schema/MigrationCoordinator.java @@ -39,6 +39,7 @@ import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; import org.apache.cassandra.concurrent.FutureTask; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.utils.concurrent.Future; import org.apache.cassandra.utils.concurrent.ImmediateFuture; import org.slf4j.Logger; @@ -78,7 +79,7 @@ public class MigrationCoordinator } - private static final int MIGRATION_DELAY_IN_MS = 60000; + private static final int MIGRATION_DELAY_IN_MS = CassandraRelevantProperties.MIGRATION_DELAY.getInt(); private static final int MAX_OUTSTANDING_VERSION_REQUESTS = 3; public static final MigrationCoordinator instance = new MigrationCoordinator(); diff --git a/src/java/org/apache/cassandra/schema/TableId.java b/src/java/org/apache/cassandra/schema/TableId.java index 695147f..0633105 100644 --- a/src/java/org/apache/cassandra/schema/TableId.java +++ b/src/java/org/apache/cassandra/schema/TableId.java @@ -68,6 +68,11 @@ public class TableId public static TableId forSystemTable(String keyspace, String table) { assert SchemaConstants.isLocalSystemKeyspace(keyspace) || SchemaConstants.isReplicatedSystemKeyspace(keyspace); + return unsafeDeterministic(keyspace, table); + } + + public static TableId unsafeDeterministic(String keyspace, String table) + { return new TableId(UUID.nameUUIDFromBytes(ArrayUtils.addAll(keyspace.getBytes(), table.getBytes()))); } diff --git a/src/java/org/apache/cassandra/schema/TableMetadata.java b/src/java/org/apache/cassandra/schema/TableMetadata.java index a5530ed..1e85d88 100644 --- a/src/java/org/apache/cassandra/schema/TableMetadata.java +++ b/src/java/org/apache/cassandra/schema/TableMetadata.java @@ -739,7 +739,10 @@ public class TableMetadata implements SchemaElement partitioner = DatabaseDescriptor.getPartitioner(); if (id == null) - id = TableId.generate(); + { + if (DatabaseDescriptor.useDeterministicTableID()) id = TableId.unsafeDeterministic(keyspace, name); + else id = TableId.generate(); + } if (Flag.isCQLTable(flags)) return new TableMetadata(this); diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java index 24099c3..ed53a20 100644 --- a/src/java/org/apache/cassandra/service/ClientState.java +++ b/src/java/org/apache/cassandra/service/ClientState.java @@ -250,7 +250,7 @@ public class ClientState * it may be returned multiple times). Note that we still ensure Paxos "ballot" are unique (for different * proposal) by (securely) randomizing the non-timestamp part of the UUID. */ - public long getTimestampForPaxos(long minTimestampToUse) + public static long getTimestampForPaxos(long minTimestampToUse) { while (true) { @@ -266,6 +266,11 @@ public class ClientState } } + public static long getLastTimestampMicros() + { + return lastTimestampMicros.get(); + } + public Optional<String> getDriverName() { return Optional.ofNullable(driverName); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 5c0f4bb..93bf478 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -39,6 +39,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; import java.util.regex.MatchResult; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -61,6 +62,7 @@ import com.google.common.util.concurrent.*; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.concurrent.*; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.ParameterizedClass; import org.apache.cassandra.dht.RangeStreamer.FetchReplica; import org.apache.cassandra.fql.FullQueryLogger; @@ -139,6 +141,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.stream.Collectors.toList; import static java.util.stream.Collectors.toMap; +import static org.apache.cassandra.concurrent.FutureTask.callable; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SCHEMA_DELAY_MS; import static org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK; import static org.apache.cassandra.config.CassandraRelevantProperties.REPLACEMENT_ALLOW_EMPTY; @@ -171,7 +174,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE private static int getRingDelay() { - String newdelay = System.getProperty("cassandra.ring_delay_ms"); + String newdelay = CassandraRelevantProperties.RING_DELAY.getString(); if (newdelay != null) { logger.info("Overriding RING_DELAY to {}ms", newdelay); @@ -1818,6 +1821,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public Future<StreamState> startBootstrap(Collection<Token> tokens) { + return startBootstrap(tokens, replacing); + } + + public Future<StreamState> startBootstrap(Collection<Token> tokens, boolean replacing) + { setMode(Mode.JOINING, "Starting to bootstrap...", true); BootStrapper bootstrapper = new BootStrapper(FBUtilities.getBroadcastAddressAndPort(), tokens, tokenMetadata); bootstrapper.addProgressListener(progressSupport); @@ -4118,6 +4126,11 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public Pair<Integer, Future<?>> repair(String keyspace, Map<String, String> repairSpec, List<ProgressListener> listeners) { RepairOption option = RepairOption.parse(repairSpec, tokenMetadata.partitioner); + return repair(keyspace, option, listeners); + } + + public Pair<Integer, Future<?>> repair(String keyspace, RepairOption option, List<ProgressListener> listeners) + { // if ranges are not specified if (option.getRanges().isEmpty()) { @@ -4209,6 +4222,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE ActiveRepairService.instance.terminateSessions(); } + // TODO: remove this after forward-porting Paxos, before final commit + public Future<?> startRepairPaxosForTopologyChange(String reason) + { + // PLACEHOLDER METHOD for simple porting of Simulator without breaking correctness + return ImmediateFuture.success(null); + } + @Nullable public List<String> getParentRepairStatus(int cmd) { @@ -4593,7 +4613,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE Uninterruptibles.sleepUninterruptibly(delay, MILLISECONDS); } - private void unbootstrap(Runnable onFinish) throws ExecutionException, InterruptedException + public Supplier<Future<StreamState>> prepareUnbootstrapStreaming() { Map<String, EndpointsByReplica> rangesToStream = new HashMap<>(); @@ -4607,11 +4627,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE rangesToStream.put(keyspaceName, rangesMM); } + return () -> streamRanges(rangesToStream); + } + + private void unbootstrap(Runnable onFinish) throws ExecutionException, InterruptedException + { + Supplier<Future<StreamState>> startStreaming = prepareUnbootstrapStreaming(); + setMode(Mode.LEAVING, "replaying batch log and streaming data to other nodes", true); // Start with BatchLog replay, which may create hints but no writes since this is no longer a valid endpoint. Future<?> batchlogReplay = BatchlogManager.instance.startBatchlogReplay(); - Future<StreamState> streamSuccess = streamRanges(rangesToStream); + Future<StreamState> streamSuccess = startStreaming.get(); // Wait for batch log to complete before streaming hints. logger.debug("waiting for batch log processing."); diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index d0ab6b2..9319fba 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -522,6 +522,8 @@ public class ByteBufferUtil return ByteBufferUtil.bytes((InetAddress) obj); else if (obj instanceof String) return ByteBufferUtil.bytes((String) obj); + else if (obj instanceof byte[]) + return ByteBuffer.wrap((byte[]) obj); else if (obj instanceof ByteBuffer) return (ByteBuffer) obj; else diff --git a/src/java/org/apache/cassandra/utils/Clock.java b/src/java/org/apache/cassandra/utils/Clock.java index 629a585..6fd0efd 100644 --- a/src/java/org/apache/cassandra/utils/Clock.java +++ b/src/java/org/apache/cassandra/utils/Clock.java @@ -22,6 +22,8 @@ import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOBAL; + /** * Wrapper around time related functions that are either implemented by using the default JVM calls * or by using a custom implementation for testing purposes. @@ -46,7 +48,7 @@ public interface Clock static { - String classname = System.getProperty("cassandra.clock"); + String classname = CLOCK_GLOBAL.getString(); Clock clock = new Default(); if (classname != null) { diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 3905a77..912d907 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -115,11 +115,17 @@ public class FBUtilities private static volatile String previousReleaseVersionString; + private static int availableProcessors = Integer.getInteger("cassandra.available_processors", DatabaseDescriptor.getAvailableProcessors()); + + public static void setAvailableProcessors(int value) + { + availableProcessors = value; + } + public static int getAvailableProcessors() { - String availableProcessors = System.getProperty("cassandra.available_processors"); - if (!Strings.isNullOrEmpty(availableProcessors)) - return Integer.parseInt(availableProcessors); + if (availableProcessors > 0) + return availableProcessors; else return Runtime.getRuntime().availableProcessors(); } diff --git a/src/java/org/apache/cassandra/utils/Hex.java b/src/java/org/apache/cassandra/utils/Hex.java index 9163067..b8044b8 100644 --- a/src/java/org/apache/cassandra/utils/Hex.java +++ b/src/java/org/apache/cassandra/utils/Hex.java @@ -86,6 +86,23 @@ public class Hex return wrapCharArray(c); } + public static long parseLong(String hex, int start, int end) + { + int len = end - start; + if (len > 16) + throw new IllegalArgumentException(); + + long result = 0; + int shift = 4 * (len - 1); + for (int i = start ; i < end ; ++i) + { + char c = hex.charAt(i); + result |= (long)(c - (c >= 'a' ? 'a' - 10 : '0')) << shift; + shift -= 4; + } + return result; + } + /** * Create a String from a char array with zero-copy (if available), using reflection to access a package-protected constructor of String. * */ diff --git a/src/java/org/apache/cassandra/utils/LazyToString.java b/src/java/org/apache/cassandra/utils/LazyToString.java new file mode 100644 index 0000000..e719445 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/LazyToString.java @@ -0,0 +1,36 @@ +/* + * 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.utils; + +import java.util.function.Supplier; + +public interface LazyToString +{ + public static LazyToString lazy(Supplier<String> castAsLambda) + { + return new LazyToString() + { + @Override + public String toString() + { + return castAsLambda.get(); + } + }; + } +} diff --git a/src/java/org/apache/cassandra/utils/MonotonicClock.java b/src/java/org/apache/cassandra/utils/MonotonicClock.java index ae98466..2ef563c 100644 --- a/src/java/org/apache/cassandra/utils/MonotonicClock.java +++ b/src/java/org/apache/cassandra/utils/MonotonicClock.java @@ -30,6 +30,8 @@ import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.Config; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_APPROX; +import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_PRECISE; import static org.apache.cassandra.utils.Clock.Global.nanoTime; /** @@ -79,7 +81,7 @@ public interface MonotonicClock private static MonotonicClock precise() { - String sclock = System.getProperty("cassandra.monotonic_clock.precise"); + String sclock = CLOCK_MONOTONIC_PRECISE.getString(); if (sclock != null) { @@ -99,7 +101,7 @@ public interface MonotonicClock private static MonotonicClock approx(MonotonicClock precise) { - String sclock = System.getProperty("cassandra.monotonic_clock.approx"); + String sclock = CLOCK_MONOTONIC_APPROX.getString(); if (sclock != null) { try diff --git a/src/java/org/apache/cassandra/utils/NativeLibrary.java b/src/java/org/apache/cassandra/utils/NativeLibrary.java index 01225aa..b34f626 100644 --- a/src/java/org/apache/cassandra/utils/NativeLibrary.java +++ b/src/java/org/apache/cassandra/utils/NativeLibrary.java @@ -23,6 +23,7 @@ import java.lang.reflect.Field; import java.nio.channels.FileChannel; import java.util.concurrent.TimeUnit; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.io.util.File; import org.apache.cassandra.io.util.FileInputStreamPlus; import org.slf4j.Logger; @@ -32,6 +33,7 @@ import com.sun.jna.LastErrorException; import org.apache.cassandra.io.FSWriteError; +import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORE_MISSING_NATIVE_FILE_HINTS; import static org.apache.cassandra.config.CassandraRelevantProperties.OS_ARCH; import static org.apache.cassandra.config.CassandraRelevantProperties.OS_NAME; import static org.apache.cassandra.utils.NativeLibrary.OSType.LINUX; @@ -42,6 +44,7 @@ import static org.apache.cassandra.utils.NativeLibrary.OSType.AIX; public final class NativeLibrary { private static final Logger logger = LoggerFactory.getLogger(NativeLibrary.class); + private static final boolean REQUIRE = !IGNORE_MISSING_NATIVE_FILE_HINTS.getBoolean(); public enum OSType { @@ -160,7 +163,8 @@ public final class NativeLibrary } catch (NoSuchMethodError x) { - logger.warn("Obsolete version of JNA present; unable to read errno. Upgrade to JNA 3.2.7 or later"); + if (REQUIRE) + logger.warn("Obsolete version of JNA present; unable to read errno. Upgrade to JNA 3.2.7 or later"); return 0; } } @@ -292,7 +296,8 @@ public final class NativeLibrary if (!(e instanceof LastErrorException)) throw e; - logger.warn("fcntl({}, {}, {}) failed, errno ({}).", fd, command, flags, errno(e)); + if (REQUIRE) + logger.warn("fcntl({}, {}, {}) failed, errno ({}).", fd, command, flags, errno(e)); } return result; @@ -315,7 +320,8 @@ public final class NativeLibrary if (!(e instanceof LastErrorException)) throw e; - logger.warn("open({}, O_RDONLY) failed, errno ({}).", path, errno(e)); + if (REQUIRE) + logger.warn("open({}, O_RDONLY) failed, errno ({}).", path, errno(e)); } return fd; @@ -339,9 +345,12 @@ public final class NativeLibrary if (!(e instanceof LastErrorException)) throw e; - String errMsg = String.format("fsync(%s) failed, errno (%s) %s", fd, errno(e), e.getMessage()); - logger.warn(errMsg); - throw new FSWriteError(e, errMsg); + if (REQUIRE) + { + String errMsg = String.format("fsync(%s) failed, errno (%s) %s", fd, errno(e), e.getMessage()); + logger.warn(errMsg); + throw new FSWriteError(e, errMsg); + } } } @@ -363,9 +372,12 @@ public final class NativeLibrary if (!(e instanceof LastErrorException)) throw e; - String errMsg = String.format("close(%d) failed, errno (%d).", fd, errno(e)); - logger.warn(errMsg); - throw new FSWriteError(e, errMsg); + if (REQUIRE) + { + String errMsg = String.format("close(%d) failed, errno (%d).", fd, errno(e)); + logger.warn(errMsg); + throw new FSWriteError(e, errMsg); + } } } @@ -377,7 +389,8 @@ public final class NativeLibrary } catch (IllegalArgumentException|IllegalAccessException e) { - logger.warn("Unable to read fd field from FileChannel"); + if (REQUIRE) + logger.warn("Unable to read fd field from FileChannel", e); } return -1; } @@ -395,8 +408,11 @@ public final class NativeLibrary } catch (Exception e) { - JVMStabilityInspector.inspectThrowable(e); - logger.warn("Unable to read fd field from FileDescriptor"); + if (REQUIRE) + { + JVMStabilityInspector.inspectThrowable(e); + logger.warn("Unable to read fd field from FileDescriptor", e); + } } return -1; @@ -417,7 +433,8 @@ public final class NativeLibrary } catch (Exception e) { - logger.info("Failed to get PID from JNA", e); + if (REQUIRE) + logger.info("Failed to get PID from JNA", e); } return -1; diff --git a/src/java/org/apache/cassandra/utils/SigarLibrary.java b/src/java/org/apache/cassandra/utils/SigarLibrary.java index 246a9c8..81e44d2 100644 --- a/src/java/org/apache/cassandra/utils/SigarLibrary.java +++ b/src/java/org/apache/cassandra/utils/SigarLibrary.java @@ -21,6 +21,8 @@ import org.hyperic.sigar.*; import org.slf4j.LoggerFactory; import org.slf4j.Logger; +import org.apache.cassandra.config.CassandraRelevantProperties; + public class SigarLibrary { private Logger logger = LoggerFactory.getLogger(SigarLibrary.class); @@ -169,7 +171,7 @@ public class SigarLibrary boolean goodAddressSpace = hasAcceptableAddressSpace(); boolean goodFileLimits = hasAcceptableFileLimits(); boolean goodProcNumber = hasAcceptableProcNumber(); - if (swapEnabled || !goodAddressSpace || !goodFileLimits || !goodProcNumber) + if (swapEnabled || !goodAddressSpace || !goodFileLimits || !goodProcNumber || CassandraRelevantProperties.TEST_IGNORE_SIGAR.getBoolean()) { logger.warn("Cassandra server running in degraded mode. Is swap disabled? : {}, Address space adequate? : {}, " + " nofile limit adequate? : {}, nproc limit adequate? : {} ", !swapEnabled, goodAddressSpace, diff --git a/src/java/org/apache/cassandra/utils/UUIDGen.java b/src/java/org/apache/cassandra/utils/UUIDGen.java index 1e727d4..cab7256 100644 --- a/src/java/org/apache/cassandra/utils/UUIDGen.java +++ b/src/java/org/apache/cassandra/utils/UUIDGen.java @@ -36,6 +36,7 @@ import java.util.function.Function; import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; import com.google.common.hash.Hasher; import com.google.common.hash.Hashing; import com.google.common.primitives.Ints; @@ -104,6 +105,16 @@ public class UUIDGen } /** + * Creates a type 1 UUID (time-based UUID) with the timestamp of @param when, in milliseconds. + * + * @return a UUID instance + */ + public static UUID getTimeUUIDWithClockSeqAndNode(long when, long clockSeqAndNode) + { + return new UUID(createTime(fromUnixTimestamp(when)), clockSeqAndNode); + } + + /** * Returns a version 1 UUID using the provided timestamp and the local clock and sequence. * <p> * Note that this method is generally only safe to use if you can guarantee that the provided @@ -130,16 +141,44 @@ public class UUIDGen * through randomization. * * @param whenInMicros a unix time in microseconds. + * @param flag a value between 0 and 9 (inclusive) to put in least significant (and unused) part of the timestamp * @return a new UUID {@code id} such that {@code microsTimestamp(id) == whenInMicros}. The UUID returned * by different calls will be unique even if {@code whenInMicros} is not. */ - public static UUID getRandomTimeUUIDFromMicros(long whenInMicros) + public static UUID getRandomTimeUUIDFromMicros(long whenInMicros, int flag) { + Preconditions.checkArgument(0 <= flag && flag < 10); long whenInMillis = whenInMicros / 1000; - long nanos = (whenInMicros - (whenInMillis * 1000)) * 10; + long nanos = (whenInMicros - (whenInMillis * 1000)) * 10 + flag; return new UUID(createTime(fromUnixTimestamp(whenInMillis, nanos)), secureRandom.nextLong()); } + @Deprecated + public static UUID getRandomTimeUUIDFromMicros(long whenInMicros) + { + return getRandomTimeUUIDFromMicros(whenInMicros, 0); + } + + /** + * Similar to {@link #getTimeUUIDFromMicros}, but randomize (using SecureRandom) the clock and sequence. + * <p> + * If you can guarantee that the {@code whenInMicros} argument is unique (for this JVM instance) for + * every call, then you should prefer {@link #getTimeUUIDFromMicros} which is faster. If you can't + * guarantee this however, this method will ensure the returned UUID are still unique (accross calls) + * through randomization. + * + * @param whenInMicros a unix time in microseconds. + * @return a new UUID {@code id} such that {@code microsTimestamp(id) == whenInMicros}. The UUID returned + * by different calls will be unique even if {@code whenInMicros} is not. + */ + public static UUID getRandomTimeUUIDFromMicrosAndRandom(long whenInMicros, long randomPart, int flag) + { + Preconditions.checkArgument(0 <= flag && flag < 10); + long whenInMillis = whenInMicros / 1000; + long nanos = (whenInMicros - (whenInMillis * 1000)) * 10 + flag; + return new UUID(createTime(fromUnixTimestamp(whenInMillis, nanos)), randomPart); + } + public static UUID getTimeUUID(long when, long nanos) { return new UUID(createTime(fromUnixTimestamp(when, nanos)), clockSeqAndNode); @@ -280,7 +319,11 @@ public class UUIDGen private static byte[] createTimeUUIDBytes(long msb) { - long lsb = clockSeqAndNode; + return createTimeUUIDBytes(msb, clockSeqAndNode); + } + + public static byte[] createTimeUUIDBytes(long msb, long lsb) + { byte[] uuidBytes = new byte[16]; for (int i = 0; i < 8; i++) @@ -308,6 +351,15 @@ public class UUIDGen private static long makeClockSeqAndNode() { + if (Boolean.getBoolean("cassandra.unsafe.deterministicuuidnode")) + return FBUtilities.getBroadcastAddressAndPort().addressBytes[3]; + + Long specified = Long.getLong("cassandra.unsafe.timeuuidnode"); + if (specified != null) + return specified + ^ FBUtilities.getBroadcastAddressAndPort().addressBytes[3] + ^ (FBUtilities.getBroadcastAddressAndPort().addressBytes[2] << 8); + long clock = new SecureRandom().nextLong(); long lsb = 0; @@ -352,7 +404,7 @@ public class UUIDGen return createTime(nanosSince); } - private static long createTime(long nanosSince) + public static long createTime(long nanosSince) { long msb = 0L; msb |= (0x00000000ffffffffL & nanosSince) << 32; diff --git a/src/java/org/apache/cassandra/utils/concurrent/Threads.java b/src/java/org/apache/cassandra/utils/concurrent/Threads.java new file mode 100644 index 0000000..439a77f --- /dev/null +++ b/src/java/org/apache/cassandra/utils/concurrent/Threads.java @@ -0,0 +1,135 @@ +/* + * 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.utils.concurrent; + +import java.util.Collections; +import java.util.Set; +import java.util.function.BiConsumer; +import java.util.function.BinaryOperator; +import java.util.function.Function; +import java.util.function.Supplier; +import java.util.stream.Collector; +import java.util.stream.Stream; + +public class Threads +{ + public static class StackTraceCombiner implements Collector<StackTraceElement, StringBuilder, String>, Supplier<StringBuilder>, Function<StringBuilder, String> + { + final boolean printBriefPackages; + final String prefix; + final String delimiter; + final String suffix; + + public StackTraceCombiner(boolean printBriefPackages, String prefix, String delimiter, String suffix) + { + this.printBriefPackages = printBriefPackages; + this.prefix = prefix; + this.delimiter = delimiter; + this.suffix = suffix; + } + + public Supplier<StringBuilder> supplier() + { + + return this; + } + + public BiConsumer<StringBuilder, StackTraceElement> accumulator() + { + return (sb, ste) -> + { + if (sb.length() > prefix.length()) + sb.append(delimiter); + + String className = ste.getClassName(); + + if (printBriefPackages) + { + int afterPrevDot = 0; + while (true) + { + int dot = className.indexOf('.', afterPrevDot); + if (dot < 0) + break; + + sb.append(className.charAt(afterPrevDot)); + sb.append('.'); + afterPrevDot = dot + 1; + } + sb.append(className, afterPrevDot, className.length()); + } + else + { + sb.append(className); + } + sb.append('.'); + sb.append(ste.getMethodName()); + sb.append(':'); + sb.append(ste.getLineNumber()); + }; + } + + public BinaryOperator<StringBuilder> combiner() + { + return (sb1, sb2) -> sb1.append("; ").append(sb2); + } + + public Function<StringBuilder, String> finisher() + { + return this; + } + + public Set<Characteristics> characteristics() + { + return Collections.emptySet(); + } + + public StringBuilder get() + { + return new StringBuilder(prefix); + } + + public String apply(StringBuilder finish) + { + finish.append(suffix); + return finish.toString(); + } + } + + public static String prettyPrintStackTrace(Thread thread, boolean printBriefPackages, String delimiter) + { + return prettyPrint(thread.getStackTrace(), printBriefPackages, delimiter); + } + + public static String prettyPrintStackTrace(Thread thread, boolean printBriefPackages, String prefix, String delimiter, String suffix) + { + return prettyPrint(thread.getStackTrace(), printBriefPackages, prefix, delimiter, suffix); + } + + public static String prettyPrint(StackTraceElement[] st, boolean printBriefPackages, String delimiter) + { + return prettyPrint(st, printBriefPackages, "", delimiter, ""); + } + + public static String prettyPrint(StackTraceElement[] st, boolean printBriefPackages, String prefix, String delimiter, String suffix) + { + return Stream.of(st).collect(new StackTraceCombiner(printBriefPackages, prefix, delimiter, suffix)); + } + +} diff --git a/src/java/org/apache/cassandra/utils/memory/HeapPool.java b/src/java/org/apache/cassandra/utils/memory/HeapPool.java index 6371bda..0596aeb 100644 --- a/src/java/org/apache/cassandra/utils/memory/HeapPool.java +++ b/src/java/org/apache/cassandra/utils/memory/HeapPool.java @@ -20,23 +20,25 @@ package org.apache.cassandra.utils.memory; import java.nio.ByteBuffer; +import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.utils.concurrent.OpOrder; public class HeapPool extends MemtablePool { + private static final EnsureOnHeap ENSURE_NOOP = new EnsureOnHeap.NoOp(); + public HeapPool(long maxOnHeapMemory, float cleanupThreshold, MemtableCleaner cleaner) { super(maxOnHeapMemory, 0, cleanupThreshold, cleaner); } - public MemtableAllocator newAllocator() + public MemtableAllocator newAllocator(ColumnFamilyStore table) { return new Allocator(this); } private static class Allocator extends MemtableBufferAllocator { - private static final EnsureOnHeap ENSURE_NOOP = new EnsureOnHeap.NoOp(); Allocator(HeapPool pool) { super(pool.onHeap.newAllocator(), pool.offHeap.newAllocator()); @@ -53,4 +55,74 @@ public class HeapPool extends MemtablePool return ENSURE_NOOP; } } + + public static class Logged extends MemtablePool + { + public interface Listener + { + public void accept(long size, String table); + } + + private static Listener onAllocated = (i, table) -> {}; + + class LoggedSubPool extends SubPool + { + public LoggedSubPool(long limit, float cleanThreshold) + { + super(limit, cleanThreshold); + } + + public MemtableAllocator.SubAllocator newAllocator(String table) + { + return new MemtableAllocator.SubAllocator(this) + { + public void allocate(long size, OpOrder.Group opGroup) + { + onAllocated.accept(size, table); + super.allocate(size, opGroup); + } + }; + } + } + + public Logged(long maxOnHeapMemory, float cleanupThreshold, MemtableCleaner cleaner) + { + super(maxOnHeapMemory, 0, cleanupThreshold, cleaner); + } + + public MemtableAllocator newAllocator(ColumnFamilyStore cfs) + { + return new Allocator(this, cfs == null ? "" : cfs.keyspace.getName() + '.' + cfs.name); + } + + private static class Allocator extends MemtableBufferAllocator + { + Allocator(Logged pool, String table) + { + super(((LoggedSubPool) pool.onHeap).newAllocator(table), ((LoggedSubPool) pool.offHeap).newAllocator(table)); + } + + public ByteBuffer allocate(int size, OpOrder.Group opGroup) + { + super.onHeap().allocate(size, opGroup); + return ByteBuffer.allocate(size); + } + + @Override + public EnsureOnHeap ensureOnHeap() + { + return ENSURE_NOOP; + } + } + + SubPool getSubPool(long limit, float cleanThreshold) + { + return new LoggedSubPool(limit, cleanThreshold); + } + + public static void setListener(Listener listener) + { + onAllocated = listener; + } + } } diff --git a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java index 2883830..d7076f5 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtableAllocator.java @@ -103,7 +103,7 @@ public abstract class MemtableAllocator } /** Mark the BB as unused, permitting it to be reclaimed */ - public static final class SubAllocator + public static class SubAllocator { // the tracker we are owning memory from private final MemtablePool.SubPool parent; diff --git a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java index 707f164..b43973d 100644 --- a/src/java/org/apache/cassandra/utils/memory/MemtablePool.java +++ b/src/java/org/apache/cassandra/utils/memory/MemtablePool.java @@ -27,6 +27,7 @@ import com.google.common.base.Preconditions; import com.codahale.metrics.Gauge; import com.codahale.metrics.Timer; +import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.metrics.DefaultNameFactory; import org.apache.cassandra.utils.concurrent.WaitQueue; @@ -81,7 +82,7 @@ public abstract class MemtablePool ExecutorUtils.shutdownNowAndWait(timeout, unit, cleaner); } - public abstract MemtableAllocator newAllocator(); + public abstract MemtableAllocator newAllocator(ColumnFamilyStore table); public boolean needsCleaning() { diff --git a/src/java/org/apache/cassandra/utils/memory/NativePool.java b/src/java/org/apache/cassandra/utils/memory/NativePool.java index e88b4d7..03c42e1 100644 --- a/src/java/org/apache/cassandra/utils/memory/NativePool.java +++ b/src/java/org/apache/cassandra/utils/memory/NativePool.java @@ -18,6 +18,8 @@ */ package org.apache.cassandra.utils.memory; +import org.apache.cassandra.db.ColumnFamilyStore; + public class NativePool extends MemtablePool { public NativePool(long maxOnHeapMemory, long maxOffHeapMemory, float cleanThreshold, MemtableCleaner cleaner) @@ -26,7 +28,7 @@ public class NativePool extends MemtablePool } @Override - public NativeAllocator newAllocator() + public NativeAllocator newAllocator(ColumnFamilyStore table) { return new NativeAllocator(this); } diff --git a/src/java/org/apache/cassandra/utils/memory/SlabPool.java b/src/java/org/apache/cassandra/utils/memory/SlabPool.java index 416d1dd..adeb14f 100644 --- a/src/java/org/apache/cassandra/utils/memory/SlabPool.java +++ b/src/java/org/apache/cassandra/utils/memory/SlabPool.java @@ -18,6 +18,8 @@ */ package org.apache.cassandra.utils.memory; +import org.apache.cassandra.db.ColumnFamilyStore; + public class SlabPool extends MemtablePool { final boolean allocateOnHeap; @@ -28,7 +30,7 @@ public class SlabPool extends MemtablePool this.allocateOnHeap = maxOffHeapMemory == 0; } - public MemtableAllocator newAllocator() + public MemtableAllocator newAllocator(ColumnFamilyStore table) { return new SlabAllocator(onHeap.newAllocator(), offHeap.newAllocator(), allocateOnHeap); } diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index 3d4de68..3478fb3 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -132,6 +132,7 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.exceptions.TransportException", "org.apache.cassandra.fql.FullQueryLogger", "org.apache.cassandra.fql.FullQueryLoggerOptions", + "org.apache.cassandra.gms.IFailureDetector", "org.apache.cassandra.locator.IEndpointSnitch", "org.apache.cassandra.io.FSWriteError", "org.apache.cassandra.io.FSError", diff --git a/test/unit/org/apache/cassandra/db/CellSpecTest.java b/test/unit/org/apache/cassandra/db/CellSpecTest.java index eac724a..24a3b12 100644 --- a/test/unit/org/apache/cassandra/db/CellSpecTest.java +++ b/test/unit/org/apache/cassandra/db/CellSpecTest.java @@ -138,7 +138,7 @@ public class CellSpecTest byte[] rawBytes = { 0, 1, 2, 3, 4, 5, 6 }; ByteBuffer bbBytes = ByteBuffer.wrap(rawBytes); NativePool pool = new NativePool(1024, 1024, 1, () -> ImmediateFuture.success(true)); - NativeAllocator allocator = pool.newAllocator(); + NativeAllocator allocator = pool.newAllocator(null); OpOrder order = new OpOrder(); List<Cell<?>> tests = new ArrayList<>(); diff --git a/test/unit/org/apache/cassandra/db/ClusteringHeapSizeTest.java b/test/unit/org/apache/cassandra/db/ClusteringHeapSizeTest.java index dbf9166..e97f067 100644 --- a/test/unit/org/apache/cassandra/db/ClusteringHeapSizeTest.java +++ b/test/unit/org/apache/cassandra/db/ClusteringHeapSizeTest.java @@ -72,7 +72,7 @@ public class ClusteringHeapSizeTest return Arrays.asList(new Object[][] { { array }, { buffer }, - { new NativeClustering(pool.newAllocator(), order.getCurrent(), array)} + { new NativeClustering(pool.newAllocator(null), order.getCurrent(), array)} }); } } \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/db/NativeCellTest.java b/test/unit/org/apache/cassandra/db/NativeCellTest.java index 713f26b..fad9e2d 100644 --- a/test/unit/org/apache/cassandra/db/NativeCellTest.java +++ b/test/unit/org/apache/cassandra/db/NativeCellTest.java @@ -48,7 +48,7 @@ public class NativeCellTest private static final NativeAllocator nativeAllocator = new NativePool(Integer.MAX_VALUE, Integer.MAX_VALUE, 1f, - () -> ImmediateFuture.success(true)).newAllocator(); + () -> ImmediateFuture.success(true)).newAllocator(null); @SuppressWarnings("resource") private static final OpOrder.Group group = new OpOrder().start(); private static Random rand; diff --git a/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java b/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java index 86b81e5..3e28040 100644 --- a/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java +++ b/test/unit/org/apache/cassandra/db/filter/ColumnFilterTest.java @@ -94,6 +94,7 @@ public class ColumnFilterTest { DatabaseDescriptor.setSeedProvider(Arrays::asList); DatabaseDescriptor.setEndpointSnitch(new SimpleSnitch()); + DatabaseDescriptor.setDefaultFailureDetector(); Gossiper.instance.start(0); } diff --git a/test/unit/org/apache/cassandra/gms/ArrivalWindowTest.java b/test/unit/org/apache/cassandra/gms/ArrivalWindowTest.java index ea59300..3a07ea3 100644 --- a/test/unit/org/apache/cassandra/gms/ArrivalWindowTest.java +++ b/test/unit/org/apache/cassandra/gms/ArrivalWindowTest.java @@ -23,13 +23,21 @@ package org.apache.cassandra.gms; import static org.junit.Assert.*; +import org.junit.BeforeClass; import org.junit.Test; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.utils.FBUtilities; public class ArrivalWindowTest { + @BeforeClass + public static void beforeClass() + { + DatabaseDescriptor.setDefaultFailureDetector(); + } + @Test public void testWithNanoTime() { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
