From 5e1820487de0308e22bd8edb756a4e76dd6afdcb Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 9 Feb 2011 21:22:01 +0000 Subject: [PATCH] add flush_largest_memtables_at and reduce_cache_sizes_at options patch by jbellis; reviewed by brandonwilliams for CASSANDRA-2142 git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1069119 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 11 +++-- NEWS.txt | 22 ++++++++-- conf/cassandra.yaml | 25 +++++++++++ .../org/apache/cassandra/config/Config.java | 5 ++- .../cassandra/config/DatabaseDescriptor.java | 16 ++++++- .../cassandra/db/ColumnFamilyStore.java | 24 +++++++++- .../service/AbstractCassandraDaemon.java | 9 ++++ .../apache/cassandra/service/GCInspector.java | 27 ++++++++++-- .../cassandra/service/StorageService.java | 44 +++++++++++++++---- 9 files changed, 159 insertions(+), 24 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 28bf806095..9739983bbd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,9 +1,3 @@ -0.7.2 - * cache writing moved to CompactionManager to reduce i/o contention and - updated to use non-cache-polluting writes (CASSANDRA-2053) - * page through large rows when exporting to JSON (CASSANDRA-2041) - - 0.7.1 * buffer network stack to avoid inefficient small TCP messages while avoiding the nagle/delayed ack problem (CASSANDRA-1896) @@ -56,6 +50,11 @@ * avoid blocking gossip while deleting handoff hints (CASSANDRA-2073) * ignore messages from newer versions, keep track of nodes in gossip regardless of version (CASSANDRA-1970) + * cache writing moved to CompactionManager to reduce i/o contention and + updated to use non-cache-polluting writes (CASSANDRA-2053) + * page through large rows when exporting to JSON (CASSANDRA-2041) + * add flush_largest_memtables_at and reduce_cache_sizes_at options + (CASSANDRA-2142) 0.7.0-final diff --git a/NEWS.txt b/NEWS.txt index df58fde9cc..798adcf986 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -1,17 +1,33 @@ 0.7.1 ===== -Uprading --------- +Upgrading +--------- - 0.7.1 is completely backwards compatible with 0.7.0. Just restart each node with the new version, one at a time. (The cluster does not all need to be upgraded simultaneously.) Features -------- - - Cassandra can perform writes efficiently across datacenters by + - added flush_largest_memtables_at and reduce_cache_sizes_at options + to cassandra.yaml as an escape valve for memory pressure + - added option to specify -Dcassandra.join_ring=false on startup + to allow "warm spare" nodes or performing JMX maintenance before + joining the ring + +Performance +----------- + - Disk writes and sequential scans avoid polluting page cache + (requires JNA to be enabled) + - Cassandra performs writes efficiently across datacenters by sending a single copy of the mutation and having the recipient forward that to other replicas in its datacenter. + - Improved network buffering + - Reduced lock contention on memtable flush + - Optimized supercolumn deserialization + - Zero-copy reads from mmapped sstable files + - Explicitly set higher JVM new generation size + - Reduced i/o contention during saving of caches 0.7.0 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 5f4384341b..90193cf40b 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -90,6 +90,31 @@ commitlog_sync: periodic # milliseconds. commitlog_sync_period_in_ms: 10000 +# emergency pressure valve: each time heap usage after a full (CMS) +# garbage collection is above this fraction of the max, Cassandra will +# flush the largest memtables. +# +# Set to 1.0 to disable. Setting this lower than +# CMSInitiatingOccupancyFraction is not likely to be useful. +# +# RELYING ON THIS AS YOUR PRIMARY TUNING MECHANISM WILL WORK POORLY: +# it is most effective under light to moderate load, or read-heavy +# workloads; under truly massive write load, it will often be too +# little, too late. +flush_largest_memtables_at: 0.75 + +# emergency pressure valve #2: the first time heap usage after a full +# (CMS) garbage collection is above this fraction of the max, +# Cassandra will reduce cache maximum _capacity_ to the given fraction +# of the current _size_. Should usually be set substantially above +# flush_largest_memtables_at, since that will have less long-term +# impact on the system. +# +# Set to 1.0 to disable. Setting this lower than +# CMSInitiatingOccupancyFraction is not likely to be useful. +reduce_cache_sizes_at: 0.85 +reduce_cache_capacity_to: 0.6 + # Addresses of hosts that are deemed contact points. # Cassandra nodes use this list of hosts to find each other and learn # the topology of the ring. You must change this if you are running diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index fd4bca0dbe..25e719bd8e 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -103,7 +103,10 @@ public class Config public Integer index_interval = 128; public List keyspaces; - + public Double flush_largest_memtables_at = 1.0; + public Double reduce_cache_sizes_at = 1.0; + public double reduce_cache_capacity_to = 0.6; + public static enum CommitLogSync { periodic, batch diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index be073cda67..75997d625f 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -20,7 +20,6 @@ package org.apache.cassandra.config; import java.io.*; import java.net.InetAddress; -import java.net.MalformedURLException; import java.net.URL; import java.net.UnknownHostException; import java.nio.ByteBuffer; @@ -1155,4 +1154,19 @@ public class DatabaseDescriptor { conf.dynamic_snitch_badness_threshold = dynamicBadnessThreshold; } + + public static double getFlushLargestMemtablesAt() + { + return conf.flush_largest_memtables_at; + } + + public static double getReduceCacheSizesAt() + { + return conf.reduce_cache_sizes_at; + } + + public static double getReduceCacheCapacityTo() + { + return conf.reduce_cache_capacity_to; + } } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index d9abe67c35..6501a8a4b8 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1878,7 +1878,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public String toString() { return "ColumnFamilyStore(" + - "table='" + table + '\'' + + "table='" + table.name + '\'' + ", columnFamily='" + columnFamily + '\'' + ')'; } @@ -2019,4 +2019,26 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean { return partitioner instanceof LocalPartitioner; } + + /** + * sets each cache's maximum capacity to 75% of its current size + */ + public void reduceCacheSizes() + { + if (ssTables.getRowCache().getCapacity() > 0) + { + int newCapacity = (int) (DatabaseDescriptor.getReduceCacheCapacityTo() * ssTables.getRowCache().getSize()); + logger.warn(String.format("Reducing %s row cache capacity from %d to %s to reduce memory pressure", + columnFamily, ssTables.getRowCache().getCapacity(), newCapacity)); + ssTables.getRowCache().setCapacity(newCapacity); + } + + if (ssTables.getKeyCache().getCapacity() > 0) + { + int newCapacity = (int) (DatabaseDescriptor.getReduceCacheCapacityTo() * ssTables.getKeyCache().getSize()); + logger.warn(String.format("Reducing %s key cache capacity from %d to %s to reduce memory pressure", + columnFamily, ssTables.getKeyCache().getCapacity(), newCapacity)); + ssTables.getKeyCache().setCapacity(newCapacity); + } + } } diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index 8375b34ee9..1d4b956795 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -162,6 +162,15 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon Table.open(table); } + try + { + GCInspector.instance.start(); + } + catch (Throwable t) + { + logger.warn("Unable to start GCInspector (currently only supported on the Sun JVM)"); + } + // replay the log if necessary and check for compaction candidates CommitLog.recover(); CompactionManager.instance.checkAllColumnFamilies(); diff --git a/src/java/org/apache/cassandra/service/GCInspector.java b/src/java/org/apache/cassandra/service/GCInspector.java index 6a3747be1c..3ed2dfe36d 100644 --- a/src/java/org/apache/cassandra/service/GCInspector.java +++ b/src/java/org/apache/cassandra/service/GCInspector.java @@ -32,6 +32,7 @@ import javax.management.ObjectName; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.utils.StatusLogger; public class GCInspector @@ -46,6 +47,7 @@ public class GCInspector private HashMap gctimes = new HashMap(); List beans = new ArrayList(); // these are instances of com.sun.management.GarbageCollectorMXBean + private volatile boolean cacheSizesReduced; public GCInspector() { @@ -87,13 +89,13 @@ public class GCInspector { public void run() { - logIntervalGCStats(); + logGCResults(); } }; StorageService.scheduledTasks.scheduleWithFixedDelay(t, INTERVAL_IN_MS, INTERVAL_IN_MS, TimeUnit.MILLISECONDS); } - private void logIntervalGCStats() + private void logGCResults() { for (Object gc : beans) { @@ -121,7 +123,7 @@ public class GCInspector } String st = String.format("GC for %s: %s ms, %s reclaimed leaving %s used; max is %s", - gcw.getName(), gcw.getDuration(), previousMemoryUsed - memoryUsed, memoryUsed, memoryMax); + gcw.getName(), gcw.getDuration(), previousMemoryUsed - memoryUsed, memoryUsed, memoryMax); if (gcw.getDuration() > MIN_DURATION) logger.info(st); else if (logger.isDebugEnabled()) @@ -129,6 +131,25 @@ public class GCInspector if (gcw.getDuration() > MIN_DURATION_TPSTATS) StatusLogger.log(); + + // if we just finished a full collection and we're still using a lot of memory, try to reduce the pressure + if (gcw.getName().equals("ConcurrentMarkSweep")) + { + double usage = (double) memoryUsed / memoryMax; + + if (memoryUsed > DatabaseDescriptor.getReduceCacheSizesAt() * memoryMax && !cacheSizesReduced) + { + cacheSizesReduced = true; + logger.warn("Heap is " + usage + " full. You may need to reduce memtable and/or cache sizes. Cassandra is now reducing cache sizes to free up memory. Adjust reduce_cache_sizes_at threshold in cassandra.yaml if you don't want Cassandra to do this automatically"); + StorageService.instance.reduceCacheSizes(); + } + + if (memoryUsed > DatabaseDescriptor.getFlushLargestMemtablesAt() * memoryMax) + { + logger.warn("Heap is " + usage + " full. You may need to reduce memtable and/or cache sizes. Cassandra will now flush up to the two largest memtables to free up memory. Adjust flush_largest_memtables_at threshold in cassandra.yaml if you don't want Cassandra to do this automatically"); + StorageService.instance.flushLargestMemtables(); + } + } } } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 0f827125d8..070ebedb0b 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -363,15 +363,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe initialized = true; isClientMode = false; - try - { - GCInspector.instance.start(); - } - catch (Throwable t) - { - logger_.warn("Unable to start GCInspector (currently only supported on the Sun JVM)"); - } - if (Boolean.parseBoolean(System.getProperty("cassandra.load_ring_state", "true"))) { logger_.info("Loading persisted ring state"); @@ -2176,4 +2167,39 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe { return efficientCrossDCWrites; } + + /** + * Flushes the two largest memtables by ops and by throughput + */ + public void flushLargestMemtables() + { + ColumnFamilyStore largestByOps = null; + ColumnFamilyStore largestByThroughput = null; + for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) + { + if (largestByOps == null || cfs.getMemtableColumnsCount() > largestByOps.getMemtableColumnsCount()) + largestByOps = cfs; + if (largestByThroughput == null || cfs.getMemtableThroughputInMB() > largestByThroughput.getMemtableThroughputInMB()) + largestByThroughput = cfs; + } + if (largestByOps == null) + { + logger_.error("Unable to reduce heap usage since there are no column families defined"); + return; + } + + logger_.warn("Flushing " + largestByOps + " to relieve memory pressure"); + largestByOps.forceFlush(); + if (largestByThroughput != largestByOps) + { + logger_.warn("Flushing " + largestByThroughput + " to relieve memory pressure"); + largestByThroughput.forceFlush(); + } + } + + public void reduceCacheSizes() + { + for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) + cfs.reduceCacheSizes(); + } }