Added a root akka folder for source files for the docs to work properly, closing ticket #541
This commit is contained in:
Viktor Klang
parent
8ecba6d740
commit
7040ef0f14
175 changed files with 10 additions and 4 deletions
264
akka-actor/src/main/scala/akka/dispatch/ThreadPoolBuilder.scala
Normal file
264
akka-actor/src/main/scala/akka/dispatch/ThreadPoolBuilder.scala
Normal file
|
|
@ -0,0 +1,264 @@
|
|||
/**
|
||||
* Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
|
||||
*/
|
||||
|
||||
package akka.dispatch
|
||||
|
||||
import java.util.Collection
|
||||
import java.util.concurrent._
|
||||
import atomic.{AtomicLong, AtomicInteger}
|
||||
import ThreadPoolExecutor.CallerRunsPolicy
|
||||
|
||||
import akka.util. {Duration, Logging}
|
||||
|
||||
object ThreadPoolConfig {
|
||||
type Bounds = Int
|
||||
type FlowHandler = Either[RejectedExecutionHandler,Bounds]
|
||||
type QueueFactory = () => BlockingQueue[Runnable]
|
||||
|
||||
val defaultAllowCoreThreadTimeout: Boolean = false
|
||||
val defaultCorePoolSize: Int = 16
|
||||
val defaultMaxPoolSize: Int = 128
|
||||
val defaultTimeout: Duration = Duration(60000L,TimeUnit.MILLISECONDS)
|
||||
def defaultFlowHandler: FlowHandler = flowHandler(new CallerRunsPolicy)
|
||||
|
||||
def flowHandler(rejectionHandler: RejectedExecutionHandler): FlowHandler = Left(rejectionHandler)
|
||||
def flowHandler(bounds: Int): FlowHandler = Right(bounds)
|
||||
|
||||
def fixedPoolSize(size: Int): Int = size
|
||||
def scaledPoolSize(multiplier: Double): Int =
|
||||
(Runtime.getRuntime.availableProcessors * multiplier).ceil.toInt
|
||||
|
||||
def arrayBlockingQueue(capacity: Int, fair: Boolean): QueueFactory =
|
||||
() => new ArrayBlockingQueue[Runnable](capacity,fair)
|
||||
|
||||
def synchronousQueue(fair: Boolean): QueueFactory =
|
||||
() => new SynchronousQueue[Runnable](fair)
|
||||
|
||||
def linkedBlockingQueue(): QueueFactory =
|
||||
() => new LinkedBlockingQueue[Runnable]()
|
||||
|
||||
def linkedBlockingQueue(capacity: Int): QueueFactory =
|
||||
() => new LinkedBlockingQueue[Runnable](capacity)
|
||||
|
||||
def reusableQueue(queue: BlockingQueue[Runnable]): QueueFactory =
|
||||
() => queue
|
||||
|
||||
def reusableQueue(queueFactory: QueueFactory): QueueFactory = {
|
||||
val queue = queueFactory()
|
||||
() => queue
|
||||
}
|
||||
}
|
||||
|
||||
case class ThreadPoolConfig(allowCorePoolTimeout: Boolean = ThreadPoolConfig.defaultAllowCoreThreadTimeout,
|
||||
corePoolSize: Int = ThreadPoolConfig.defaultCorePoolSize,
|
||||
maxPoolSize: Int = ThreadPoolConfig.defaultMaxPoolSize,
|
||||
threadTimeout: Duration = ThreadPoolConfig.defaultTimeout,
|
||||
flowHandler: ThreadPoolConfig.FlowHandler = ThreadPoolConfig.defaultFlowHandler,
|
||||
queueFactory: ThreadPoolConfig.QueueFactory = ThreadPoolConfig.linkedBlockingQueue()) {
|
||||
|
||||
final def createLazyExecutorService(threadFactory: ThreadFactory): ExecutorService =
|
||||
new LazyExecutorServiceWrapper(createExecutorService(threadFactory))
|
||||
|
||||
final def createExecutorService(threadFactory: ThreadFactory): ExecutorService = {
|
||||
flowHandler match {
|
||||
case Left(rejectHandler) =>
|
||||
val service = new ThreadPoolExecutor(corePoolSize, maxPoolSize, threadTimeout.length, threadTimeout.unit, queueFactory(), threadFactory, rejectHandler)
|
||||
service.allowCoreThreadTimeOut(allowCorePoolTimeout)
|
||||
service
|
||||
case Right(bounds) =>
|
||||
val service = new ThreadPoolExecutor(corePoolSize, maxPoolSize, threadTimeout.length, threadTimeout.unit, queueFactory(), threadFactory)
|
||||
service.allowCoreThreadTimeOut(allowCorePoolTimeout)
|
||||
new BoundedExecutorDecorator(service,bounds)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
trait DispatcherBuilder {
|
||||
def build: MessageDispatcher
|
||||
}
|
||||
|
||||
object ThreadPoolConfigDispatcherBuilder {
|
||||
def conf_?[T](opt: Option[T])(fun: (T) => ThreadPoolConfigDispatcherBuilder => ThreadPoolConfigDispatcherBuilder):
|
||||
Option[(ThreadPoolConfigDispatcherBuilder) => ThreadPoolConfigDispatcherBuilder] = opt map fun
|
||||
}
|
||||
|
||||
case class ThreadPoolConfigDispatcherBuilder(dispatcherFactory: (ThreadPoolConfig) => MessageDispatcher, config: ThreadPoolConfig) extends DispatcherBuilder {
|
||||
import ThreadPoolConfig._
|
||||
def build = dispatcherFactory(config)
|
||||
|
||||
//TODO remove this, for backwards compat only
|
||||
def buildThreadPool = build
|
||||
|
||||
def withNewBoundedThreadPoolWithLinkedBlockingQueueWithUnboundedCapacity(bounds: Int): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(flowHandler = flowHandler(bounds), queueFactory = linkedBlockingQueue()))
|
||||
|
||||
def withNewThreadPoolWithCustomBlockingQueue(newQueueFactory: QueueFactory): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(flowHandler = defaultFlowHandler, queueFactory = newQueueFactory))
|
||||
|
||||
def withNewThreadPoolWithCustomBlockingQueue(queue: BlockingQueue[Runnable]): ThreadPoolConfigDispatcherBuilder =
|
||||
withNewThreadPoolWithCustomBlockingQueue(reusableQueue(queue))
|
||||
|
||||
def withNewThreadPoolWithLinkedBlockingQueueWithUnboundedCapacity: ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(queueFactory = linkedBlockingQueue(), flowHandler = defaultFlowHandler))
|
||||
|
||||
def withNewThreadPoolWithLinkedBlockingQueueWithCapacity(capacity: Int): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(queueFactory = linkedBlockingQueue(capacity), flowHandler = defaultFlowHandler))
|
||||
|
||||
def withNewThreadPoolWithSynchronousQueueWithFairness(fair: Boolean): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(queueFactory = synchronousQueue(fair), flowHandler = defaultFlowHandler))
|
||||
|
||||
def withNewThreadPoolWithArrayBlockingQueueWithCapacityAndFairness(capacity: Int, fair: Boolean): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(queueFactory = arrayBlockingQueue(capacity,fair), flowHandler = defaultFlowHandler))
|
||||
|
||||
def setCorePoolSize(size: Int): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(corePoolSize = size))
|
||||
|
||||
def setMaxPoolSize(size: Int): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(maxPoolSize = size))
|
||||
|
||||
def setCorePoolSizeFromFactor(multiplier: Double): ThreadPoolConfigDispatcherBuilder =
|
||||
setCorePoolSize(scaledPoolSize(multiplier))
|
||||
|
||||
def setMaxPoolSizeFromFactor(multiplier: Double): ThreadPoolConfigDispatcherBuilder =
|
||||
setMaxPoolSize(scaledPoolSize(multiplier))
|
||||
|
||||
def setExecutorBounds(bounds: Int): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(flowHandler = flowHandler(bounds)))
|
||||
|
||||
def setKeepAliveTimeInMillis(time: Long): ThreadPoolConfigDispatcherBuilder =
|
||||
setKeepAliveTime(Duration(time,TimeUnit.MILLISECONDS))
|
||||
|
||||
def setKeepAliveTime(time: Duration): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(threadTimeout = time))
|
||||
|
||||
def setRejectionPolicy(policy: RejectedExecutionHandler): ThreadPoolConfigDispatcherBuilder =
|
||||
setFlowHandler(flowHandler(policy))
|
||||
|
||||
def setFlowHandler(newFlowHandler: FlowHandler): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(flowHandler = newFlowHandler))
|
||||
|
||||
def setAllowCoreThreadTimeout(allow: Boolean): ThreadPoolConfigDispatcherBuilder =
|
||||
this.copy(config = config.copy(allowCorePoolTimeout = allow))
|
||||
|
||||
def configure(fs: Option[Function[ThreadPoolConfigDispatcherBuilder,ThreadPoolConfigDispatcherBuilder]]*):
|
||||
ThreadPoolConfigDispatcherBuilder = fs.foldLeft(this)( (c,f) => f.map( _(c) ).getOrElse(c))
|
||||
}
|
||||
|
||||
/**
|
||||
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
||||
*/
|
||||
class MonitorableThreadFactory(val name: String) extends ThreadFactory {
|
||||
protected val counter = new AtomicLong
|
||||
|
||||
def newThread(runnable: Runnable) = new MonitorableThread(runnable, name)
|
||||
}
|
||||
|
||||
/**
|
||||
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
||||
*/
|
||||
object MonitorableThread {
|
||||
val DEFAULT_NAME = "MonitorableThread"
|
||||
val created = new AtomicInteger
|
||||
val alive = new AtomicInteger
|
||||
@volatile var debugLifecycle = false
|
||||
}
|
||||
|
||||
// FIXME fix the issues with using the monitoring in MonitorableThread
|
||||
|
||||
/**
|
||||
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
||||
*/
|
||||
class MonitorableThread(runnable: Runnable, name: String)
|
||||
extends Thread(runnable, name + "-" + MonitorableThread.created.incrementAndGet) with Logging {
|
||||
|
||||
setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
|
||||
def uncaughtException(thread: Thread, cause: Throwable) =
|
||||
log.error(cause, "UNCAUGHT in thread [%s]", thread.getName)
|
||||
})
|
||||
|
||||
override def run = {
|
||||
val debug = MonitorableThread.debugLifecycle
|
||||
log.debug("Created thread %s", getName)
|
||||
try {
|
||||
MonitorableThread.alive.incrementAndGet
|
||||
super.run
|
||||
} finally {
|
||||
MonitorableThread.alive.decrementAndGet
|
||||
log.debug("Exiting thread %s", getName)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @author <a href="http://jonasboner.com">Jonas Bonér</a>
|
||||
*/
|
||||
class BoundedExecutorDecorator(val executor: ExecutorService, bound: Int) extends ExecutorServiceDelegate {
|
||||
protected val semaphore = new Semaphore(bound)
|
||||
|
||||
override def execute(command: Runnable) = {
|
||||
semaphore.acquire
|
||||
try {
|
||||
executor.execute(new Runnable() {
|
||||
def run = {
|
||||
try {
|
||||
command.run
|
||||
} finally {
|
||||
semaphore.release
|
||||
}
|
||||
}
|
||||
})
|
||||
} catch {
|
||||
case e: RejectedExecutionException =>
|
||||
semaphore.release
|
||||
case e =>
|
||||
log.error(e,"Unexpected exception")
|
||||
throw e
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
trait ExecutorServiceDelegate extends ExecutorService with Logging {
|
||||
|
||||
def executor: ExecutorService
|
||||
|
||||
def execute(command: Runnable) = executor.execute(command)
|
||||
|
||||
def shutdown = executor.shutdown
|
||||
|
||||
def shutdownNow = executor.shutdownNow
|
||||
|
||||
def isShutdown = executor.isShutdown
|
||||
|
||||
def isTerminated = executor.isTerminated
|
||||
|
||||
def awaitTermination(l: Long, timeUnit: TimeUnit) = executor.awaitTermination(l, timeUnit)
|
||||
|
||||
def submit[T](callable: Callable[T]) = executor.submit(callable)
|
||||
|
||||
def submit[T](runnable: Runnable, t: T) = executor.submit(runnable, t)
|
||||
|
||||
def submit(runnable: Runnable) = executor.submit(runnable)
|
||||
|
||||
def invokeAll[T](callables: Collection[_ <: Callable[T]]) = executor.invokeAll(callables)
|
||||
|
||||
def invokeAll[T](callables: Collection[_ <: Callable[T]], l: Long, timeUnit: TimeUnit) = executor.invokeAll(callables, l, timeUnit)
|
||||
|
||||
def invokeAny[T](callables: Collection[_ <: Callable[T]]) = executor.invokeAny(callables)
|
||||
|
||||
def invokeAny[T](callables: Collection[_ <: Callable[T]], l: Long, timeUnit: TimeUnit) = executor.invokeAny(callables, l, timeUnit)
|
||||
}
|
||||
|
||||
trait LazyExecutorService extends ExecutorServiceDelegate {
|
||||
|
||||
def createExecutor: ExecutorService
|
||||
|
||||
lazy val executor = {
|
||||
log.info("Lazily initializing ExecutorService for ",this)
|
||||
createExecutor
|
||||
}
|
||||
}
|
||||
|
||||
class LazyExecutorServiceWrapper(executorFactory: => ExecutorService) extends LazyExecutorService {
|
||||
def createExecutor = executorFactory
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue