diff --git a/CHANGES.txt b/CHANGES.txt index eccb32d71d..3bdd26f988 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -64,6 +64,13 @@ * 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) + * 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 036399b17d..0f54918015 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -67,6 +67,14 @@ Features - Streaming is now multithreaded. +1.0.7 +===== + +Upgrading +--------- + - Nothing specific to 1.0.7, please report to instruction for 1.0.6 + + 1.0.6 ===== @@ -80,6 +88,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/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 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 diff --git a/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java b/src/java/org/apache/cassandra/concurrent/DebuggableThreadPoolExecutor.java index 11c15810b6..60265b1d81 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 bf8382e356..c0c6289f78 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -32,6 +32,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; @@ -58,7 +59,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/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java index f67a6e32c8..6c3d8c30e7 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; @@ -203,11 +204,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() @@ -413,12 +417,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); } } diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 2845495b20..c54cdec2de 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -533,15 +533,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 68e64cfbbb..ff3446e82e 100644 --- a/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java @@ -30,27 +30,27 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import com.google.common.collect.Iterables; import org.apache.log4j.PropertyConfigurator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; +import org.apache.cassandra.concurrent.NamedThreadFactory; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; -import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.SystemTable; import org.apache.cassandra.db.Table; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.migration.Migration; +import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.utils.CLibrary; import org.apache.cassandra.utils.Mx4jTool; -import com.google.common.collect.Iterables; - /** * The CassandraDaemon is an abstraction for a Cassandra daemon * service, which defines not only a way to activate and deactivate it, but also @@ -403,13 +403,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; } @@ -420,7 +423,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 4fe8c50061..db95259035 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -327,7 +327,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe daemon.startRPCServer(); } - // should only be called via JMX public void stopRPCServer() { if (daemon == null) @@ -351,7 +350,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(); @@ -454,7 +453,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(); @@ -2118,7 +2117,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 @@ -2520,7 +2519,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); }