Actors mem leak or I'm doing something wrong?

If one actor is sending message too fast to another actor it's pretty easy to create and OOME by simply filling up the actor's mailbox faster than it can process the messages.

It might work to use a loop instead of a while(true) in sender.  I usually have the sender pause every once in a while and wait for a pingpack of some sort from the receiver.  That way I know it won't send too many messages before they are processed, as opposed to relying on luck that the actors run in sync with one another.

On Mon, Mar 16, 2009 at 6:16 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

On Scala 2.7.3 this code quickly runs out of memory:

import scala.actors.Actor._

object Test {

 val rt = Runtime.getRuntime()

 val sender = actor {
   while(true) {
     receiver ! new Array[Int] (1048576)
     System.gc()
     println ("Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
1048576.) formatted "%.2f") + " Mb")
   }
 }

 val receiver = actor {
   loop {
     react {
       case x: Array[Int] => println ("received ["+x.length+"]")
     }
   }
 }

 def main (args: Array[String]) {
   sender
 }

}

Where is the leak?
--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22548629.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.

If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

I've just run the code where the sender blocks until the receiver receives the message (so there's at most 1 element in the mailbox) and still see the OOME.

On Mon, Mar 16, 2009 at 3:50 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

Really?  Can you think of why?

I'll try to take a look when I get off of work...

On Mon, Mar 16, 2009 at 6:51 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

I've just run the code where the sender blocks until the receiver receives the message (so there's at most 1 element in the mailbox) and still see the OOME.

On Mon, Mar 16, 2009 at 3:50 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

On Mon, Mar 16, 2009 at 3:55 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Really?  Can you think of why?

It seems to happen when the message is sent in the context of an Actor.

The following works just fine:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  def sender = {
    while(true) {
      receiver ! new Array[Int] (1048576)
      cnt += 1
      System.gc()
      println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                        1048576.) formatted "%.2f") + " Mb")
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender
  }

}

But this gets the OOME:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  val sender = actor {
    loop {
      react {
    case _ =>
      while(true) {
        receiver ! new Array[Int] (1048576)
        cnt += 1
        System.gc()
        println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                          1048576.) formatted "%.2f") + " Mb")
      }
      }
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender ! "Start"
  }

}


 

I'll try to take a look when I get off of work...

On Mon, Mar 16, 2009 at 6:51 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

I've just run the code where the sender blocks until the receiver receives the message (so there's at most 1 element in the mailbox) and still see the OOME.

On Mon, Mar 16, 2009 at 3:50 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

I ran the original (with a message cnt added and the System.gc() call removed*) with memory profiling enabled and took a look at the resulting heap dump.  The arrays of integers are being collected - just not fast enough.  It will run a little longer if you use -XX:+UseConcMarkSweepGC.  Both with and without -XX:+UseConcMarkSweepGC specified the heap contains 18 big int arrays, but with -XX:+UseConcMarkSweepGC it processes ~28 messages and without it processes ~22.
I've seen this before and with some laborious GC tuning you should be able to get around it.  My theory is that when you have multiple threads producing garbage at a high rate it is easy for them to overrun the GC.  Might I suggest:
http://java.sun.com/performance/reference/whitepapers/tuning.html
*removing the call to System.gc() increased the number of messages processed when using -XX:+UseConcMarkSweepGC and did not change it with no explicit GC specified.
On Mon, Mar 16, 2009 at 6:58 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

On Mon, Mar 16, 2009 at 3:55 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Really?  Can you think of why?

It seems to happen when the message is sent in the context of an Actor.

The following works just fine:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  def sender = {
    while(true) {
      receiver ! new Array[Int] (1048576)
      cnt += 1
      System.gc()
      println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                        1048576.) formatted "%.2f") + " Mb")
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender
  }

}

But this gets the OOME:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  val sender = actor {
    loop {
      react {
    case _ =>
      while(true) {
        receiver ! new Array[Int] (1048576)
        cnt += 1
        System.gc()
        println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                          1048576.) formatted "%.2f") + " Mb")
      }
      }
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender ! "Start"
  }

}


 

I'll try to take a look when I get off of work...

On Mon, Mar 16, 2009 at 6:51 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

I've just run the code where the sender blocks until the receiver receives the message (so there's at most 1 element in the mailbox) and still see the OOME.

On Mon, Mar 16, 2009 at 3:50 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

There does seem to be some weirdness going on...I'll look into this more this weekend.

On Mon, Mar 16, 2009 at 6:58 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

On Mon, Mar 16, 2009 at 3:55 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Really?  Can you think of why?

It seems to happen when the message is sent in the context of an Actor.

The following works just fine:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  def sender = {
    while(true) {
      receiver ! new Array[Int] (1048576)
      cnt += 1
      System.gc()
      println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                        1048576.) formatted "%.2f") + " Mb")
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender
  }

}

But this gets the OOME:

import scala.actors.Actor._

object Test {
  var cnt = 0

  val rt = Runtime.getRuntime()

  val sender = actor {
    loop {
      react {
    case _ =>
      while(true) {
        receiver ! new Array[Int] (1048576)
        cnt += 1
        System.gc()
        println ("Cnt "+cnt+" Used Mem: " + (((rt.totalMemory() - rt.freeMemory()) /
                          1048576.) formatted "%.2f") + " Mb")
      }
      }
    }
  }

  val receiver = actor {
    loop {
      react {
    case x: Array[Int] =>
      println ("received ["+x.length+"]")
      }
    }
  }

  def main (args: Array[String]) {
    sender ! "Start"
  }

}


 

I'll try to take a look when I get off of work...

On Mon, Mar 16, 2009 at 6:51 PM, David Pollak <feeder.of.the.bears@gmail.com> wrote:

I've just run the code where the sender blocks until the receiver receives the message (so there's at most 1 element in the mailbox) and still see the OOME.

On Mon, Mar 16, 2009 at 3:50 PM, Erik Engbrecht <erik.engbrecht@gmail.com> wrote:

Have you run the code in a memory profiler and verified that when then OOME occurs the mailbox for the receiver has almost no elements in it?

while(true) receive { ... } "works" because it is an order of magnitude faster than loop { react { ... } } and having both threads contending for locks on the receiver actor's mailbox will somewhat keep them in sync with one another.

On Mon, Mar 16, 2009 at 6:41 PM, Alex Yakovlev <alex14n@gmail.com> wrote:

This is not a problem. Anyway, System.gc() is called each iteration.

Same code works without any leaks with while(true) receive { ... } instead
of react,
so I think the problem is that loop { receive { ... } } does not release its
links to argument.


If one actor is sending message too fast to another actor it's pretty easy
to create and OOME by simply filling up the actor's mailbox faster than it
can process the messages.

--
View this message in context: http://www.nabble.com/Actors-mem-leak-or-I%27m-doing-something-wrong--tp22548629p22549024.html
Sent from the Scala mailing list archive at Nabble.com.

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

--
Lift, the simply functional web framework http://liftweb.net
Beginning Scala http://www.apress.com/book/view/1430219890
Follow me: http://twitter.com/dpp
Git some: http://github.com/dpp

--
http://erikengbrecht.blogspot.com/

If I comment FJTaskRunner stuff in scala.actors.Scheduler.execute and leave
only
sched execute { fun }
it works without leak.

There does seem to be some weirdness going on...I'll look into this more
this weekend.

Alex Yakovlev wrote:
> If I comment FJTaskRunner stuff in scala.actors.Scheduler.execute and leave
> only
> sched execute { fun }
> it works without leak.

Good to know!
I'll try to look into it next week from Tuesday (important deadline
before that), but maybe you can give some more info as to what you
commented?

