From 2bb862c320987bcc696765f4f5d8e3029db44371 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Tue, 10 Jan 2012 13:59:50 -0600 Subject: [PATCH 1/4] Add help for INSERT to cqlsh. Patch by Paul Cannon, reviewed by brandonwilliams for CASSANDRA-3718. --- bin/cqlsh | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/bin/cqlsh b/bin/cqlsh index 72906b4ffb..0f02a0c836 100755 --- a/bin/cqlsh +++ b/bin/cqlsh @@ -1021,6 +1021,33 @@ class Shell(cmd.Cmd): Cassandra documentation. """ + def help_insert(self): + print """ + INSERT INTO + ( , [, [, ...]] ) + VALUES ( , [, [, ...]] ) + [USING CONSISTENCY + [AND TIMESTAMP ] + [AND TTL [USING CONSISTENCY From 10a8f67783d9987063424212129be29690628bca Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Wed, 11 Jan 2012 08:52:42 +0100 Subject: [PATCH 2/4] Avoid < 0 value for pending tasks in leveled compaction patch by jbellis; reviewed by slebresne for CASSANDRA-3693 --- CHANGES.txt | 1 + .../db/compaction/LeveledManifest.java | 20 ++++++++++++------- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 09bb85d843..b31f3a007f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -19,6 +19,7 @@ * Don't ignore IOException during compaction (CASSANDRA-3655) * Fix assertion error for CF with gc_grace=0 (CASSANDRA-3579) * Shutdown ParallelCompaction reducer executor after use (CASSANDRA-3711) + * Avoid < 0 value for pending tasks in leveled compaction (CASSANDRA-3693) Merged from 0.8: * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656) * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626) diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java index f61a26aefa..40a0a17b6e 100644 --- a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java +++ b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java @@ -27,6 +27,7 @@ import java.io.IOException; import java.util.*; import com.google.common.collect.Iterables; +import com.google.common.primitives.Ints; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -202,11 +203,14 @@ public class LeveledManifest return builder.toString(); } - private double maxBytesForLevel (int level) + private long maxBytesForLevel(int level) { - return level == 0 - ? 4 * maxSSTableSizeInMB * 1024 * 1024 - : Math.pow(10, level) * maxSSTableSizeInMB * 1024 * 1024; + if (level == 0) + return 4 * maxSSTableSizeInMB * 1024 * 1024; + double bytes = Math.pow(10, level) * maxSSTableSizeInMB * 1024 * 1024; + if (bytes > Long.MAX_VALUE) + throw new RuntimeException("At most " + Long.MAX_VALUE + " bytes may be in a compaction level; your maxSSTableSize must be absurdly high to compute " + bytes); + return (long) bytes; } public synchronized Collection getCompactionCandidates() @@ -424,12 +428,14 @@ public class LeveledManifest public int getEstimatedTasks() { - int n = 0; + long tasks = 0; for (int i = generations.length - 1; i >= 0; i--) { List sstables = generations[i]; - n += Math.max(0L, SSTableReader.getTotalBytes(sstables) - maxBytesForLevel(i)) / (maxSSTableSizeInMB * 1024 * 1024); + long n = Math.max(0L, SSTableReader.getTotalBytes(sstables) - maxBytesForLevel(i)) / (maxSSTableSizeInMB * 1024 * 1024); + logger.debug("Estimating " + n + " compaction tasks in level " + i); + tasks += n; } - return n; + return Ints.checkedCast(tasks); } } From d10da15526cef59ee4837930eb1a2c185bab2e7e Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Wed, 11 Jan 2012 09:54:59 +0100 Subject: [PATCH 3/4] Fix changelog/news and update version for 1.0.7 release --- CHANGES.txt | 6 ++++++ NEWS.txt | 15 +++++++++++++++ build.xml | 2 +- debian/changelog | 6 ++++++ 4 files changed, 28 insertions(+), 1 deletion(-) diff --git a/CHANGES.txt b/CHANGES.txt index b31f3a007f..c0f2c6651a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -20,6 +20,12 @@ * Fix assertion error for CF with gc_grace=0 (CASSANDRA-3579) * Shutdown ParallelCompaction reducer executor after use (CASSANDRA-3711) * Avoid < 0 value for pending tasks in leveled compaction (CASSANDRA-3693) + * Support TimeUUID in CassandraStorage (CASSANDRA-3327) + * Check schema is ready before continuin boostrapping (CASSANDRA-3629) + * Catch overflows during parsing of chunk_length_kb (CASSANDRA-3644) + * Improve stream protocol mismatch errors (CASSANDRA-3652) + * Avoid multiple thread doing HH to the same target (CASSANDRA-3681) + * Add JMX property for rp_timeout_in_ms (CASSANDRA-2940) Merged from 0.8: * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656) * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626) diff --git a/NEWS.txt b/NEWS.txt index 6d956504ca..2356faad6f 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -8,6 +8,14 @@ upgrade, just in case you need to roll back to the previous version. (Cassandra version X + 1 will always be able to read data files created by version X, but the inverse is not necessarily the case.) +1.0.7 +===== + +Upgrading +--------- + - Nothing specific to 1.0.7, please report to instruction for 1.0.6 + + 1.0.6 ===== @@ -21,6 +29,13 @@ Upgrading setting the right value and then run scrub on the column family. - Please report to instruction for 1.0.5 if coming from an older version. +Other +----- + - Adds new setstreamthroughput to nodetool to configure streaming + throttling + - Adds JMX property to get/set rpc_timeout_in_ms at runtime + - Allow configuring (per-CF) bloom_filter_fp_chance + 1.0.5 ===== diff --git a/build.xml b/build.xml index 0cb7f9e419..7169ff05f5 100644 --- a/build.xml +++ b/build.xml @@ -25,7 +25,7 @@ - + diff --git a/debian/changelog b/debian/changelog index 9047f19df5..70578c8ea1 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +cassandra (1.0.7) unstable; urgency=low + + * New release + + -- Sylvain Lebresne Wed, 11 Jan 2012 09:53:43 +0100 + cassandra (1.0.6) unstable; urgency=low * New release From 185eca5d1fa7e384bb888c144d06abbced0fd577 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 11 Jan 2012 11:44:32 -0600 Subject: [PATCH 4/4] prevent slow clients from postponing shutdown indefinitely patch by jbellis; reviewed by brandonwilliams for CASSANDRA-3727 --- .../DebuggableThreadPoolExecutor.java | 2 +- .../concurrent/NamedThreadFactory.java | 1 + .../org/apache/cassandra/db/Memtable.java | 8 +++- .../cassandra/net/MessagingService.java | 10 +---- .../service/AbstractCassandraDaemon.java | 8 ++-- .../cassandra/service/StorageService.java | 9 ++-- .../thrift/CustomTThreadPoolServer.java | 43 +++++++++---------- .../apache/cassandra/utils/ExpiringMap.java | 17 ++++++++ .../apache/cassandra/service/RemoveTest.java | 2 +- 9 files changed, 59 insertions(+), 41 deletions(-) diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 7a344d8896..f111d37f74 100644 --- a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java +++ b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java @@ -108,7 +108,7 @@ public class DebuggableThreadPoolExecutor extends ThreadPoolExecutor protected void onFinalRejection(Runnable task) {} @Override - public void afterExecute(Runnable r, Throwable t) + protected void afterExecute(Runnable r, Throwable t) { super.afterExecute(r,t); logExceptionsAfterExecute(r, t); diff --git a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java index 4cee8dcd47..a60a0d5412 100644 --- a/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java +++ b/src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java @@ -50,6 +50,7 @@ public class NamedThreadFactory implements ThreadFactory String name = id + ":" + n.getAndIncrement(); Thread thread = new Thread(runnable, name); thread.setPriority(priority); + thread.setDaemon(true); return thread; } } diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index fdafc6df8b..412b800898 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -31,6 +31,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.db.columniterator.IColumnIterator; import org.apache.cassandra.db.columniterator.SimpleAbstractColumnIterator; import org.apache.cassandra.db.commitlog.ReplayPosition; @@ -57,7 +58,12 @@ public class Memtable // we're careful to only allow one count to run at a time because counting is slow // (can be minutes, for a large memtable and a busy server), so we could keep memtables // alive after they're flushed and would otherwise be GC'd. - private static final ExecutorService meterExecutor = new ThreadPoolExecutor(1, 1, Integer.MAX_VALUE, TimeUnit.MILLISECONDS, new SynchronousQueue()) + private static final ExecutorService meterExecutor = new DebuggableThreadPoolExecutor(1, + 1, + Integer.MAX_VALUE, + TimeUnit.MILLISECONDS, + new SynchronousQueue(), + new NamedThreadFactory("MemoryMeter")) { @Override protected void afterExecute(Runnable r, Throwable t) diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 1526fa3552..9ff110e4a2 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -475,15 +475,9 @@ public final class MessagingService implements MessagingServiceMBean } /** - * There isn't a good way to shut down the MessagingService. One problem (but not the only one) - * is that StorageProxy has no way to communicate back to clients, "I'm nominally alive, but I can't - * send that request to the nodes with your data." Neither TimedOut nor Unavailable is appropriate - * to return in that situation. - * - * So instead of shutting down MS and letting StorageProxy/clients cope somehow, we shut down - * the Thrift service and then wait for all the outstanding requests to finish or timeout. + * Wait for callbacks and don't allow any more to be created (since they could require writing hints) */ - public void waitForCallbacks() + public void shutdown() { logger_.info("Waiting for messaging service to quiesce"); // We may need to schedule hints on the mutation stage, so it's erroneous to shut down the mutation stage first diff --git a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java index 1a3ebc9f2a..028f82fcf0 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -31,6 +31,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.Schema; import org.apache.cassandra.gms.Gossiper; import org.apache.log4j.PropertyConfigurator; @@ -391,13 +392,16 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon /** * A subclass of Java's ThreadPoolExecutor which implements Jetty's ThreadPool * interface (for integration with Avro), and performs ClientState cleanup. + * + * (Note that the tasks being executed perform their own while-command-process + * loop until the client disconnects.) */ public static class CleaningThreadPool extends ThreadPoolExecutor { private ThreadLocal state; public CleaningThreadPool(ThreadLocal state, int minWorkerThread, int maxWorkerThreads) { - super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue()); + super(minWorkerThread, maxWorkerThreads, 60, TimeUnit.SECONDS, new SynchronousQueue(), new NamedThreadFactory("Thrift")); this.state = state; } @@ -408,7 +412,5 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon DebuggableThreadPoolExecutor.logExceptionsAfterExecute(r, t); state.get().logout(); } - - } } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 9c1195d5d7..33f58a00c6 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -323,7 +323,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe daemon.startRPCServer(); } - // should only be called via JMX public void stopRPCServer() { if (daemon == null) @@ -347,7 +346,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Gossiper.instance.unregister(migrationManager); Gossiper.instance.unregister(this); Gossiper.instance.stop(); - MessagingService.instance().waitForCallbacks(); + MessagingService.instance().shutdown(); // give it a second so that task accepted before the MessagingService shutdown gets submitted to the stage (to avoid RejectedExecutionException) try { Thread.sleep(1000L); } catch (InterruptedException e) {} StageManager.shutdownNow(); @@ -449,7 +448,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // In-progress writes originating here could generate hints to be written, so shut down MessagingService // before mutation stage, so we can get all the hints saved before shutting down - MessagingService.instance().waitForCallbacks(); + MessagingService.instance().shutdown(); mutationStage.shutdown(); mutationStage.awaitTermination(3600, TimeUnit.SECONDS); StorageProxy.instance.verifyNoHintsInProgress(); @@ -2110,7 +2109,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe public void run() { Gossiper.instance.stop(); - MessagingService.instance().waitForCallbacks(); + MessagingService.instance().shutdown(); StageManager.shutdownNow(); setMode(Mode.DECOMMISSIONED, true); // let op be responsible for killing the process @@ -2512,7 +2511,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Gossiper.instance.stop(); setMode(Mode.DRAINING, "shutting down MessageService", false); - MessagingService.instance().waitForCallbacks(); + MessagingService.instance().shutdown(); setMode(Mode.DRAINING, "waiting for streaming", false); MessagingService.instance().waitForStreaming(); diff --git a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java index c9a5f5b384..161ff121aa 100644 --- a/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java +++ b/src/java/org/apache/cassandra/thrift/CustomTThreadPoolServer.java @@ -119,27 +119,24 @@ public class CustomTThreadPoolServer extends TServer } executorService_.shutdown(); - - // Loop until awaitTermination finally does return without a interrupted - // exception. If we don't do this, then we'll shut down prematurely. We want - // to let the executorService clear it's task queue, closing client sockets - // appropriately. - long timeoutMS = args.stopTimeoutUnit.toMillis(args.stopTimeoutVal); - long now = System.currentTimeMillis(); - while (timeoutMS >= 0) - { - try - { - executorService_.awaitTermination(timeoutMS, TimeUnit.MILLISECONDS); - break; - } - catch (InterruptedException ix) - { - long newnow = System.currentTimeMillis(); - timeoutMS -= (newnow - now); - now = newnow; - } - } + // Thrift's default shutdown waits for the WorkerProcess threads to complete. We do not, + // because doing that allows a client to hold our shutdown "hostage" by simply not sending + // another message after stop is called (since process will block indefinitely trying to read + // the next meessage header). + // + // The "right" fix would be to update thrift to set a socket timeout on client connections + // (and tolerate unintentional timeouts until stopped_ is set). But this requires deep + // changes to the code generator, so simply setting these threads to daemon (in our custom + // CleaningThreadPool) and ignoring them after shutdown is good enough. + // + // Remember, our goal on shutdown is not necessarily that each client request we receive + // gets answered first [to do that, you should redirect clients to a different coordinator + // first], but rather (1) to make sure that for each update we ack as successful, we generate + // hints for any non-responsive replicas, and (2) to make sure that we quickly stop + // accepting client connections so shutdown can continue. Not waiting for the WorkerProcess + // threads here accomplishes (2); MessagingService's shutdown method takes care of (1). + // + // See CASSANDRA-3335 and CASSANDRA-3727. } public void stop() @@ -184,7 +181,9 @@ public class CustomTThreadPoolServer extends TServer inputProtocol = inputProtocolFactory_.getProtocol(inputTransport); outputProtocol = outputProtocolFactory_.getProtocol(outputTransport); // we check stopped_ first to make sure we're not supposed to be shutting - // down. this is necessary for graceful shutdown. + // down. this is necessary for graceful shutdown. (but not sufficient, + // since process() can take arbitrarily long waiting for client input. + // See comments at the end of serve().) while (!stopped_ && processor.process(inputProtocol, outputProtocol)) { inputProtocol = inputProtocolFactory_.getProtocol(inputTransport); diff --git a/src/java/org/apache/cassandra/utils/ExpiringMap.java b/src/java/org/apache/cassandra/utils/ExpiringMap.java index ffd3c2e8a0..0672259eb0 100644 --- a/src/java/org/apache/cassandra/utils/ExpiringMap.java +++ b/src/java/org/apache/cassandra/utils/ExpiringMap.java @@ -30,6 +30,7 @@ import org.cliffc.high_scale_lib.NonBlockingHashMap; public class ExpiringMap { private static final Logger logger = LoggerFactory.getLogger(ExpiringMap.class); + private volatile boolean shutdown; private static class CacheableObject { @@ -104,6 +105,7 @@ public class ExpiringMap public void shutdown() { + shutdown = true; while (!cache.isEmpty()) { logger.trace("Waiting for {} entries before shutting down ExpiringMap", cache.size()); @@ -131,6 +133,21 @@ public class ExpiringMap public V put(K key, V value, long timeout) { + if (shutdown) + { + // StorageProxy isn't equipped to deal with "I'm nominally alive, but I can't send any messages out." + // So we'll just sit on this thread until the rest of the server shutdown completes. + // + // See comments in CustomTThreadPoolServer.serve, CASSANDRA-3335, and CASSANDRA-3727. + try + { + Thread.sleep(Long.MAX_VALUE); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + } CacheableObject previous = cache.put(key, new CacheableObject(value, timeout)); return (previous == null) ? null : previous.getValue(); } diff --git a/test/unit/org/apache/cassandra/service/RemoveTest.java b/test/unit/org/apache/cassandra/service/RemoveTest.java index 3034ab3399..7ee7d69eb7 100644 --- a/test/unit/org/apache/cassandra/service/RemoveTest.java +++ b/test/unit/org/apache/cassandra/service/RemoveTest.java @@ -85,7 +85,7 @@ public class RemoveTest extends CleanupHelper { SinkManager.clear(); MessagingService.instance().clearCallbacksUnsafe(); - MessagingService.instance().waitForCallbacks(); + MessagingService.instance().shutdown(); ss.setPartitionerUnsafe(oldPartitioner); }