Reactor uses Executor-based scheduler that does...

Reactor uses Executor-based scheduler that does not attempt to resize
thread pool.

4 files changed, 20 insertions, 20 deletions

diff --git a/src/actors/scala/actors/Actor.scala b/src/actors/scala/actors/Actor.scala
index 476bf85411..366141f33b 100644
--- a/src/actors/scala/actors/Actor.scala
+++ b/src/actors/scala/actors/Actor.scala
@@ -399,6 +399,8 @@ trait Actor extends AbstractActor with ReplyReactor with ReplyableActor {
    */
   private var onTimeout: Option[TimerTask] = None
 
+  protected[actors] override def scheduler: IScheduler = Scheduler
+
   private[actors] override def startSearch(msg: Any, replyTo: OutputChannel[Any], handler: Any => Boolean) =
     if (isSuspended) {
       () => synchronized {
diff --git a/src/actors/scala/actors/Reactor.scala b/src/actors/scala/actors/Reactor.scala
index 07a59a8c28..4a1f5ac051 100644
--- a/src/actors/scala/actors/Reactor.scala
+++ b/src/actors/scala/actors/Reactor.scala
@@ -10,8 +10,20 @@
 
 package scala.actors
 
+import scala.actors.scheduler.{DelegatingScheduler, DefaultThreadPoolScheduler}
 import scala.collection.mutable.Queue
 
+private object Reactor {
+  val scheduler = new DelegatingScheduler {
+    def makeNewScheduler: IScheduler = {
+      val s = new DefaultThreadPoolScheduler(false)
+      Debug.info(this+": starting new "+s+" ["+s.getClass+"]")
+      s.start()
+      s
+    }
+  }
+}
+
 /**
  * The Reactor trait provides lightweight actors.
  *
@@ -52,7 +64,7 @@ trait Reactor extends OutputChannel[Any] {
     Map()
 
   protected[actors] def scheduler: IScheduler =
-    Scheduler
+    Reactor.scheduler
 
   protected[actors] def mailboxSize: Int =
     mailbox.size
diff --git a/src/actors/scala/actors/Scheduler.scala b/src/actors/scala/actors/Scheduler.scala
index e50e848c73..7bc5eb20df 100644
--- a/src/actors/scala/actors/Scheduler.scala
+++ b/src/actors/scala/actors/Scheduler.scala
@@ -11,7 +11,7 @@
 package scala.actors
 
 import java.util.concurrent._
-import scheduler.{DelegatingScheduler, ThreadPoolConfig, ThreadPoolScheduler, ForkJoinScheduler, DefaultThreadPoolScheduler}
+import scheduler.{DelegatingScheduler, ForkJoinScheduler, DefaultThreadPoolScheduler}
 
 /**
  * The <code>Scheduler</code> object is used by <code>Actor</code> to
@@ -24,8 +24,8 @@ object Scheduler extends DelegatingScheduler {
   Debug.info("initializing "+this+"...")
 
   def makeNewScheduler: IScheduler = {
-    val s = new DefaultThreadPoolScheduler(false)
-    //val s = new ForkJoinScheduler
+    //val s = new DefaultThreadPoolScheduler(false)
+    val s = new ForkJoinScheduler
     Debug.info(this+": starting new "+s+" ["+s.getClass+"]")
     s.start()
     s
diff --git a/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala b/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
index e429e89274..08c842e71b 100644
--- a/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
+++ b/src/actors/scala/actors/scheduler/ThreadPoolScheduler.scala
@@ -29,13 +29,13 @@ import scala.concurrent.ManagedBlocker
  * @author Philipp Haller
  */
 class ThreadPoolScheduler(protected var executor: ThreadPoolExecutor,
-                          protected var terminate: Boolean,
+                          protected val terminate: Boolean,
                           protected val daemon: Boolean)
   extends Thread with ExecutorScheduler with TerminationMonitor {
 
   setDaemon(daemon)
 
-  private var terminating = false
+  private var terminating = false // guarded by this
   protected val CHECK_FREQ = 10
 
   /* This constructor (and the var above) is currently only used to work
@@ -47,18 +47,6 @@ class ThreadPoolScheduler(protected var executor: ThreadPoolExecutor,
     this(null, true, d)
   }
 
-  private def adjustCorePoolSize() {
-    val coreSize = executor.getCorePoolSize()
-    if (coreSize < ThreadPoolConfig.maxPoolSize && (executor.getActiveCount() >= coreSize - 1)) {
-      executor.setCorePoolSize(coreSize + 1)
-    }
-  }
-
-  override def managedBlock(blocker: ManagedBlocker) {
-    adjustCorePoolSize()
-    blocker.block()
-  }
-
   override def run() {
     try {
       while (true) {
@@ -73,8 +61,6 @@ class ThreadPoolScheduler(protected var executor: ThreadPoolExecutor,
             throw new QuitException
 
           gc()
-
-          adjustCorePoolSize()
         }
       }
     } catch {