Thanks,
Philipp

I've fired a ticket at the official Scala Trac:
http://lampsvn.epfl.ch/trac/scala/ticket/1801

First:

object Scheduler extends IScheduler {
...
def execute(task: Runnable) {
/*
val t = currentThread
if (t.isInstanceOf[FJTaskRunner]) {
val tr = t.asInstanceOf[FJTaskRunner]
tr.push(new FJTask {
def run() { task.run() }
})
} else
*/
sched execute task
}

def execute(fun: => Unit) {
/*
val t = currentThread
if (t.isInstanceOf[FJTaskRunner]) {
val tr = t.asInstanceOf[FJTaskRunner]
tr.push(new FJTask {
def run() { fun }
})
} else
*/
sched execute { fun }
}
...
}

Second:

class Reaction extends Runnable {
...
def run() {
...
} finally {
Actor.tl.set(saved)
// Add:
this.a = null
this.f = null
this.msg = null
}
}
...
}

Second patch ensures that even if FJTaskRunner will hold 'leaked' references
to FJTask/Reaction, they will hold no references to (possibly huge) user's
closures/actors due to get() instead of take() in FJTaskRunner. I think it
can happen if we create a lot of actors that will fill FJTaskRunner's array
of VolatileTaskRef's but I have not written a test suite to (re)produce it.

Philipp Haller-2 wrote:
>
> Alex Yakovlev wrote:
>
> Good to know!
> I'll try to look into it next week from Tuesday (important deadline
> before that), but maybe you can give some more info as to what you
> commented?
>
>

Philipp,

alternative solution is to null Reaction's references and call
FJTaskRunner.put instead of FJTaskRunner.push from Scheduler. Why is there a
direct call to current thread's FJTaskRunner? Doest it give performance
increase?

I've rested a little more and it appears that just nulling Reaction's
references is enough. push adds new tasks to the end of array, and they are
taken from the beginning, so first several thousand iteration it is filled
(about 500-600 kilobytes), and after that there's no more memory increases.
Is there a reason to call FJTaskRunner.push instead or put? With put there
will be no array sliding.

Anyway, just nulling Reaction's fields at the end of run() method is
absolutely enough to fix this.

Regards,
Alex

Philipp Haller-2 wrote:
>
>
> Good to know!
> I'll try to look into it next week from Tuesday (important deadline
> before that), but maybe you can give some more info as to what you
> commented?
>
>

Hi Alex,

Many thanks for your excellent diagnosis!

> alternative solution is to null Reaction's references and call
> FJTaskRunner.put instead of FJTaskRunner.push from Scheduler. Why is there a
> direct call to current thread's FJTaskRunner? Doest it give performance
> increase?

Yes, the direct calls to the FJTaskRunner give a noticeable performance
increase (off-hand I can't give you numbers, though).

> I've rested a little more and it appears that just nulling Reaction's
> references is enough. push adds new tasks to the end of array, and they are
> taken from the beginning, so first several thousand iteration it is filled
> (about 500-600 kilobytes), and after that there's no more memory increases.
> Is there a reason to call FJTaskRunner.push instead or put? With put there
> will be no array sliding.

Push is faster than put, since in most cases it doesn't require grabbing
a lock, while it can run concurrently with stealing; OTOH, put always
requires grabbing a lock. Also, push has the "right" semantics in the
sense that a worker thread executes tasks in its own queue in submission
order.

> Anyway, just nulling Reaction's fields at the end of run() method is
> absolutely enough to fix this.

Great, will be fixed very soon, then.

Best regards,
Philipp

Hi Philipp,

performance-wise, have you considered inheriting Reaction from FJTask
instead of wrapping it? I mean:

class Reaction extends FJTask {
...
}

and in Scheduler.scala:

def execute(task: Runnable) {
currentThread match {
case tr: FJTaskRunner =>
tr.push(task match {
case fjt: FJTask => fjt
case _ => new FJTask {
def run() { task.run() }
}})
case _ =>
sched execute task
}
}

match here works a bit faster than if(isInstanceOf)asInstanceOf because of
eliminating tr variable. Or even modifying Scheduler to handle only FJTasks
instead of Runnables?

diff for Reaction.scala:

51c51
< class Reaction extends FJTask {
---
> class Reaction extends Runnable {

diff for Scheduler.scala:

74,78c74,81
< def execute(task: FJTask) {
< currentThread match {
< case tr: FJTaskRunner =>
< tr.push(task)
< case _ =>
---
> def execute(task: Runnable) {
> val t = currentThread
> if (t.isInstanceOf[FJTaskRunner]) {
> val tr = t.asInstanceOf[FJTaskRunner]
> tr.push(new FJTask {
> def run() { task.run() }
> })
> } else
80d82
< }
84,85c86,88
< currentThread match {
< case tr: FJTaskRunner =>
---
> val t = currentThread
> if (t.isInstanceOf[FJTaskRunner]) {
> val tr = t.asInstanceOf[FJTaskRunner]
89c92
< case _ =>
---
> } else
91d93
< }
132c134
< def execute(task: FJTask): Unit
---
> def execute(task: Runnable): Unit
164c166
< def execute(task: FJTask) {
---
> def execute(task: Runnable) {
169c171
< execute(new FJTask {
---
> execute(new Runnable {

diff for FJTaskScheduler2.scala:

152c152
< def execute(task: FJTask): Unit =
---
> def execute(task: Runnable): Unit =
156c156
< executor.execute(new FJTask {
---
> executor.execute(new Runnable {

I did a little test, but it's results are very unstable. Original version
runs 160-200k iterations/s, modified - 190-220k, so improvement is ~3-20%.
Test source:

import scala.actors._
import scala.actors.Actor._

object SpeedTest {
val iterations = 1000000
var i = 0
val t0 = System.currentTimeMillis

val receiver = actor {
loop {
react {
case "Exit" => exit
case _ => reply ("Ok")
}
}
}

val sender = actor {
loop {
react {
case "Run" => receiver ! "Run"
case "Ok" => if (i < iterations) {
reply ("test")
i += 1
} else {
reply ("Exit")
println ((iterations * 1000. / (System.currentTimeMillis-t0))+"
iterations/s")
exit
}
}
}
}

def main (args: Array[String]) {
sender ! "Run"
}
}

Regards,
Alex

Hi Alex,

That's a good point.

The only issue is that we'd have to make sure that it is still possible
to override the def `scheduler` in the `Actor` trait. Also, we probably
have to change the `SchedulerAdapter` trait, so that it wraps
`Runnable`s into `FJTask`s.

I didn't have time to look into it in detail, yet. But, if we can make
sure that your suggested changes integrate well with all the rest of the
scheduler framework, I'd be very happy to integrate your patch.

Thanks,
Philipp

Alex Yakovlev wrote:
> Hi Philipp,
>
> performance-wise, have you considered inheriting Reaction from FJTask
> instead of wrapping it? I mean:
>
> class Reaction extends FJTask {
> ...
> }
>
> and in Scheduler.scala:
>
> def execute(task: Runnable) {
> currentThread match {
> case tr: FJTaskRunner =>
> tr.push(task match {
> case fjt: FJTask => fjt
> case _ => new FJTask {
> def run() { task.run() }
> }})
> case _ =>
> sched execute task
> }
> }
>
> match here works a bit faster than if(isInstanceOf)asInstanceOf because of
> eliminating tr variable. Or even modifying Scheduler to handle only FJTasks
> instead of Runnables?
>
> diff for Reaction.scala:
>
> 51c51
> < class Reaction extends FJTask {
> ---
>> class Reaction extends Runnable {
>
> diff for Scheduler.scala:
>
> 74,78c74,81
> < def execute(task: FJTask) {
> < currentThread match {
> < case tr: FJTaskRunner =>
> < tr.push(task)
> < case _ =>
> ---
>> def execute(task: Runnable) {
>> val t = currentThread
>> if (t.isInstanceOf[FJTaskRunner]) {
>> val tr = t.asInstanceOf[FJTaskRunner]
>> tr.push(new FJTask {
>> def run() { task.run() }
>> })
>> } else
> 80d82
> < }
> 84,85c86,88
> < currentThread match {
> < case tr: FJTaskRunner =>
> ---
>> val t = currentThread
>> if (t.isInstanceOf[FJTaskRunner]) {
>> val tr = t.asInstanceOf[FJTaskRunner]
> 89c92
> < case _ =>
> ---
>> } else
> 91d93
> < }
> 132c134
> < def execute(task: FJTask): Unit
> ---
>> def execute(task: Runnable): Unit
> 164c166
> < def execute(task: FJTask) {
> ---
>> def execute(task: Runnable) {
> 169c171
> < execute(new FJTask {
> ---
>> execute(new Runnable {
>
> diff for FJTaskScheduler2.scala:
>
> 152c152
> < def execute(task: FJTask): Unit =
> ---
>> def execute(task: Runnable): Unit =
> 156c156
> < executor.execute(new FJTask {
> ---
>> executor.execute(new Runnable {
>
> I did a little test, but it's results are very unstable. Original version
> runs 160-200k iterations/s, modified - 190-220k, so improvement is ~3-20%.
> Test source:
>
> import scala.actors._
> import scala.actors.Actor._
>
> object SpeedTest {
> val iterations = 1000000
> var i = 0
> val t0 = System.currentTimeMillis
>
> val receiver = actor {
> loop {
> react {
> case "Exit" => exit
> case _ => reply ("Ok")
> }
> }
> }
>
> val sender = actor {
> loop {
> react {
> case "Run" => receiver ! "Run"
> case "Ok" => if (i < iterations) {
> reply ("test")
> i += 1
> } else {
> reply ("Exit")
> println ((iterations * 1000. / (System.currentTimeMillis-t0))+"
> iterations/s")
> exit
> }
> }
> }
> }
>
> def main (args: Array[String]) {
> sender ! "Run"
> }
> }
>
> Regards,
> Alex

Hi Philipp,

Philipp Haller-2 wrote:
>
> The only issue is that we'd have to make sure that it is still possible
> to override the def `scheduler` in the `Actor` trait.
>

Since it is important to change actor's schedulers, I suppose it is better
to completely hide current implementation with FJTask and leave Runnable in
all *Scheduler traits and classes. So that alternative scheduler
implementations will not depend on FJTask and don't even need to know about
it.

The only overhead will be is checking in object Scheduler.execute if task is
FJTask or general Runnable. IMHO, it is small and worth avoiding of strong
binding of all scheduler framework to current FJTask-based implementation.

So what I suggest is the following:

1. in Reaction.scala:

class Reaction extends FJTask // instead of Runnable

2. in Scheduler.scala:

object Scheduler extends IScheduler {
...
def execute(task: Runnable) {
currentThread match {
case tr: FJTaskRunner =>
tr.push(task match {
case fjt: FJTask => fjt // we do not need to wrap FJTask
case _ => new FJTask {
def run() { task.run() }
}})
case _ =>
sched execute task
}
}

def execute(fun: => Unit) {
currentThread match {
case tr: FJTaskRunner =>
tr.push(new FJTask {
def run() { fun }
})
case _ =>
sched execute { fun }
}
}
...
}

3. in FJTaskScheduler2.scala directly wrap closure into FJTask instead of
Runnable:

def execute(fun: => Unit): Unit =
executor.execute(new FJTask { // was: Runnable
def run() { fun }
})

BTW, have you looked at fork/join implementation in JDK7?

Regards,
Alex