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

inline IStage.executorService, removing the useless Stage wrappers 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@898049 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-01-11 20:14:53 +00:00
parent 6b5d8bf802
commit a6cd2727d0
10 changed files with 36 additions and 341 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
src/java/org/apache/cassandra/concurrent/IStage.java
View File

@ -1,121 +0,0 @@
/**
* 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;
import java.util.concurrent.*;
/**
* An abstraction for stages as described in the SEDA paper by Matt Welsh.
* For reference to the paper look over here
* <a href="http://www.eecs.harvard.edu/~mdw/papers/seda-sosp01.pdf">SEDA: An Architecture for WellConditioned,
Scalable Internet Services</a>.
*/
public interface IStage
{
/**
* Get the name of the associated stage.
* @return name of the associated stage.
*/
public String getName();
/**
* Get the thread pool used by this stage
* internally.
*/
public ExecutorService getInternalThreadPool();
/**
* This method is used to execute a piece of code on
* this stage. The idea is that the <i>run()</i> method
* of this Runnable instance is invoked on a thread from a
* thread pool that belongs to this stage.
* @param runnable instance whose run() method needs to be invoked.
*/
public void execute(Runnable runnable);
/**
* This method is used to execute a piece of code on
* this stage which returns a Future pointer. The idea
* is that the <i>call()</i> method of this Runnable
* instance is invoked on a thread from a thread pool
* that belongs to this stage.
* @param callable instance that needs to be invoked.
* @return the future return object from the callable.
*/
public Future<Object> execute(Callable<Object> callable);
/**
* This method is used to submit tasks to this stage
* that execute periodically.
*
* @param command the task to execute.
* @param delay the time to delay first execution
* @param unit the time unit of the initialDelay and period parameters
* @return the future return object from the runnable.
*/
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
/**
* This method is used to submit tasks to this stage
* that execute periodically.
* @param command the task to execute.
* @param initialDelay the time to delay first execution
* @param period the period between successive executions
* @param unit the time unit of the initialDelay and period parameters
* @return the future return object from the runnable.
*/
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
/**
* This method is used to submit tasks to this stage
* that execute periodically.
* @param command the task to execute.
* @param initialDelay the time to delay first execution
* @param delay the delay between the termination of one execution and the commencement of the next.
* @param unit the time unit of the initialDelay and delay parameters
* @return the future return object from the runnable.
*/
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
/**
* Shutdown the stage. All the threads of this stage
* are forcefully shutdown. Any pending tasks on this
* stage could be dropped or the stage could wait for
* these tasks to be completed. This is however an
* implementation detail.
*/
public void shutdown();
/**
* Checks if the stage has been shutdown.
* @return true if shut down, otherwise false.
*/
public boolean isShutdown();
/**
* This method returns the number of tasks that are
* pending on this stage to be executed.
* @return task count.
*/
public long getPendingTasks();
public long getCompletedTasks();
}

95
src/java/org/apache/cassandra/concurrent/MultiThreadedStage.java
View File

@ -1,95 +0,0 @@
/**
* 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;
import java.util.concurrent.*;
/**
* This class is an implementation of the <i>IStage</i> interface. In particular
* it is for a stage that has a thread pool with multiple threads. For details
* please refer to the <i>IStage</i> documentation.
*/
public class MultiThreadedStage implements IStage
{
private String name_;
private JMXEnabledThreadPoolExecutor executorService_;
public MultiThreadedStage(String name, int numThreads)
{
name_ = name;
executorService_ = new JMXEnabledThreadPoolExecutor( numThreads,
numThreads,
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory(name)
);
}
public String getName()
{
return name_;
}
public ExecutorService getInternalThreadPool()
{
return executorService_;
}
public Future<Object> execute(Callable<Object> callable) {
return executorService_.submit(callable);
}
public void execute(Runnable runnable) {
executorService_.execute(runnable);
}
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
{
throw new UnsupportedOperationException("This operation is not supported");
}
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
throw new UnsupportedOperationException("This operation is not supported");
}
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
throw new UnsupportedOperationException("This operation is not supported");
}
public void shutdown() {
executorService_.shutdownNow();
}
public boolean isShutdown()
{
return executorService_.isShutdown();
}
public long getPendingTasks(){
return executorService_.getPendingTasks();
}
public long getCompletedTasks()
{
return executorService_.getCompletedTasks();
}
}

