Skip to content

Instantly share code, notes, and snippets.

@fomkin
Forked from viktorklang/minscalaactors.scala
Last active July 10, 2017 12:10
Show Gist options
  • Save fomkin/61c623a45ccc5c6bfafedc3a9673405f to your computer and use it in GitHub Desktop.
Save fomkin/61c623a45ccc5c6bfafedc3a9673405f to your computer and use it in GitHub Desktop.
Minimalist Scala Actors (typed version)
/*
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[T] = T => Effect[T]
def behavior[T](f: Actor.Behavior[T]): Actor.Behavior[T] = f
sealed trait Effect[T] extends (Behavior[T] => Behavior[T])
final def Stay[T] = new Effect[T] { def apply(old: Behavior[T]): Behavior[T] = old }
case class Become[T](like: Behavior[T]) extends Effect[T] { def apply(old: Behavior[T]): Behavior[T] = like }
// Stay Dead plz
final def Die[T] = Become[T](msg => { println("Dropping msg [" + msg + "] due to severe case of death."); Stay})
// The notion of an Address to where you can post messages to
trait Address[-T] { def !(msg: T): Unit }
private abstract class AtomicRunnableAddress[-T] extends Address[T] with Runnable { val on = new AtomicInteger(0) }
// Seeded by the self-reference that yields the initial behavior
def apply[T](initial: Address[T] => Behavior[T])(implicit e: Executor): Address[T] =
// Memory visibility of "behavior" is guarded by "on" using volatile piggybacking
new AtomicRunnableAddress[T] {
// Our awesome little mailbox, free of blocking and evil
private final val mbox = new ConcurrentLinkedQueue[T]
// Rebindable top of the mailbox, bootstrapped to identity
private var behavior: Behavior[T] = {
case self: Address[_] => Become(initial(self.asInstanceOf[Address[T]]))
}
// As an optimization, we peek at our threads local copy
// of our behavior to see if we should bail out early
final override def !(msg: T): Unit = behavior match {
// Efficiently bail out if we're _known_ to be dead
//case dead @ Die.`like` => dead(msg.asInstanceOf[Nothing])
// Enqueue the message onto the mailbox and try to schedule for execution
case _ => mbox.offer(msg); async()
}
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
// Schedule to run on the Executor and back out on failure
try e.execute(this) catch { case t: Throwable => on.set(0); throw t } //
} match {
// Make the actor self aware by seeding its address to the initial
case a: Address[T] => a ! a.asInstanceOf[T]; a
}
}
import Actor._
implicit val e: java.util.concurrent.Executor = java.util.concurrent.Executors.newCachedThreadPool
type Natural = Actor.Address[Int]
val alice = Actor[(Natural, String)] { self =>
behavior {
case (sender, name) =>
println(s"Alice: Oh! $name! I should die")
sender ! 1
Die
}
}
val bob = Actor[Int] { self =>
behavior {
case 0 =>
println(s"Bob: Hello, Alice!")
alice ! (self, "Bob")
Stay
case 1 =>
println(s"Bob: I should die too!")
Die
}
}
bob ! 0
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment