Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Minimalist Scala Actors
/*
Copyright 2012 Viktor Klang
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
object Actor {
import java.util.concurrent.{ConcurrentLinkedQueue, Executor}
import java.util.concurrent.atomic.{AtomicInteger}
type Behavior = Any => Effect
sealed trait Effect extends (Behavior => Behavior)
case object Stay extends Effect { def apply(old: Behavior): Behavior = old }
case class Become(like: Behavior) extends Effect { def apply(old: Behavior): Behavior = like }
final val Die = Become(msg => { println("Dropping msg [" + msg + "] due to severe case of death."); Stay }) // Stay Dead plz
trait Address { def !(msg: Any): Unit } // The notion of an Address to where you can post messages to
private abstract class AtomicRunnableAddress extends Address with Runnable { val on = new AtomicInteger(0) }
def apply(initial: Address => Behavior)(implicit e: Executor): Address = // Seeded by the self-reference that yields the initial behavior
new AtomicRunnableAddress { // Memory visibility of "behavior" is guarded by "on" using volatile piggybacking
private final val mbox = new ConcurrentLinkedQueue[Any] // Our awesome little mailbox, free of blocking and evil
private var behavior: Behavior = { case self: Address => Become(initial(self)) } // Rebindable top of the mailbox, bootstrapped to identity
final override def !(msg: Any): Unit = behavior match { // As an optimization, we peek at our threads local copy of our behavior to see if we should bail out early
case dead @ Die.`like` => dead(msg) // Efficiently bail out if we're _known_ to be dead
case _ => mbox.offer(msg); async() // Enqueue the message onto the mailbox and try to schedule for execution
}
final def run(): Unit = try { if (on.get == 1) behavior = behavior(mbox.poll())(behavior) } finally { on.set(0); async() } // Switch ourselves off, and then see if we should be rescheduled for execution
final def async(): Unit = if(!mbox.isEmpty && on.compareAndSet(0, 1)) // If there's something to process, and we're not already scheduled
try e.execute(this) catch { case t => on.set(0); throw t } // Schedule to run on the Executor and back out on failure
} match { case a: Address => a ! a; a } // Make the actor self aware by seeding its address to the initial behavior
}
//Usage
import Actor._
implicit val e: java.util.concurrent.Executor = java.util.concurrent.Executors.newCachedThreadPool
//Creates an actor that will, after it's first message is received, Die
val actor = Actor( self => msg => { println("self: " + self + " got msg " + msg); Die } )
actor ! "foo"
actor ! "foo"
@jboner

This comment has been minimized.

Copy link

jboner commented Apr 13, 2012

This is pretty damn awesome. Brilliant.

@hedefalk

This comment has been minimized.

Copy link

hedefalk commented Apr 13, 2012

Sweet!

@santiagobasulto

This comment has been minimized.

Copy link

santiagobasulto commented Jun 1, 2012

This is awesome.

@bhuemer

This comment has been minimized.

Copy link

bhuemer commented Jul 20, 2013

Couldn't this cause issues if you access an actor as a runnable and schedule it for execution directly? That is, if you do both actor ! "foo" and e.execute((Runnable) actor), couldn't you end up having multiple threads that execute the same actor concurrently?

@bhuemer

This comment has been minimized.

Copy link

bhuemer commented Jul 20, 2013

Or to put it differently, why do you test on.compareAndSet(0, 1) in the async() method and not in the run() method?

@viktorklang

This comment has been minimized.

Copy link
Owner Author

viktorklang commented Jan 10, 2014

@bhuemer: Sorry for the late reply, didn't see this until now.

The compareAndSet(0,1) ensures that the runnable is only added once, so that the runnable will never run in parallel, ensuring sequential processing.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.