Actors: sending from a different thread

Where do you specify sender? I cannot see the relevant code in your example.

-Stefan

2010/4/19 Ludovic Courtès <ludovic.courtes@inria.fr>

Hello,

There appears to be thread-safety issues with actors/channels as of 2.7,
which the attached program illustrates.

The program creates two actors: ‘Bar’ sends a message to ‘Foo’, which in
turn registers a java.util.TimerTask that eventually sends a message
back to ‘sender’.  When everything goes well, the output looks like
this:

--8<---------------cut here---------------start------------->8---
got object hello from Bar@9d04653
sending to Bar@9d04653
tick!
--8<---------------cut here---------------end--------------->8---

However, every so often, the program fails this way:

--8<---------------cut here---------------start------------->8---
got object hello from Bar@9d04653
Exception in thread "Timer-0" java.util.NoSuchElementException: head of empty list
       at scala.Nil$.head(List.scala:1336)
       at scala.Nil$.head(List.scala:1333)
       at scala.actors.Actor$class.sender(Actor.scala:624)
       at Foo.sender(ActorRunnable.scala:8)
       at Foo$$anonfun$act$2$$anonfun$apply$1$$anon$1.run(ActorRunnable.scala:17)
       at java.util.TimerThread.mainLoop(Timer.java:512)
       at java.util.TimerThread.run(Timer.java:462)
--8<---------------cut here---------------end--------------->8---

Is this a known issue?  What’s the recommended way to work around it?

(My goal is to write a ‘TimerActor’, which can be sent a duration and
reacts by setting up a timer that sends back a ‘Tick’ after that
duration.)

Thanks,
Ludo’.


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

import java.util.{Timer, TimerTask}

case object Tick

class Foo extends Actor {
 val timer = new Timer
 def act: Unit = {
   loop {
     receive {
       case o: Object => {
         println("got object " + o + " from " + sender)
         val thunk = new TimerTask {
           def run = {
             println("sending to " + sender)
             sender ! Tick
           }
         }
         timer.schedule(thunk, 0)
       }
     }
   }
 }
}

class Bar(foo: Foo) extends Actor {
 def act: Unit = {
   foo ! "hello"
   receive {
     case Tick => {
       println("tick!")
     }
   }
 }
}

object ActorRunnable {
 def main(args: Array[String]): Unit = {
   val foo = new Foo
   val bar = new Bar(foo)

   foo.start
   bar.start

   Thread.sleep(2)
 }
}

Hi,

Stefan Langer writes:

> Where do you specify sender? I cannot see the relevant code in your example.

This part:

--8<---------------cut here---------------start------------->8---
val thunk = new TimerTask {
def run = {
println("sending to " + sender)
sender ! Tick
}
}
--8<---------------cut here---------------end--------------->8---

Here ‘sender’ is taken from the enclosing scope; it represents the
sender of the message being processed by actor ‘Foo’.

Thanks,
Ludo’.

But how do you make sure that sender does actually contain a value and wasn't consumed already. The scala docs state that sender returns the last actor that send a message to you so it could change from the time you received the message to the time the timer is executed.
Try saving it to a val and putting that in your TimerTask.

-Stefan

2010/4/19 Ludovic Courtès <ludovic.courtes@inria.fr>

Hi,

Stefan Langer <mailtolanger@googlemail.com> writes:

> Where do you specify sender? I cannot see the relevant code in your example.

This part:

--8<---------------cut here---------------start------------->8---
         val thunk = new TimerTask {
           def run = {
             println("sending to " + sender)
             sender ! Tick
           }
         }
--8<---------------cut here---------------end--------------->8---

Here ‘sender’ is taken from the enclosing scope; it represents the
sender of the message being processed by actor ‘Foo’.

Thanks,
Ludo’.

Hi,

Stefan Langer
writes:

> Try saving it to a val and putting that in your TimerTask.

Unfortunately that doesn’t fix the problem.

Thanks,
Ludo’.

hi,

i want to do something like this:
val x = buffer.filter(predicate)

but instead of creating a new buffer, i want to remove all elements that
do not satisfy the predicate. i found no method on buffers to achieve
this. is there a reason for it not being there?

