natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-2.1' into trunk 

Conflicts:
	src/java/org/apache/cassandra/cql3/QueryProcessor.java
	src/java/org/apache/cassandra/db/ColumnFamilyStore.java
	src/java/org/apache/cassandra/db/HintedHandOffManager.java
	src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
	src/java/org/apache/cassandra/service/StorageService.java
	test/unit/org/apache/cassandra/db/KeyCacheTest.java
This commit is contained in:
Aleksey Yeschenko 2014-11-20 02:10:40 +03:00
parent 1bd5c64ba1 4397c34476
commit 8badc28566
22 changed files with 156 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

17
src/java/org/apache/cassandra/auth/Auth.java
View File

@ -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
{

18
src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
View File

@ -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
{

10
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -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<K extends CacheKey, V> extends InstrumentingCache<K
submitWrite(keysToSave);
}
};
saveTask = StorageService.optionalTasks.scheduleWithFixedDelay(runnable,
savePeriodInSeconds,
savePeriodInSeconds,
TimeUnit.SECONDS);
saveTask = ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(runnable,
savePeriodInSeconds,
savePeriodInSeconds,
TimeUnit.SECONDS);
}
}

43
src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java Normal file
View File

@ -0,0 +1,43 @@
/*
* 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.concurrent;
/**
* Centralized location for shared executors
*/
public class ScheduledExecutors
{
/**
* This pool is used for periodic short (sub-second) tasks.
*/
public static final DebuggableScheduledThreadPoolExecutor scheduledTasks = new DebuggableScheduledThreadPoolExecutor("ScheduledTasks");
/**
* This executor is used for tasks that can have longer execution times, and usually are non periodic.
*/
public static final DebuggableScheduledThreadPoolExecutor nonPeriodicTasks = new DebuggableScheduledThreadPoolExecutor("NonPeriodicTasks");
static
{
nonPeriodicTasks.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
}
/**
* This executor is used for tasks that do not need to be waited for on shutdown/drain.
*/
public static final DebuggableScheduledThreadPoolExecutor optionalTasks = new DebuggableScheduledThreadPoolExecutor("OptionalTasks");
}

5
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -31,6 +31,8 @@ import org.antlr.runtime.*;
import org.github.jamm.MemoryMeter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.cql3.functions.*;
import org.apache.cassandra.cql3.statements.*;
import org.apache.cassandra.db.*;
@ -88,7 +90,6 @@ public class QueryProcessor implements QueryHandler
public static final CQLMetrics metrics = new CQLMetrics();
private static final AtomicInteger lastMinuteEvictionsCount = new AtomicInteger(0);
private static final ScheduledExecutorService evictionCheckTimer = Executors.newScheduledThreadPool(1);
static
{
@ -117,7 +118,7 @@ public class QueryProcessor implements QueryHandler
})
.build();
evictionCheckTimer.scheduleAtFixedRate(new Runnable()
ScheduledExecutors.scheduledTasks.scheduleAtFixedRate(new Runnable()
{
public void run()
{

8
src/java/org/apache/cassandra/db/BatchlogManager.java
View File

@ -69,7 +69,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private final AtomicLong totalBatchesReplayed = new AtomicLong();
// Single-thread executor service for scheduling and serializing log replay.
public static final ScheduledExecutorService batchlogTasks = new DebuggableScheduledThreadPoolExecutor("BatchlogTasks");
private static final ScheduledExecutorService batchlogTasks = new DebuggableScheduledThreadPoolExecutor("BatchlogTasks");
public void start()
{
@ -94,6 +94,12 @@ public class BatchlogManager implements BatchlogManagerMBean
batchlogTasks.scheduleWithFixedDelay(runnable, StorageService.RING_DELAY, REPLAY_INTERVAL, TimeUnit.MILLISECONDS);
}
public static void shutdown() throws InterruptedException
{
batchlogTasks.shutdown();
batchlogTasks.awaitTermination(60, TimeUnit.SECONDS);
}
public int countAllBatches()
{
String query = String.format("SELECT count(*) FROM %s.%s", SystemKeyspace.NAME, SystemKeyspace.BATCHLOG_TABLE);

49
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -33,19 +33,13 @@ import com.google.common.collect.*;
import com.google.common.util.concurrent.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.SSTableWriter;
import org.apache.cassandra.io.sstable.format.Version;
import org.json.simple.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cache.*;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.concurrent.*;
import org.apache.cassandra.config.*;
import org.apache.cassandra.config.CFMetaData.SpeculativeRetry;
import org.apache.cassandra.db.commitlog.CommitLog;
@ -65,9 +59,11 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.sstable.format.*;
import org.apache.cassandra.io.sstable.metadata.CompactionMetadata;
import org.apache.cassandra.io.sstable.metadata.MetadataType;
import org.apache.cassandra.io.util.FileUtils;
@ -90,18 +86,21 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
new LinkedBlockingQueue<Runnable>(),
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<Runnable>(),
new NamedThreadFactory("MemtablePostFlush"),
"internal");
public static final ExecutorService reclaimExecutor = new JMXEnabledThreadPoolExecutor(1, StageManager.KEEPALIVE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory("MemtableReclaimMemory"),
"internal");
private static final ExecutorService postFlushExecutor = new JMXEnabledThreadPoolExecutor(1,
StageManager.KEEPALIVE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory("MemtablePostFlush"),
"internal");
private static final ExecutorService reclaimExecutor = new JMXEnabledThreadPoolExecutor(1,
StageManager.KEEPALIVE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
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()
{

7
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -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();
}

2
src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java
View File

@ -52,7 +52,7 @@ public class CommitLogArchiver
}
public final Map<String, Future<?>> archivePending = new ConcurrentHashMap<String, Future<?>>();
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;

6
src/java/org/apache/cassandra/io/sstable/SSTableDeletingTask.java
View File

@ -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));
}
}

28
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -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()
{

4
src/java/org/apache/cassandra/io/util/FileUtils.java
View File

@ -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)

5
src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java
View File

@ -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();
}

4
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -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<Pair<Integer, ExpiringMap.CacheableObject<CallbackInfo>>, ?> timeoutReporter = new Function<Pair<Integer, ExpiringMap.CacheableObject<CallbackInfo>>, Object>()
{

3
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -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();

3
src/java/org/apache/cassandra/service/LoadBroadcaster.java
View File

@ -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);
}
}

3
src/java/org/apache/cassandra/service/MigrationManager.java
View File

@ -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);
}
}

43
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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);

4
src/java/org/apache/cassandra/utils/ResourceWatcher.java
View File

@ -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

6
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -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()
{

6
test/unit/org/apache/cassandra/db/KeyCacheTest.java
View File

@ -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);