diff --git a/CHANGES.txt b/CHANGES.txt index f41e7c380f..08c31a9523 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -33,6 +33,7 @@ * improve concurrency of repair (CASSANDRA-6455, 8208) 2.1.3 + * Centralize shared executors (CASSANDRA-8055) * Fix filtering for CONTAINS (KEY) relations on frozen collection clustering columns when the query is restricted to a single partition (CASSANDRA-8203) diff --git a/src/java/org/apache/cassandra/auth/Auth.java b/src/java/org/apache/cassandra/auth/Auth.java index 8c12df6934..041ce2b846 100644 --- a/src/java/org/apache/cassandra/auth/Auth.java +++ b/src/java/org/apache/cassandra/auth/Auth.java @@ -29,6 +29,7 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.KSMetaData; @@ -189,15 +190,13 @@ public class Auth implements AuthMBean // the delay is here to give the node some time to see its peers - to reduce // "Skipped default superuser setup: some nodes were not ready" log spam. // It's the only reason for the delay. - StorageService.tasks.schedule(new Runnable() - { - public void run() - { - setupDefaultSuperuser(); - } - }, - SUPERUSER_SETUP_DELAY, - TimeUnit.MILLISECONDS); + ScheduledExecutors.nonPeriodicTasks.schedule(new Runnable() + { + public void run() + { + setupDefaultSuperuser(); + } + }, SUPERUSER_SETUP_DELAY, TimeUnit.MILLISECONDS); try { diff --git a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java index 1218ee2443..957077042e 100644 --- a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java +++ b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java @@ -30,6 +30,7 @@ import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.QueryOptions; @@ -37,7 +38,6 @@ import org.apache.cassandra.cql3.statements.SelectStatement; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.exceptions.*; import org.apache.cassandra.service.QueryState; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.ByteBufferUtil; import org.mindrot.jbcrypt.BCrypt; @@ -169,15 +169,13 @@ public class PasswordAuthenticator implements ISaslAwareAuthenticator // the delay is here to give the node some time to see its peers - to reduce // "skipped default user setup: some nodes are were not ready" log spam. // It's the only reason for the delay. - StorageService.tasks.schedule(new Runnable() - { - public void run() - { - setupDefaultUser(); - } - }, - Auth.SUPERUSER_SETUP_DELAY, - TimeUnit.MILLISECONDS); + ScheduledExecutors.nonPeriodicTasks.schedule(new Runnable() + { + public void run() + { + setupDefaultUser(); + } + }, Auth.SUPERUSER_SETUP_DELAY, TimeUnit.MILLISECONDS); try { diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java index 4ac2328309..92e6a6d0bb 100644 --- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java +++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java @@ -27,6 +27,7 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; @@ -39,7 +40,6 @@ import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.util.*; import org.apache.cassandra.service.CacheService; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.Pair; @@ -115,10 +115,10 @@ public class AutoSavingCache extends InstrumentingCache(), new NamedThreadFactory("MemtableFlushWriter"), "internal"); + // post-flush executor is single threaded to provide guarantee that any flush Future on a CF will never return until prior flushes have completed - public static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor(1, - StageManager.KEEPALIVE, - TimeUnit.SECONDS, - new LinkedBlockingQueue(), - new NamedThreadFactory("MemtablePostFlush"), - "internal"); - public static final ExecutorService reclaimExecutor = new JMXEnabledThreadPoolExecutor(1, StageManager.KEEPALIVE, - TimeUnit.SECONDS, - new LinkedBlockingQueue(), - new NamedThreadFactory("MemtableReclaimMemory"), - "internal"); + private static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor(1, + StageManager.KEEPALIVE, + TimeUnit.SECONDS, + new LinkedBlockingQueue(), + new NamedThreadFactory("MemtablePostFlush"), + "internal"); + + private static final ExecutorService reclaimExecutor = new JMXEnabledThreadPoolExecutor(1, + StageManager.KEEPALIVE, + TimeUnit.SECONDS, + new LinkedBlockingQueue(), + new NamedThreadFactory("MemtableReclaimMemory"), + "internal"); public final Keyspace keyspace; public final String name; @@ -138,6 +137,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public final ColumnFamilyMetrics metric; public volatile long sampleLatencyNanos; + public static void shutdownPostFlushExecutor() throws InterruptedException + { + postFlushExecutor.shutdown(); + postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS); + } + public void reload() { // metadata object has been mutated directly. make all the members jibe with new settings. @@ -192,7 +197,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } }; - StorageService.scheduledTasks.schedule(runnable, period, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.schedule(runnable, period, TimeUnit.MILLISECONDS); } } @@ -314,7 +319,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean throw new RuntimeException(e); } logger.debug("retryPolicy for {} is {}", name, this.metadata.getSpeculativeRetry()); - StorageService.optionalTasks.scheduleWithFixedDelay(new Runnable() + ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(new Runnable() { public void run() { diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index bab9aa8cd2..b976ef7425 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.composites.CellName; @@ -175,7 +176,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean metrics.log(); } }; - StorageService.optionalTasks.scheduleWithFixedDelay(runnable, 10, 10, TimeUnit.MINUTES); + ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(runnable, 10, 10, TimeUnit.MINUTES); } private static void deleteHint(ByteBuffer tokenBytes, CellName columnName, long timestamp) @@ -227,7 +228,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean } } }; - StorageService.optionalTasks.submit(runnable); + ScheduledExecutors.optionalTasks.submit(runnable); } //foobar @@ -248,7 +249,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean } } }; - StorageService.optionalTasks.submit(runnable).get(); + ScheduledExecutors.optionalTasks.submit(runnable).get(); } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java index 1b1a1e0935..6cba603773 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java @@ -52,7 +52,7 @@ public class CommitLogArchiver } public final Map> archivePending = new ConcurrentHashMap>(); - public final ExecutorService executor = new JMXEnabledThreadPoolExecutor("CommitLogArchiver"); + private final ExecutorService executor = new JMXEnabledThreadPoolExecutor("CommitLogArchiver"); private final String archiveCommand; private final String restoreCommand; private final String restoreDirectories; diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java b/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java index 2d453431fd..2aafd3dc51 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java @@ -28,9 +28,9 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.db.DataTracker; import org.apache.cassandra.db.SystemKeyspace; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; public class SSTableDeletingTask implements Runnable @@ -70,7 +70,7 @@ public class SSTableDeletingTask implements Runnable public void schedule() { - StorageService.tasks.submit(this); + ScheduledExecutors.nonPeriodicTasks.submit(this); } public void run() @@ -120,7 +120,7 @@ public class SSTableDeletingTask implements Runnable } }; - FBUtilities.waitOnFuture(StorageService.tasks.schedule(runnable, 0, TimeUnit.MILLISECONDS)); + FBUtilities.waitOnFuture(ScheduledExecutors.nonPeriodicTasks.schedule(runnable, 0, TimeUnit.MILLISECONDS)); } } 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 7eac94b297..54a244b838 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -1,5 +1,3 @@ -package org.apache.cassandra.io.sstable.format; - /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -17,19 +15,30 @@ package org.apache.cassandra.io.sstable.format; * See the License for the specific language governing permissions and * limitations under the License. */ +package org.apache.cassandra.io.sstable.format; + +import java.io.*; +import java.nio.ByteBuffer; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; -import com.clearspring.analytics.stream.cardinality.CardinalityMergeException; -import com.clearspring.analytics.stream.cardinality.ICardinality; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Predicate; import com.google.common.collect.Iterators; import com.google.common.collect.Ordering; import com.google.common.primitives.Longs; import com.google.common.util.concurrent.RateLimiter; + +import com.clearspring.analytics.stream.cardinality.CardinalityMergeException; +import com.clearspring.analytics.stream.cardinality.ICardinality; import org.apache.cassandra.cache.CachingOptions; import org.apache.cassandra.cache.InstrumentingCache; import org.apache.cassandra.cache.KeyCacheKey; import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.*; import org.apache.cassandra.db.*; import org.apache.cassandra.db.columniterator.OnDiskAtomIterator; @@ -55,17 +64,8 @@ import org.apache.cassandra.utils.concurrent.OpOrder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.*; -import java.nio.ByteBuffer; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; - import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR; - /** * SSTableReaders are open()ed by Keyspace.onStart; after that they are created by SSTableWriter.renameAndOpen. * Do not re-call open() on existing SSTable files; use the references kept by ColumnFamilyStore post-start instead. @@ -571,7 +571,7 @@ public abstract class SSTableReader extends SSTable else barrier = null; - StorageService.tasks.execute(new Runnable() + ScheduledExecutors.nonPeriodicTasks.execute(new Runnable() { public void run() { diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index efd9b3a58a..080caa557c 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -32,7 +32,7 @@ import sun.nio.ch.DirectBuffer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.config.Config; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.BlacklistedDirectories; import org.apache.cassandra.db.Keyspace; @@ -320,7 +320,7 @@ public class FileUtils deleteWithConfirm(new File(file)); } }; - StorageService.tasks.execute(runnable); + ScheduledExecutors.nonPeriodicTasks.execute(runnable); } public static String stringifyFileSize(double value) diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 49442c815a..e4b714c459 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import javax.management.MBeanServer; import javax.management.ObjectName; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; @@ -84,8 +85,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa reset(); } }; - StorageService.scheduledTasks.scheduleWithFixedDelay(update, UPDATE_INTERVAL_IN_MS, UPDATE_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); - StorageService.scheduledTasks.scheduleWithFixedDelay(reset, RESET_INTERVAL_IN_MS, RESET_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(update, UPDATE_INTERVAL_IN_MS, UPDATE_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(reset, RESET_INTERVAL_IN_MS, RESET_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); registerMBean(); } diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index b9490b0e0f..83ca384e3d 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -37,6 +37,8 @@ import com.google.common.collect.Lists; import org.cliffc.high_scale_lib.NonBlockingHashMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.concurrent.TracingAwareExecutorService; @@ -330,7 +332,7 @@ public final class MessagingService implements MessagingServiceMBean logDroppedMessages(); } }; - StorageService.scheduledTasks.scheduleWithFixedDelay(logDropped, LOG_DROPPED_INTERVAL_IN_MS, LOG_DROPPED_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(logDropped, LOG_DROPPED_INTERVAL_IN_MS, LOG_DROPPED_INTERVAL_IN_MS, TimeUnit.MILLISECONDS); Function>, ?> timeoutReporter = new Function>, Object>() { diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index d3ba08528e..e5644f0997 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory; import com.addthis.metrics.reporter.config.ReporterConfig; import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.CFMetaData; @@ -348,7 +349,7 @@ public class CassandraDaemon } } }; - StorageService.optionalTasks.schedule(runnable, 5 * 60, TimeUnit.SECONDS); + ScheduledExecutors.optionalTasks.schedule(runnable, 5 * 60, TimeUnit.SECONDS); SystemKeyspace.finishStartup(); diff --git a/src/java/org/apache/cassandra/service/LoadBroadcaster.java b/src/java/org/apache/cassandra/service/LoadBroadcaster.java index 4996e52fee..d12ffba8d9 100644 --- a/src/java/org/apache/cassandra/service/LoadBroadcaster.java +++ b/src/java/org/apache/cassandra/service/LoadBroadcaster.java @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.gms.*; public class LoadBroadcaster implements IEndpointStateChangeSubscriber @@ -91,7 +92,7 @@ public class LoadBroadcaster implements IEndpointStateChangeSubscriber StorageService.instance.valueFactory.load(StorageService.instance.getLoad())); } }; - StorageService.scheduledTasks.scheduleWithFixedDelay(runnable, 2 * Gossiper.intervalInMillis, BROADCAST_INTERVAL, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(runnable, 2 * Gossiper.intervalInMillis, BROADCAST_INTERVAL, TimeUnit.MILLISECONDS); } } diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java index 8c3199f258..73bab6687b 100644 --- a/src/java/org/apache/cassandra/service/MigrationManager.java +++ b/src/java/org/apache/cassandra/service/MigrationManager.java @@ -32,6 +32,7 @@ import java.lang.management.RuntimeMXBean; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.config.CFMetaData; @@ -130,7 +131,7 @@ public class MigrationManager submitMigrationTask(endpoint); } }; - StorageService.optionalTasks.schedule(runnable, MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS); + ScheduledExecutors.optionalTasks.schedule(runnable, MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS); } } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 9cc643baba..e9e36f4f19 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -51,12 +51,10 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.time.DurationFormatUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import org.apache.cassandra.auth.Auth; import org.apache.cassandra.concurrent.*; -import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.config.KSMetaData; -import org.apache.cassandra.config.Schema; +import org.apache.cassandra.config.*; import org.apache.cassandra.db.*; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.compaction.CompactionManager; @@ -79,7 +77,6 @@ import org.apache.cassandra.net.ResponseVerbHandler; import org.apache.cassandra.repair.RepairMessageVerbHandler; import org.apache.cassandra.repair.RepairSessionResult; import org.apache.cassandra.repair.messages.RepairOption; -import org.apache.cassandra.repair.RepairResult; import org.apache.cassandra.repair.RepairSession; import org.apache.cassandra.service.paxos.CommitVerbHandler; import org.apache.cassandra.service.paxos.PrepareVerbHandler; @@ -120,24 +117,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE return 30 * 1000; } - /** - * This pool is used for periodic short (sub-second) tasks. - */ - public static final DebuggableScheduledThreadPoolExecutor scheduledTasks = new DebuggableScheduledThreadPoolExecutor("ScheduledTasks"); - - /** - * This pool is used by tasks that can have longer execution times, and usually are non periodic. - */ - public static final DebuggableScheduledThreadPoolExecutor tasks = new DebuggableScheduledThreadPoolExecutor("NonPeriodicTasks"); - /** - * tasks that do not need to be waited for on shutdown/drain - */ - public static final DebuggableScheduledThreadPoolExecutor optionalTasks = new DebuggableScheduledThreadPoolExecutor("OptionalTasks"); - static - { - tasks.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); - } - /* This abstraction maintains the token/endpoint metadata information */ private TokenMetadata tokenMetadata = new TokenMetadata(); @@ -551,7 +530,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE if (daemon != null) shutdownClientServers(); - optionalTasks.shutdown(); + ScheduledExecutors.optionalTasks.shutdown(); Gossiper.instance.stop(); // In-progress writes originating here could generate hints to be written, so shut down MessagingService @@ -587,8 +566,8 @@ public class StorageService extends NotificationBroadcasterSupport implements IE CommitLog.instance.shutdownBlocking(); // wait for miscellaneous tasks like sstable and commitlog segment deletion - tasks.shutdown(); - if (!tasks.awaitTermination(1, TimeUnit.MINUTES)) + ScheduledExecutors.nonPeriodicTasks.shutdown(); + if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, TimeUnit.MINUTES)) logger.warn("Miscellaneous task executor still busy after one minute; proceeding with shutdown"); } }, "StorageServiceShutdownHook"); @@ -3580,7 +3559,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } setMode(Mode.DRAINING, "starting drain process", true); shutdownClientServers(); - optionalTasks.shutdown(); + ScheduledExecutors.optionalTasks.shutdown(); Gossiper.instance.stop(); setMode(Mode.DRAINING, "shutting down MessageService", false); @@ -3625,21 +3604,19 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } FBUtilities.waitOnFutures(flushes); - BatchlogManager.batchlogTasks.shutdown(); - BatchlogManager.batchlogTasks.awaitTermination(60, TimeUnit.SECONDS); + BatchlogManager.shutdown(); // whilst we've flushed all the CFs, which will have recycled all completed segments, we want to ensure // there are no segments to replay, so we force the recycling of any remaining (should be at most one) CommitLog.instance.forceRecycleAllSegments(); - ColumnFamilyStore.postFlushExecutor.shutdown(); - ColumnFamilyStore.postFlushExecutor.awaitTermination(60, TimeUnit.SECONDS); + ColumnFamilyStore.shutdownPostFlushExecutor(); CommitLog.instance.shutdownBlocking(); // wait for miscellaneous tasks like sstable and commitlog segment deletion - tasks.shutdown(); - if (!tasks.awaitTermination(1, TimeUnit.MINUTES)) + ScheduledExecutors.nonPeriodicTasks.shutdown(); + if (!ScheduledExecutors.nonPeriodicTasks.awaitTermination(1, TimeUnit.MINUTES)) logger.warn("Miscellaneous task executor still busy after one minute; proceeding with shutdown"); setMode(Mode.DRAINED, true); diff --git a/src/java/org/apache/cassandra/utils/ResourceWatcher.java b/src/java/org/apache/cassandra/utils/ResourceWatcher.java index 2dfab95f14..5e7cbdd4f3 100644 --- a/src/java/org/apache/cassandra/utils/ResourceWatcher.java +++ b/src/java/org/apache/cassandra/utils/ResourceWatcher.java @@ -23,13 +23,13 @@ import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.concurrent.ScheduledExecutors; public class ResourceWatcher { public static void watch(String resource, Runnable callback, int period) { - StorageService.scheduledTasks.scheduleWithFixedDelay(new WatchedResource(resource, callback), period, period, TimeUnit.MILLISECONDS); + ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(new WatchedResource(resource, callback), period, period, TimeUnit.MILLISECONDS); } public static class WatchedResource implements Runnable diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index d6c3182e7f..9105b9d4b5 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -37,6 +37,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.Directories; @@ -45,7 +46,6 @@ import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.exceptions.*; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.serializers.TypeSerializer; -import org.apache.cassandra.service.StorageService; /** * Base class for CQL tests. @@ -97,7 +97,7 @@ public abstract class CQLTester currentTypes.clear(); // We want to clean up after the test, but dropping a table is rather long so just do that asynchronously - StorageService.optionalTasks.execute(new Runnable() + ScheduledExecutors.optionalTasks.execute(new Runnable() { public void run() { @@ -114,7 +114,7 @@ public abstract class CQLTester // mono-threaded, just push a task on the queue to find when it's empty. No perfect but good enough. final CountDownLatch latch = new CountDownLatch(1); - StorageService.tasks.execute(new Runnable() + ScheduledExecutors.nonPeriodicTasks.execute(new Runnable() { public void run() { diff --git a/test/unit/org/apache/cassandra/db/KeyCacheTest.java b/test/unit/org/apache/cassandra/db/KeyCacheTest.java index 629b414dc4..e31b4398cd 100644 --- a/test/unit/org/apache/cassandra/db/KeyCacheTest.java +++ b/test/unit/org/apache/cassandra/db/KeyCacheTest.java @@ -33,13 +33,13 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; import org.apache.cassandra.cache.KeyCacheKey; import org.apache.cassandra.config.KSMetaData; +import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.db.composites.*; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.locator.SimpleStrategy; import org.apache.cassandra.service.CacheService; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.ByteBufferUtil; import static org.junit.Assert.assertEquals; @@ -178,8 +178,8 @@ public class KeyCacheTest for (SSTableReader reader : readers) reader.releaseReference(); - Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS); - while (StorageService.tasks.getActiveCount() + StorageService.tasks.getQueue().size() > 0); + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);; + while (ScheduledExecutors.nonPeriodicTasks.getActiveCount() + ScheduledExecutors.nonPeriodicTasks.getQueue().size() > 0); // after releasing the reference this should drop to 2 assertKeyCacheSize(2, KEYSPACE1, COLUMN_FAMILY1);