can you provide a failing and runnable example?

Ludovic Courtès schrieb:

87pr1v62un [dot] fsf [at] inria [dot] fr" type="cite">

Hi,

Stefan Langer 
writes:

  

Try saving it to a val and putting that in your TimerTask.
    

Unfortunately that doesn’t fix the problem.

Thanks,
Ludo’.

  

You use sender in println(***). You should not.
            def run = {
              println("sending to " + sender) // should be println("sending to " + target )
              target ! Tick
            }
-----ludovic.courtes@inria.fr (Ludovic Courtès) wrote: -----

To: scala-user@listes.epfl.ch
From: ludovic.courtes@inria.fr (Ludovic Courtès)
Date: 04/19/2010 05:06PM
Subject: Re: [scala-user] Re: Actors: sending from a different thread

HamsterofDeath <h-star@gmx.de> writes:

> can you provide a failing and runnable example?

Reattached below.

Ludo'.

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

import java.util.{Timer, TimerTask}

case object Tick

class Foo extends Actor {
  val timer = new Timer
  def act: Unit = {
    loop {
      receive {
        case o: Object => {
          println("got object " + o + " from " + sender)
          val target = sender
          val thunk = new TimerTask {
            def run = {
              println("sending to " + sender)
              target ! Tick
            }
          }
          timer.schedule(thunk, 0)
        }
      }
    }
  }
}

class Bar(foo: Foo) extends Actor {
  def act: Unit = {
    foo ! "hello"
    receive {
      case Tick => {
        println("tick!")
      }
    }
  }
}

object ActorRunnable {
  def main(args: Array[String]): Unit = {
    val foo = new Foo
    val bar = new Bar(foo)

    foo.start
    bar.start

    Thread.sleep(2)
  }
}

HamsterofDeath writes:

> can you provide a failing and runnable example?

Reattached below.

Ludo'.

>>>>> "HamsterofDeath" == HamsterofDeath writes:

HamsterofDeath> hi, i want to do something like this: val x =
HamsterofDeath> buffer.filter(predicate)

HamsterofDeath> but instead of creating a new buffer, i want to remove
HamsterofDeath> all elements that do not satisfy the predicate. i found
HamsterofDeath> no method on buffers to achieve this. is there a reason
HamsterofDeath> for it not being there?

It's called filterNot and it's new in 2.8.

Fedor Bobin writes:

> You use sender in println(***). You should not.
>
> def run = {
> println("sending to " + sender) // should be println("sending to " + target )
> target ! Tick
> }

Oops, indeed. Removing this reference seems to fix the problem, thanks!

Ludo’.

that still creates a new collection and does not modify the existing one:

def filterNot(p: A => Boolean): Repr = filter(!p(_))

Seth Tisue schrieb:

13862 [dot] 1271683215 [at] monk [dot] ccl [dot] northwestern [dot] edu" type="cite">

"HamsterofDeath" == HamsterofDeath   writes:
            

 HamsterofDeath> hi, i want to do something like this: val x =
 HamsterofDeath> buffer.filter(predicate)

 HamsterofDeath> but instead of creating a new buffer, i want to remove
 HamsterofDeath> all elements that do not satisfy the predicate. i found
 HamsterofDeath> no method on buffers to achieve this. is there a reason
 HamsterofDeath> for it not being there?

It's called filterNot and it's new in 2.8.

  

Le 19/04/2010 14:57, HamsterofDeath a écrit :

> but instead of creating a new buffer, i want to remove all elements that
> do not satisfy the predicate. i found no method on buffers to achieve
> this. is there a reason for it not being there?

Perhaps it's just because modifying a collection currently processed
could be tricky, even more in a multithreaded context.

All in all, you can still build the collection of (index of?) elements
you want to remove, and then remove them in another loop.

>>>>> "HamsterofDeath" == HamsterofDeath writes:

HamsterofDeath> that still creates a new collection and does not modify
HamsterofDeath> the existing one: def filterNot(p: A => Boolean): Repr
HamsterofDeath> = filter(!p(_))

Oh, I read your question too fast. Yeah, I agree, it's a method that
should be added.

>>>>> "Francois" == Francois writes:

Francois> All in all, you can still build the collection of (index of?)
Francois> elements you want to remove, and then remove them in another
Francois> loop.

The problem being that on a collection like ArrayBuffer where individual
removes are O(n), the overall runtime for the filtering operation will
end up being O(n^2), when it really ought to be O(n).

Francois schrieb:
> Le 19/04/2010 14:57, HamsterofDeath a écrit :
>
>> but instead of creating a new buffer, i want to remove all elements that
>> do not satisfy the predicate. i found no method on buffers to achieve
>> this. is there a reason for it not being there?
>
> Perhaps it's just because modifying a collection currently processed
> could be tricky, even more in a multithreaded context.
>
> All in all, you can still build the collection of (index of?) elements
> you want to remove, and then remove them in another loop.
>
i'm doing it this way:
def removeMatches(filter: (T) => Boolean) {
val before = b.toList
b.clear()
for (value <- before) {
if (!filter(value)) {
b += value
}
}
}

"Perhaps it's just because modifying a collection currently processed
could be tricky, even more in a multithreaded context" ->

multithreading: yes, that would mess the buffer up. but the current
arraybuffer is not thread safe either, so that cannot be the reason.
single threaded, there would not be a problem:

var i = 0
while (i < size) {
if (f(array(i).asInstanceOf[A])) {
removeAt(i);
i-=1
}
i += 1
length = whatever
}

HamsterofDeath wrote:
> hi,
>
> i want to do something like this:
> val x = buffer.filter(predicate)
>
> but instead of creating a new buffer, i want to remove all elements that
> do not satisfy the predicate. i found no method on buffers to achieve
> this. is there a reason for it not being there?
>
>

Can I ask why you want to do this?

because it's the same as
val removeThese = coll.filter(x)
coll--=removeThese

just shorter

Eric Bowman schrieb:
> HamsterofDeath wrote:
>> hi,
>>
>> i want to do something like this:
>> val x = buffer.filter(predicate)
>>
>> but instead of creating a new buffer, i want to remove all elements that
>> do not satisfy the predicate. i found no method on buffers to achieve
>> this. is there a reason for it not being there?
>>
>>
>
> Can I ask why you want to do this?
>

Le 19/04/2010 16:55, HamsterofDeath a écrit :
> Francois schrieb:
>
>> Le 19/04/2010 14:57, HamsterofDeath a écrit :
>>
>>
>>> but instead of creating a new buffer, i want to remove all elements that
>>> do not satisfy the predicate. i found no method on buffers to achieve
>>> this. is there a reason for it not being there?
>>>
>> Perhaps it's just because modifying a collection currently processed
>> could be tricky, even more in a multithreaded context.
>>
>> All in all, you can still build the collection of (index of?) elements
>> you want to remove, and then remove them in another loop.
>>
>>
> i'm doing it this way:
> def removeMatches(filter: (T) => Boolean) {
> val before = b.toList
> b.clear()
> for (value<- before) {
> if (!filter(value)) {
> b += value
> }
> }
> }
>
If you extend an ArrayBuffer, it may be more efficient to do it this way
(not tested):

def removeMatches(filter: (T) => Boolean) {
var j = 0
for (i <- 0 until size0) {
if (!filter(array(i))) {
array(j) = array(i)
j += 1
}
}
reduceToSize(j)
}

Cheers
Sylvain

HamsterofDeath wrote:
> because it's the same as
> val removeThese = coll.filter(x)
> coll--=removeThese
>
> just shorter
>
> Eric Bowman schrieb:
>
>> HamsterofDeath wrote:
>> i want to do something like this:
>> val x = buffer.filter(predicate)
>>
>> but instead of creating a new buffer, i want to remove all elements that
>> do not satisfy the predicate. i found no method on buffers to achieve
>> this. is there a reason for it not being there?
>>
>> Can I ask why you want to do this?
>>

Sorry, wasn't clear. By "this" I mean, modify it in place instead of
just creating a new one. I'm struggling to visualize a non-legacy use
case where this would be the right approach in scala. Is there
something about the size of this list that makes you want to do it? Or
the size of the nodes in the list? Are you worried about GC performance?

cheers,
Eric

On Mon, Apr 19, 2010 at 12:06 PM, Eric Bowman <ebowman@boboco.ie> wrote:

Sorry, wasn't clear.  By "this" I mean, modify it in place instead of just creating a new one.  I'm struggling to visualize a non-legacy use case where this would be the right approach in scala.  Is there something about the size of this list that makes you want to do it?  Or the size of the nodes in the list?  Are you worried about GC performance?

GC performance + heap footprint are the right reasons to want to do this.  Scala has an annoying habit of not fitting problems in memory even on my 16GB RAM workstation.

Using parallel garbage collection helps with GC performance, at least.

  --Rex

Eric Bowman schrieb:
> HamsterofDeath wrote:
>> because it's the same as
>> val removeThese = coll.filter(x)
>> coll--=removeThese
>>
>> just shorter
>>
>> Eric Bowman schrieb:
>>
>>> HamsterofDeath wrote:
>>> i want to do something like this:
>>> val x = buffer.filter(predicate)
>>>
>>> but instead of creating a new buffer, i want to remove all elements
>>> that
>>> do not satisfy the predicate. i found no method on buffers to achieve
>>> this. is there a reason for it not being there?
>>> Can I ask why you want to do this?
>>>
>
> Sorry, wasn't clear. By "this" I mean, modify it in place instead of
> just creating a new one. I'm struggling to visualize a non-legacy use
> case where this would be the right approach in scala. Is there
> something about the size of this list that makes you want to do it?
> Or the size of the nodes in the list? Are you worried about GC
> performance?
>
> cheers,
> Eric
>
because in my case, i give this buffer to several objects (users). they
keep a reference and operate on it. every x seconds, the buffer's
content is updated from another class (dead elements are removed, new
added). of course i could use a listener-like solution
(onContentChange(newList:List[Y])) or wrap the buffer in another object
which always returns the updated content and is updated instead of the
buffer itself, but that just adds a layer of complexity to the simple
update-the-buffer-magically-solution. it requires no additional code in
any of the classes at all. a listener needs to be registered by any
user. the wrapper just tries to hide the update which happens anyway,
just differently (new collection instead of modification).
besides, "--=" already exists, so any argument that goes against
"removeMatches(filter)" also goes against "--="

On Mon, Apr 19, 2010 at 2:51 PM, HamsterofDeath <h-star@gmx.de> wrote:

> Sorry, wasn't clear.  By "this" I mean, modify it in place instead of
> just creating a new one.  I'm struggling to visualize a non-legacy use
> case where this would be the right approach in scala.

because in my case, i give this buffer to several objects (users). they
keep a reference and operate on it. every x seconds, the buffer's
content is updated from another class

But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me.

Anyway, if this sounds like simplicity to you, and performance is not the issue
  collection --= collection.filter(!condition)
is quite brief, and you can use a pimp-my-library pattern if you want to use it inline in longer chains of operations.

But what you're describing sounds like almost a textbook case of the arguments for immutable data structures.

  --Rex
 

Rex Kerr schrieb:

m2l95867d4f1004191145x48953b70u3b0fab843c485efd [at] mail [dot] gmail [dot] com" type="cite"> On Mon, Apr 19, 2010 at 12:06 PM, Eric Bowman <ebowman [at] boboco [dot] ie" rel="nofollow">ebowman@boboco.ie> wrote:

Sorry, wasn't clear.  By "this" I mean, modify it in place instead of just creating a new one.  I'm struggling to visualize a non-legacy use case where this would be the right approach in scala.  Is there something about the size of this list that makes you want to do it?  Or the size of the nodes in the list?  Are you worried about GC performance?

GC performance + heap footprint are the right reasons to want to do this.  Scala has an annoying habit of not fitting problems in memory even on my 16GB RAM workstation.

Using parallel garbage collection helps with GC performance, at least.

  --Rex

i didn't hit 100mb yet, so performance is not my reason. it's simplycity. if i can have a buffer that always contains the data i need, why should i make things more complicated by forcing additional logic into them?

Rex Kerr schrieb:

n2w95867d4f1004191206h7a578c80hdef650f4976aea7c [at] mail [dot] gmail [dot] com" type="cite"> On Mon, Apr 19, 2010 at 2:51 PM, HamsterofDeath <h-star [at] gmx [dot] de" rel="nofollow">h-star@gmx.de> wrote:

> Sorry, wasn't clear.  By "this" I mean, modify it in place instead of
> just creating a new one.  I'm struggling to visualize a non-legacy use
> case where this would be the right approach in scala.

because in my case, i give this buffer to several objects (users). they
keep a reference and operate on it. every x seconds, the buffer's
content is updated from another class

But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me.

Anyway, if this sounds like simplicity to you, and performance is not the issue
  collection --= collection.filter(!condition)
is quite brief, and you can use a pimp-my-library pattern if you want to use it inline in longer chains of operations.

But what you're describing sounds like almost a textbook case of the arguments for immutable data structures.

  --Rex
 

"But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me."

yes... ideally, the buffer should appear to be immutable to the users. the other problems are no problems, at least not in my case. the update never takes place when a user is active

On Mon, Apr 19, 2010 at 9:20 PM, HamsterofDeath <h-star@gmx.de> wrote:

Rex Kerr schrieb:

On Mon, Apr 19, 2010 at 2:51 PM, HamsterofDeath <h-star@gmx.de> wrote:

> Sorry, wasn't clear.  By "this" I mean, modify it in place instead of
> just creating a new one.  I'm struggling to visualize a non-legacy use
> case where this would be the right approach in scala.

because in my case, i give this buffer to several objects (users). they
keep a reference and operate on it. every x seconds, the buffer's
content is updated from another class

But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me.

Anyway, if this sounds like simplicity to you, and performance is not the issue
  collection --= collection.filter(!condition)
is quite brief, and you can use a pimp-my-library pattern if you want to use it inline in longer chains of operations.

But what you're describing sounds like almost a textbook case of the arguments for immutable data structures.

  --Rex
 

"But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me."

yes... ideally, the buffer should appear to be immutable to the users. the other problems are no problems, at least not in my case. the update never takes place when a user is active

What I'd like to know is why the buffer contains things it shouldn't?


--
Viktor Klang
| "A complex system that works is invariably
| found to have evolved from a simple system
| that worked." - John Gall

Akka - the Actor Kernel: Akkasource.org
Twttr: twitter.com/viktorklang

Viktor Klang schrieb:

p2p4d0f9db1004191222i33b5ea2etbae32b3b9023ad94 [at] mail [dot] gmail [dot] com" type="cite">

On Mon, Apr 19, 2010 at 9:20 PM, HamsterofDeath <h-star [at] gmx [dot] de" rel="nofollow">h-star@gmx.de> wrote:

Rex Kerr schrieb:

On Mon, Apr 19, 2010 at 2:51 PM, HamsterofDeath <h-star [at] gmx [dot] de" target="_blank" rel="nofollow">h-star@gmx.de> wrote:

> Sorry, wasn't clear.  By "this" I mean, modify it in place instead of
> just creating a new one.  I'm struggling to visualize a non-legacy use
> case where this would be the right approach in scala.

because in my case, i give this buffer to several objects (users). they
keep a reference and operate on it. every x seconds, the buffer's
content is updated from another class

But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me.

Anyway, if this sounds like simplicity to you, and performance is not the issue
  collection --= collection.filter(!condition)
is quite brief, and you can use a pimp-my-library pattern if you want to use it inline in longer chains of operations.

But what you're describing sounds like almost a textbook case of the arguments for immutable data structures.

  --Rex
 

"But now all the users have to know to keep their hands off the buffer while it's being updated, or they might end up accessing a buffer that is in the process of being modified.  That doesn't sound like simplicity to me."

yes... ideally, the buffer should appear to be immutable to the users. the other problems are no problems, at least not in my case. the update never takes place when a user is active

What I'd like to know is why the buffer contains things it shouldn't?