From 92fc7b37b01bc871966a49524197c1cbbeec8988 Mon Sep 17 00:00:00 2001
From: Philipp Haller Reaction
.
- */
- override def terminated(a: Reactor) = synchronized {
- // execute registered termination handler (if any)
- termHandlers.get(a) match {
- case Some(handler) =>
- handler()
- // remove mapping
- termHandlers -= a
- case None =>
- // do nothing
- }
-
- // find the weak reference that points to the terminated actor, if any
- refSet.find((ref: Reference[t] forSome { type t <: Reactor }) => ref.get() == a) match {
- case Some(r) =>
- // invoking clear will not cause r to be enqueued
- r.clear()
- refSet -= r.asInstanceOf[Reference[t] forSome { type t <: Reactor }]
- case None =>
- // do nothing
+ override def terminated(a: Reactor) = {
+ super.terminated(a)
+
+ synchronized {
+ // find the weak reference that points to the terminated actor, if any
+ refSet.find((ref: Reference[t] forSome { type t <: Reactor }) => ref.get() == a) match {
+ case Some(r) =>
+ // invoking clear will not cause r to be enqueued
+ r.clear()
+ refSet -= r.asInstanceOf[Reference[t] forSome { type t <: Reactor }]
+ case None =>
+ // do nothing
+ }
}
-
- pendingReactions -= 1
}
private[actors] def getPendingCount = synchronized {
- pendingReactions
+ activeActors
}
private[actors] def setPendingCount(cnt: Int) = synchronized {
- pendingReactions = cnt
+ activeActors = cnt
}
}
diff --git a/src/actors/scala/actors/FJTaskScheduler2.scala b/src/actors/scala/actors/FJTaskScheduler2.scala
index 9a2832d85f..cf26b33916 100644
--- a/src/actors/scala/actors/FJTaskScheduler2.scala
+++ b/src/actors/scala/actors/FJTaskScheduler2.scala
@@ -48,20 +48,8 @@ class FJTaskScheduler2(val initCoreSize: Int, val maxSize: Int, daemon: Boolean)
private var submittedTasks = 0
- def printActorDump {}
-
private val CHECK_FREQ = 100
- def onLockup(handler: () => Unit) =
- lockupHandler = handler
-
- def onLockup(millis: Int)(handler: () => Unit) = {
- //LOCKUP_CHECK_FREQ = millis / CHECK_FREQ
- lockupHandler = handler
- }
-
- private var lockupHandler: () => Unit = null
-
private def allWorkersBlocked: Boolean =
executor.threads.forall(t => {
val s = t.getState()
diff --git a/src/actors/scala/actors/Future.scala b/src/actors/scala/actors/Future.scala
index 5ed758a65b..ec07f2a1a8 100644
--- a/src/actors/scala/actors/Future.scala
+++ b/src/actors/scala/actors/Future.scala
@@ -25,9 +25,9 @@ import scheduler.DefaultThreadPoolScheduler
*
Futures
object contains methods that operate on Futures.
*
- * @version 0.9.8
* @author Philipp Haller
*/
object Futures {
- private lazy val sched = new DefaultThreadPoolScheduler(true)
+ private case object Eval
def future[T](body: => T): Future[T] = {
- case object Eval
- val a = new Actor {
- override def scheduler: IScheduler = sched
+ val a = new DaemonActor {
def act() {
Actor.react {
case Eval => Actor.reply(body)
@@ -139,7 +136,7 @@ object Futures {
results
}
- def fromInputChannel[T](inputChannel: InputChannel[T]): Future[T] =
+ private[actors] def fromInputChannel[T](inputChannel: InputChannel[T]): Future[T] =
new Future[T](inputChannel) {
def apply() =
if (isSet) value.get.asInstanceOf[T]
diff --git a/src/actors/scala/actors/IScheduler.scala b/src/actors/scala/actors/IScheduler.scala
index f4034fbfab..8afed3aa86 100644
--- a/src/actors/scala/actors/IScheduler.scala
+++ b/src/actors/scala/actors/IScheduler.scala
@@ -67,4 +67,17 @@ trait IScheduler {
def onTerminate(a: Reactor)(f: => Unit): Unit
def managedBlock(blocker: scala.concurrent.ManagedBlocker): Unit
+
+ @deprecated("this member is going to be removed in a future release")
+ def tick(a: Actor) {}
+
+ @deprecated("this member is going to be removed in a future release")
+ def onLockup(handler: () => Unit) {}
+
+ @deprecated("this member is going to be removed in a future release")
+ def onLockup(millis: Int)(handler: () => Unit) {}
+
+ @deprecated("this member is going to be removed in a future release")
+ def printActorDump {}
+
}
diff --git a/src/actors/scala/actors/MessageQueue.scala b/src/actors/scala/actors/MessageQueue.scala
index 4435759264..540a992640 100644
--- a/src/actors/scala/actors/MessageQueue.scala
+++ b/src/actors/scala/actors/MessageQueue.scala
@@ -14,10 +14,10 @@ package scala.actors
* This class is used by our efficient message queue
* implementation.
*
- * @version 0.9.9
* @author Philipp Haller
*/
@serializable @SerialVersionUID(7124278808020037465L)
+@deprecated("this class is going to be removed in a future release")
class MessageQueueElement(val msg: Any, val session: OutputChannel[Any], var next: MessageQueueElement) {
def this() = this(null, null, null)
def this(msg: Any, session: OutputChannel[Any]) = this(msg, session, null)
@@ -29,10 +29,10 @@ class MessageQueueElement(val msg: Any, val session: OutputChannel[Any], var nex
* library. Classes in this package are supposed to be the only
* clients of this class.
*
- * @version 0.9.9
* @author Philipp Haller
*/
@serializable @SerialVersionUID(2168935872884095767L)
+@deprecated("this class is going to be removed in a future release")
class MessageQueue(protected val label: String) {
protected var first: MessageQueueElement = null
protected var last: MessageQueueElement = null // last eq null iff list is empty
diff --git a/src/actors/scala/actors/Reaction.scala b/src/actors/scala/actors/Reaction.scala
index 1c9541546a..80db4b2c1e 100644
--- a/src/actors/scala/actors/Reaction.scala
+++ b/src/actors/scala/actors/Reaction.scala
@@ -23,10 +23,10 @@ private[actors] class KillActorException extends Throwable with ControlException
* java.lang.Runnable
.
*
*
- * @version 0.9.10
+ * @deprecated("this class is going to be removed in a future release")
* @author Philipp Haller
*/
-private[actors] class Reaction(a: Actor, f: PartialFunction[Any, Unit], msg: Any) extends ActorTask(a, () => {
+class Reaction(a: Actor, f: PartialFunction[Any, Unit], msg: Any) extends ActorTask(a, () => {
if (f == null)
a.act()
else
diff --git a/src/actors/scala/actors/Reactor.scala b/src/actors/scala/actors/Reactor.scala
index 9c8a7a223e..c07612f1eb 100644
--- a/src/actors/scala/actors/Reactor.scala
+++ b/src/actors/scala/actors/Reactor.scala
@@ -20,14 +20,14 @@ import scala.collection.mutable.Queue
trait Reactor extends OutputChannel[Any] {
/* The actor's mailbox. */
- protected val mailbox = new MessageQueue("Reactor")
+ protected[this] val mailbox = new MessageQueue("Reactor")
- protected var sendBuffer = new Queue[(Any, OutputChannel[Any])]
+ protected[this] var sendBuffer = new Queue[(Any, OutputChannel[Any])]
/* If the actor waits in a react, continuation holds the
* message handler that react was called with.
*/
- protected var continuation: PartialFunction[Any, Unit] = null
+ protected[this] var continuation: PartialFunction[Any, Unit] = null
/* Whenever this Actor executes on some thread, waitingFor is
* guaranteed to be equal to waitingForNone.
@@ -36,8 +36,8 @@ trait Reactor extends OutputChannel[Any] {
* waitingForNone, this Actor is guaranteed not to execute on some
* thread.
*/
- protected val waitingForNone = (m: Any) => false
- protected var waitingFor: Any => Boolean = waitingForNone
+ protected[this] val waitingForNone = (m: Any) => false
+ protected[this] var waitingFor: Any => Boolean = waitingForNone
/**
* The behavior of an actor is specified by implementing this
@@ -147,7 +147,7 @@ trait Reactor extends OutputChannel[Any] {
/* This method is guaranteed to be executed from inside
an actors act method.
*/
- protected def scheduleActor(f: PartialFunction[Any, Unit], msg: Any) = {
+ protected[this] def scheduleActor(f: PartialFunction[Any, Unit], msg: Any) = {
scheduler executeFromActor (new LightReaction(this,
if (f eq null) continuation else f,
msg))
@@ -184,7 +184,7 @@ trait Reactor extends OutputChannel[Any] {
throw new KillActorException
}
- protected[actors] def exit(): Nothing = {
+ protected[this] def exit(): Nothing = {
terminated()
throw Actor.suspendException
}
diff --git a/src/actors/scala/actors/SchedulerAdapter.scala b/src/actors/scala/actors/SchedulerAdapter.scala
index ba7822c372..2c5ccd2dd8 100644
--- a/src/actors/scala/actors/SchedulerAdapter.scala
+++ b/src/actors/scala/actors/SchedulerAdapter.scala
@@ -15,7 +15,7 @@ package scala.actors
*
* Providing an implementation for the
* execute(f: => Unit)
method is sufficient to
- * obtain a concrete class that extends IScheduler
.
+ * obtain a concrete IScheduler
implementation.
*
* @author Philipp Haller
*/
diff --git a/src/actors/scala/actors/package.scala b/src/actors/scala/actors/package.scala
new file mode 100644
index 0000000000..d74fb23086
--- /dev/null
+++ b/src/actors/scala/actors/package.scala
@@ -0,0 +1,17 @@
+
+package object actors {
+ @deprecated("use scala.actors.scheduler.ForkJoinScheduler instead")
+ type FJTaskScheduler2 = scala.actors.scheduler.ForkJoinScheduler
+
+ @deprecated("use scala.actors.scheduler.ForkJoinScheduler instead")
+ type TickedScheduler = scala.actors.scheduler.ForkJoinScheduler
+
+ @deprecated("use scala.actors.scheduler.ForkJoinScheduler instead")
+ type WorkerThreadScheduler = scala.actors.scheduler.ForkJoinScheduler
+
+ @deprecated("this class is going to be removed in a future release")
+ type WorkerThread = java.lang.Thread
+
+ @deprecated("this value is going to be removed in a future release")
+ val ActorGC = scala.actors.Scheduler.impl.asInstanceOf[scala.actors.scheduler.ThreadPoolScheduler]
+}
diff --git a/src/actors/scala/actors/scheduler/DaemonScheduler.scala b/src/actors/scala/actors/scheduler/DaemonScheduler.scala
index d748e8fa75..d88842535c 100644
--- a/src/actors/scala/actors/scheduler/DaemonScheduler.scala
+++ b/src/actors/scala/actors/scheduler/DaemonScheduler.scala
@@ -16,6 +16,6 @@ package scheduler
*/
object DaemonScheduler extends DelegatingScheduler {
- def makeNewScheduler(): IScheduler = new DefaultExecutorScheduler(true)
+ def makeNewScheduler(): IScheduler = new DefaultThreadPoolScheduler(true)
}
diff --git a/src/actors/scala/actors/scheduler/DefaultExecutorScheduler.scala b/src/actors/scala/actors/scheduler/DefaultExecutorScheduler.scala
deleted file mode 100644
index 4a8f9d034d..0000000000
--- a/src/actors/scala/actors/scheduler/DefaultExecutorScheduler.scala
+++ /dev/null
@@ -1,53 +0,0 @@
-/* __ *\
-** ________ ___ / / ___ Scala API **
-** / __/ __// _ | / / / _ | (c) 2005-2009, LAMP/EPFL **
-** __\ \/ /__/ __ |/ /__/ __ | http://scala-lang.org/ **
-** /____/\___/_/ |_/____/_/ | | **
-** |/ **
-\* */
-
-// $Id$
-
-package scala.actors
-package scheduler
-
-import java.util.concurrent.{ThreadPoolExecutor, TimeUnit, LinkedBlockingQueue, ThreadFactory}
-
-/**
- * The DefaultExecutorScheduler
class uses a default
- * ThreadPoolExecutor
for executing Actor
s.
- *
- * It can be configured using the two JVM properties
- * actors.corePoolSize
and
- * actors.maxPoolSize
that control the initial and
- * maximum size of the thread pool, respectively.
- *
- * @author Philipp Haller
- */
-class DefaultExecutorScheduler(daemon: Boolean)
- extends SchedulerService(daemon) with ExecutorScheduler {
-
- def this() =
- this(false)
-
- private val workQueue = new LinkedBlockingQueue[Runnable]
-
- private val threadFactory = new ThreadFactory {
- def newThread(r: Runnable): Thread = {
- val result = new Thread(r)
- result.setDaemon(daemon)
- result
- }
- }
-
- private val threadPool = new ThreadPoolExecutor(ThreadPoolConfig.corePoolSize,
- ThreadPoolConfig.maxPoolSize,
- 60000L,
- TimeUnit.MILLISECONDS,
- workQueue,
- threadFactory)
-
- val executor = threadPool
-
- override val CHECK_FREQ = 10
-}
diff --git a/src/actors/scala/actors/scheduler/DelegatingScheduler.scala b/src/actors/scala/actors/scheduler/DelegatingScheduler.scala
index e72acd09d8..b25e3d26f5 100644
--- a/src/actors/scala/actors/scheduler/DelegatingScheduler.scala
+++ b/src/actors/scala/actors/scheduler/DelegatingScheduler.scala
@@ -14,7 +14,7 @@ import scala.concurrent.ManagedBlocker
/**
* @author Erik Engbrecht
*/
-trait DelegatingScheduler extends IScheduler {
+private[actors] trait DelegatingScheduler extends IScheduler {
protected def makeNewScheduler(): IScheduler
protected var sched: IScheduler = null
diff --git a/src/actors/scala/actors/scheduler/ExecutorScheduler.scala b/src/actors/scala/actors/scheduler/ExecutorScheduler.scala
index b1c0da256c..52ec977b1f 100644
--- a/src/actors/scala/actors/scheduler/ExecutorScheduler.scala
+++ b/src/actors/scala/actors/scheduler/ExecutorScheduler.scala
@@ -20,7 +20,7 @@ import scala.concurrent.ThreadPoolRunner
*
* @author Philipp Haller
*/
-trait ExecutorScheduler extends IScheduler with ThreadPoolRunner {
+private[scheduler] trait ExecutorScheduler extends IScheduler with ThreadPoolRunner {
def execute(task: Runnable) {
super[ThreadPoolRunner].execute(task.asInstanceOf[Task[Unit]])
diff --git a/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala b/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala
index b7f68be3b3..bbcbfa90f4 100644
--- a/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala
+++ b/src/actors/scala/actors/scheduler/ForkJoinScheduler.scala
@@ -10,19 +10,32 @@ import scala.concurrent.forkjoin._
*
* @author Philipp Haller
*/
-class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor {
+class ForkJoinScheduler(val initCoreSize: Int, val maxSize: Int, daemon: Boolean) extends Runnable with IScheduler with TerminationMonitor {
private var pool = makeNewPool()
private var terminating = false
private var snapshoting = false
+
+ // this has to be a java.util.Collection, since this is what
+ // the ForkJoinPool returns.
private var drainedTasks: Collection[ForkJoinTask[_]] = null
- private val CHECK_FREQ = 10
+ protected val CHECK_FREQ = 10
+
+ def this(d: Boolean) {
+ this(ThreadPoolConfig.corePoolSize, ThreadPoolConfig.maxPoolSize, d)
+ }
+
+ def this() {
+ this(false)
+ }
private def makeNewPool(): DrainableForkJoinPool = {
val p = new DrainableForkJoinPool()
// enable locally FIFO scheduling mode
p.setAsyncMode(true)
+ p.setParallelism(initCoreSize)
+ p.setMaximumPoolSize(maxSize)
Debug.info(this+": parallelism "+p.getParallelism())
Debug.info(this+": max pool size "+p.getMaximumPoolSize())
p
@@ -31,7 +44,15 @@ class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor
/** Starts this scheduler.
*/
def start() {
- (new Thread(this)).start()
+ try {
+ val t = new Thread(this)
+ t.setDaemon(daemon)
+ t.setName("ForkJoinScheduler")
+ t.start()
+ } catch {
+ case e: Exception =>
+ Debug.info(this+": could not create scheduler thread: "+e)
+ }
}
private def allWorkersBlocked: Boolean =
@@ -62,6 +83,9 @@ class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor
}
if (!snapshoting) {
+ gc()
+
+ // check if we need more threads to avoid deadlock
val poolSize = pool.getPoolSize()
if (allWorkersBlocked && (poolSize < ThreadPoolConfig.maxPoolSize)) {
pool.setParallelism(poolSize + 1)
@@ -90,6 +114,7 @@ class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor
}
}
+ // TODO: when do we pass a task that is not a RecursiveAction?
def execute(task: Runnable) {
pool.execute(task)
}
@@ -111,13 +136,6 @@ class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor
def run() { fun }
})
- override def managedBlock(blocker: scala.concurrent.ManagedBlocker) {
- ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker {
- def block = blocker.block()
- def isReleasable() = blocker.isReleasable
- }, true)
- }
-
/** Shuts down the scheduler.
*/
def shutdown(): Unit = synchronized {
@@ -127,6 +145,13 @@ class ForkJoinScheduler extends Runnable with IScheduler with TerminationMonitor
def isActive =
(pool ne null) && !pool.isShutdown()
+ override def managedBlock(blocker: scala.concurrent.ManagedBlocker) {
+ ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker {
+ def block = blocker.block()
+ def isReleasable() = blocker.isReleasable
+ }, true)
+ }
+
/** Suspends the scheduler. All threads that were in use by the
* scheduler and its internal thread pool are terminated.
*/
diff --git a/src/actors/scala/actors/scheduler/SchedulerService.scala b/src/actors/scala/actors/scheduler/SchedulerService.scala
index 16937f52ea..37cfa5ca59 100644
--- a/src/actors/scala/actors/scheduler/SchedulerService.scala
+++ b/src/actors/scala/actors/scheduler/SchedulerService.scala
@@ -32,20 +32,8 @@ abstract class SchedulerService(daemon: Boolean) extends Thread with IScheduler
private var terminating = false
- def printActorDump {}
-
protected val CHECK_FREQ = 100
- def onLockup(handler: () => Unit) =
- lockupHandler = handler
-
- def onLockup(millis: Int)(handler: () => Unit) = {
- //LOCKUP_CHECK_FREQ = millis / CHECK_FREQ
- lockupHandler = handler
- }
-
- private var lockupHandler: () => Unit = null
-
def onShutdown(): Unit
override def run() {
diff --git a/src/actors/scala/actors/scheduler/TerminationMonitor.scala b/src/actors/scala/actors/scheduler/TerminationMonitor.scala
index d26d8e4e3d..875b79a8c1 100644
--- a/src/actors/scala/actors/scheduler/TerminationMonitor.scala
+++ b/src/actors/scala/actors/scheduler/TerminationMonitor.scala
@@ -15,13 +15,13 @@ import scala.collection.mutable.HashMap
trait TerminationMonitor {
- protected var pendingReactions = 0
- protected val termHandlers = new HashMap[Reactor, () => Unit]
+ protected var activeActors = 0
+ protected val terminationHandlers = new HashMap[Reactor, () => Unit]
private var started = false
/** newActor is invoked whenever a new actor is started. */
def newActor(a: Reactor) = synchronized {
- pendingReactions += 1
+ activeActors += 1
if (!started)
started = true
}
@@ -33,16 +33,20 @@ trait TerminationMonitor {
* @param f the closure to be registered
*/
def onTerminate(a: Reactor)(f: => Unit): Unit = synchronized {
- termHandlers += (a -> (() => f))
+ terminationHandlers += (a -> (() => f))
}
- def terminated(a: Reactor) = synchronized {
+ /** Registers that the specified actor has terminated.
+ *
+ * @param a the actor that has terminated
+ */
+ def terminated(a: Reactor) = {
// obtain termination handler (if any)
val todo = synchronized {
- termHandlers.get(a) match {
+ terminationHandlers.get(a) match {
case Some(handler) =>
- termHandlers -= a
- () => handler
+ terminationHandlers -= a
+ handler
case None =>
() => { /* do nothing */ }
}
@@ -52,13 +56,16 @@ trait TerminationMonitor {
todo()
synchronized {
- pendingReactions -= 1
+ activeActors -= 1
}
}
- protected def allTerminated: Boolean = synchronized {
- started && pendingReactions <= 0
+ /** Checks whether all actors have terminated. */
+ @deprecated("this method is going to be removed in a future release")
+ def allTerminated: Boolean = synchronized {
+ started && activeActors <= 0
}
+ /** Checks for actors that have become garbage. */
protected def gc() {}
}
diff --git a/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala b/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
index 1a68dc669b..e429e89274 100644
--- a/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
+++ b/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
@@ -47,7 +47,7 @@ class ThreadPoolScheduler(protected var executor: ThreadPoolExecutor,
this(null, true, d)
}
- protected def adjustCorePoolSize() {
+ private def adjustCorePoolSize() {
val coreSize = executor.getCorePoolSize()
if (coreSize < ThreadPoolConfig.maxPoolSize && (executor.getActiveCount() >= coreSize - 1)) {
executor.setCorePoolSize(coreSize + 1)
--
cgit v1.2.3