102
src/java/org/apache/cassandra/concurrent/SingleThreadedStage.java
View File

@ -1,102 +0,0 @@
/**
* 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;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
/**
* This class is an implementation of the <i>IStage</i> interface. In particular
* it is for a stage that has a thread pool with a single thread. For details
* please refer to the <i>IStage</i> documentation.
*/
public class SingleThreadedStage implements IStage
{
protected JMXEnabledThreadPoolExecutor executorService_;
private String name_;
public SingleThreadedStage(String name)
{
executorService_ = new JMXEnabledThreadPoolExecutor(name);
name_ = name;
}
/* Implementing the IStage interface methods */
public String getName()
{
return name_;
}
public ExecutorService getInternalThreadPool()
{
return executorService_;
}
public void execute(Runnable runnable)
{
executorService_.execute(runnable);
}
public Future<Object> execute(Callable<Object> callable)
{
return executorService_.submit(callable);
}
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
{
//return executorService_.schedule(command, delay, unit);
throw new UnsupportedOperationException("This operation is not supported");
}
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)
{
//return executorService_.scheduleAtFixedRate(command, initialDelay, period, unit);
throw new UnsupportedOperationException("This operation is not supported");
}
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
{
//return executorService_.scheduleWithFixedDelay(command, initialDelay, delay, unit);
throw new UnsupportedOperationException("This operation is not supported");
}
public void shutdown()
{
executorService_.shutdownNow();
}
public boolean isShutdown()
{
return executorService_.isShutdown();
}
public long getPendingTasks(){
return executorService_.getPendingTasks();
}
public long getCompletedTasks()
{
return executorService_.getCompletedTasks();
}
}

42
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -21,6 +21,10 @@ package org.apache.cassandra.concurrent;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.net.MessagingService;
@ -35,7 +39,7 @@ import static org.apache.cassandra.config.DatabaseDescriptor.getConcurrentReader
*/
public class StageManager
{
private static Map<String, IStage> stageQueues = new HashMap<String, IStage>();
private static Map<String, ThreadPoolExecutor> stages = new HashMap<String, ThreadPoolExecutor>();
public final static String READ_STAGE = "ROW-READ-STAGE";
public final static String MUTATION_STAGE = "ROW-MUTATION-STAGE";
@ -47,22 +51,33 @@ public class StageManager
static
{
stageQueues.put(MUTATION_STAGE, new MultiThreadedStage(MUTATION_STAGE, getConcurrentWriters()));
stageQueues.put(READ_STAGE, new MultiThreadedStage(READ_STAGE, getConcurrentReaders()));
stageQueues.put(STREAM_STAGE, new SingleThreadedStage(STREAM_STAGE));
stageQueues.put(GOSSIP_STAGE, new SingleThreadedStage("GMFD"));
stageQueues.put(RESPONSE_STAGE, new MultiThreadedStage("RESPONSE-STAGE", MessagingService.MESSAGE_DESERIALIZE_THREADS));
stageQueues.put(AE_SERVICE_STAGE, new SingleThreadedStage(AE_SERVICE_STAGE));
stageQueues.put(LOADBALANCE_STAGE, new SingleThreadedStage(LOADBALANCE_STAGE));
stages.put(MUTATION_STAGE, multiThreadedStage(MUTATION_STAGE, getConcurrentWriters()));
stages.put(READ_STAGE, multiThreadedStage(READ_STAGE, getConcurrentReaders()));
stages.put(RESPONSE_STAGE, multiThreadedStage("RESPONSE-STAGE", MessagingService.MESSAGE_DESERIALIZE_THREADS));
// the rest are all single-threaded
stages.put(STREAM_STAGE, new JMXEnabledThreadPoolExecutor(STREAM_STAGE));
stages.put(GOSSIP_STAGE, new JMXEnabledThreadPoolExecutor("GMFD"));
stages.put(AE_SERVICE_STAGE, new JMXEnabledThreadPoolExecutor(AE_SERVICE_STAGE));
stages.put(LOADBALANCE_STAGE, new JMXEnabledThreadPoolExecutor(LOADBALANCE_STAGE));
}
private static ThreadPoolExecutor multiThreadedStage(String name, int numThreads)
{
return new JMXEnabledThreadPoolExecutor(numThreads,
numThreads,
Integer.MAX_VALUE,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory(name));
}
/**
* Retrieve a stage from the StageManager
* @param stageName name of the stage to be retrieved.
*/
public static IStage getStage(String stageName)
public static ThreadPoolExecutor getStage(String stageName)
{
return stageQueues.get(stageName);
return stages.get(stageName);
}
/**
@ -70,11 +85,10 @@ public class StageManager
*/
public static void shutdown()
{
Set<String> stages = stageQueues.keySet();
for ( String stage : stages )
Set<String> stages = StageManager.stages.keySet();
for (String stage : stages)
{
IStage registeredStage = stageQueues.get(stage);
registeredStage.shutdown();
StageManager.stages.get(stage).shutdown();
}
}
}

4
src/java/org/apache/cassandra/db/CommitLog.java
View File

@ -280,7 +280,7 @@ public class CommitLog
void recover(File[] clogs) throws IOException
{
Set<Table> tablesRecovered = new HashSet<Table>();
assert StageManager.getStage(StageManager.MUTATION_STAGE).getCompletedTasks() == 0;
assert StageManager.getStage(StageManager.MUTATION_STAGE).getCompletedTaskCount() == 0;
int rows = 0;
for (File file : clogs)
{
@ -363,7 +363,7 @@ public class CommitLog
}
// wait for all the writes to finish on the mutation stage
while (StageManager.getStage(StageManager.MUTATION_STAGE).getCompletedTasks() < rows)
while (StageManager.getStage(StageManager.MUTATION_STAGE).getCompletedTaskCount() < rows)
{
try
{

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

@ -493,7 +493,7 @@ public class MessagingService implements IFailureDetectionEventListener
private static void enqueueRunnable(String stageName, Runnable runnable){
IStage stage = StageManager.getStage(stageName);
ExecutorService stage = StageManager.getStage(stageName);
if ( stage != null )
{

2
src/java/org/apache/cassandra/service/AntiEntropyService.java
View File

@ -483,7 +483,7 @@ public class AntiEntropyService
for (MerkleTree.RowHash minrow : minrows)
range.addHash(minrow);
StageManager.getStage(StageManager.AE_SERVICE_STAGE).execute(this);
StageManager.getStage(StageManager.AE_SERVICE_STAGE).submit(this);
logger.debug("Validated " + validated + " rows into AEService tree for " + cf);
}

2
src/java/org/apache/cassandra/service/StorageLoadBalancer.java
View File

@ -25,8 +25,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.log4j.Logger;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.SingleThreadedStage;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.EndPointState;

2
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -496,7 +496,7 @@ public class StorageProxy implements StorageProxyMBean
for (ReadCommand command: commands)
{
Callable<Object> callable = new weakReadLocalCallable(command);
futures.add(StageManager.getStage(StageManager.READ_STAGE).execute(callable));
futures.add(StageManager.getStage(StageManager.READ_STAGE).submit(callable));
}
for (Future<Object> future : futures)
{

5
test/unit/org/apache/cassandra/service/AntiEntropyServiceTest.java
View File

@ -235,11 +235,12 @@ public class AntiEntropyServiceTest extends CleanupHelper
Future<Object> flushAES()
{
return StageManager.getStage(StageManager.AE_SERVICE_STAGE).execute(new Callable<Object>(){
return StageManager.getStage(StageManager.AE_SERVICE_STAGE).submit(new Callable<Object>()
{
public Boolean call()
{
return true;
}
});
});
}
}