More erlang style actor for scala via continuations plugin.

Nice! Might also make a good add-on for Akka, though its direct translation would have more strict semantics, as in Akka you must handle all messages in order; Philipp Haller is working on a trait which adds Erlang-style mailbox behavior to Akka which will be part of the 2.0 release, and then this CPS transform might be applied to both styles so you can pick and choose.

Regards,

Roland

Am Sonntag, 4. Dezember 2011 17:41:19 UTC+1 schrieb IL:

For fun.
https://github.com/iron9light/erla

erlang code:

run() ->
  Number = receive
    I when is_integer(I) ->
      I
    S when is_list(S) ->
      {N, _Rest} = string:to_integer(S),
      N
  end,
  io:format("got int: ~p", [Number]),
  Other = receive
    X -> X
  end,
  io:format("got other: ~p", [Number]).

Pid = spawn(run)

erla code:

val actor = new Actor with ErlActor {
  def actX() = {
    val number = reactX {
      case i: Int => i
      case s: String => s.toInt
    }
    println("got int: %s" format number)
    val other = reactX {
      case x => x
    }
    println("got other: %s" format other)
  }
}.start()

CPS is easy. But supporting erlang-style mailbox need to change the MessageDispatcher and Mailbox.Where can I find the code for that? Did they committed?
IL

There are some details which need to be worked out, still. My guess is that only a new subtype of Mailbox is needed, but that is not yet proven correct.

Regards,

Roland

A new Dispatcher is needed to createMailbox and make sure which executor to use.We can use the same executor for default dispatcher and the erlang-style one.
If we do not consider any performance fact, I think I can implement the erlang-style with just Priority Dispatcher.I've just check the latest akka source from github.I think things gonna be more complicated.
Here is the core in abstract class Mailbox:

  private final def processMailbox() {
    if (shouldProcessMessage) {
      var nextMessage = dequeue()
      if (nextMessage ne null) { //If we have a message
        if (dispatcher.isThroughputDefined) { //If we're using throughput, we need to do some book-keeping
          var processedMessages = 0
          val deadlineNs = if (dispatcher.isThroughputDeadlineTimeDefined) System.nanoTime + dispatcher.throughputDeadlineTime.toNanos else 0
          do {
            actor invoke nextMessage
            processAllSystemMessages() //After we're done, process all system messages
            nextMessage = if (shouldProcessMessage) { // If we aren't suspended, we need to make sure we're not overstepping our boundaries
              processedMessages += 1
              if ((processedMessages >= dispatcher.throughput) || (dispatcher.isThroughputDeadlineTimeDefined && System.nanoTime >= deadlineNs)) // If we're throttled, break out
                null //We reached our boundaries, abort
              else dequeue //Dequeue the next message
            } else null //Abort
          } while (nextMessage ne null)
        } else { //If we only run one message per process
          actor invoke nextMessage //Just run it
          processAllSystemMessages() //After we're done, process all system messages
        }
      }
    }
  }

The actor inside is ActorCell which is private[akka].
This method must be changed.

A new Dispatcher is needed to createMailbox and make sure which executor to use.We can use the same executor for default dispatcher and the erlang-style one.
If we do not consider any performance fact, I think I can implement the erlang-style with just Priority Dispatcher.I've just check the latest akka source from github.I think things gonna be more complicated.
Here is the core in abstract class Mailbox:

  private final def processMailbox() {
    if (shouldProcessMessage) {
      var nextMessage = dequeue()
      if (nextMessage ne null) { //If we have a message
        if (dispatcher.isThroughputDefined) { //If we're using throughput, we need to do some book-keeping
          var processedMessages = 0
          val deadlineNs = if (dispatcher.isThroughputDeadlineTimeDefined) System.nanoTime + dispatcher.throughputDeadlineTime.toNanos else 0
          do {
            actor invoke nextMessage
            processAllSystemMessages() //After we're done, process all system messages
            nextMessage = if (shouldProcessMessage) { // If we aren't suspended, we need to make sure we're not overstepping our boundaries
              processedMessages += 1
              if ((processedMessages >= dispatcher.throughput) || (dispatcher.isThroughputDeadlineTimeDefined && System.nanoTime >= deadlineNs)) // If we're throttled, break out
                null //We reached our boundaries, abort
              else dequeue //Dequeue the next message
            } else null //Abort
          } while (nextMessage ne null)
        } else { //If we only run one message per process
          actor invoke nextMessage //Just run it
          processAllSystemMessages() //After we're done, process all system messages
        }
      }
    }
  }

The actor inside is ActorCell which is private[akka].
This method must be changed.

The idea behind my last mail was to create a special mailbox which supports (bulk) prepending, i.e. a Deque. Then the actor would do just normal processing, and every time it receives a message which it currently does not handle, it appends to a local stash, which it always scans every time it actually processed a message. The mailbox change is only needed in order to transfer the stash back into the queue upon restart, because in Akka it is customary to not lose the mailbox content upon failure. This is (one of the many things that | what) Philipp is currently working on.

Of course this is not the end of it: I think that Erlang-style mailboxes are fundamentally at odds with durable mailboxes, so that combination would not be supportable.

Regards,

Roland

That method needn't be changed. The mailbox has access to the ActorCell, so you can in dequeue access the cell and check the current behavior of the actor against the contents of the mailbox to determine what to return.

I'm not saying you should, but you could.

Cheers,
√

On Dec 5, 2011 7:53 AM, "IL" <iron9light@gmail.com> wrote:

A new Dispatcher is needed to createMailbox and make sure which executor to use. We can use the same executor for default dispatcher and the erlang-style one.
If we do not consider any performance fact, I think I can implement the erlang-style with just Priority Dispatcher. I've just check the latest akka source from github. I think things gonna be more complicated.
Here is the core in abstract class Mailbox:

  private final def processMailbox() {


    if (shouldProcessMessage) {
      var nextMessage = dequeue()


      if (nextMessage ne null) { //If we have a message


        if (dispatcher.isThroughputDefined) { //If we're using throughput, we need to do some book-keeping


          var processedMessages = 0
          val deadlineNs = if (dispatcher.isThroughputDeadlineTimeDefined) System.nanoTime + dispatcher.throughputDeadlineTime.toNanos else 0


          do {
            actor invoke nextMessage
            processAllSystemMessages() //After we're done, process all system messages


            nextMessage = if (shouldProcessMessage) { // If we aren't suspended, we need to make sure we're not overstepping our boundaries


              processedMessages += 1
              if ((processedMessages >= dispatcher.throughput) || (dispatcher.isThroughputDeadlineTimeDefined && System.nanoTime >= deadlineNs)) // If we're throttled, break out


                null //We reached our boundaries, abort
              else dequeue //Dequeue the next message


            } else null //Abort
          } while (nextMessage ne null)


        } else { //If we only run one message per process


          actor invoke nextMessage //Just run it
          processAllSystemMessages() //After we're done, process all system messages


        }
      }
    }
  }

The actor inside is ActorCell which is private[akka].
This method must be changed.

Ok, Just commited the akka version.

val actor = Actor.actorOf(new ErlActor {
  def act() = {
    val number = react {
      case i: Int => i
      case s: String => s.toInt
    }
    println("got int: %s" format number)
    val other = react {
      case x => x
    }
    println("got other: %s" format other)
  }
}).start()

Nearly the same as scala one.

cool!