Akka vs Erlang

Erlang Advent Calendar 2014の22日目です。よろしくお願いいたします。

最近すごいErlangゆかいに学ぼう！を読んだので、Akkaとの違いをまとめてみることにしました。

AkkaもErlangもアマチュアなので正しくない部分があるかもしれません。その場合はご指摘ください。
Erlangは特に素人で、AkkaのactorとErlangのプロセス間でメッセージパッシングしてみたの記事で初めて「-compile(export_all)」でないプログラムを書いたレベルです。

Akkaとは

AkkaはErlangにインスパイヤされて制作されたオープンソースのライブラリです。
アクターモデルで並行処理を記述し、スーパーバイザーツリーにより高度な耐障害性を実現し、また分散システムを実装してくれるのを助けてくれる、そのようなフレームワークとなっています。
Java用のAPIとScala用のAPIが用意されておりますが、Scala製プログラムで利用されていることが多いです。

利用例

• Web Framework(Play Framework)
• ストレステストツール(Gatling)
• 各種DBクライアント(Redis, Riak等)

より多くの例をご覧になりたい方は、Community Projectsを眺めてみるのがよろしいかと思います。

Scala vs Erlang

まずはAkkaとErlang、というよりはScalaとErlangの言語仕様に起因する違いについて語っていきます。

型付け

Erlangは動的型付け、Scalaは静的型付けです。
私は静的型が好きな人間だったので、その点ではあまりErlangを好きになれないな、と感じていたのですが、Dialyzerの章を読んでそんなに悪くないなと思いました。

直和を「|」で連結するだけで表現できたり、数値の範囲を直接表現できる点はScalaの型システムの表現力を上回っているとすら感じました。

-type num() :: integer() | float() | infinity.
-type month() :: 1..12.

とはいえDialyzerは静的解析ツールにすぎない上に、若干楽観的に動作するので、基本的にはScalaほどの型検査はしてくれないと考えてよいと思います。
Erlangが得意なOさんは「型情報を利用したプログラミングができない」とおっしゃっていました。

GC

Erlangの場合

プロセス単位でGCが走るとうかがっております。

Scalaの場合

ScalaはJVM上で動くので、ヒープに対してGCが実行されます。その結果、いわゆる「stop the world」が発生します。
ハートビートが届かなくなるので、Akkaで分散システムを構築する場合は厳しそうですが、一応Failure DetectorにはGCを考慮した設定を反映させることができます。

パターンマッチ

ScalaもErlangもパターンマッチ機能を言語に備えており、パワフルな記述を実現できます。
どちらも同じくらい使いやすく、ほとんど同じような使い勝手なのですが、一点Erlangならではの機能がありました。バイナリパターンマッチです。

Erlangの場合

この機能の強力さを感じるには、以下のTCPセグメントのパーズ例を見るのがよいと思います。

< AckNumber:32,
DataOffset:4, _Reserved:4, Flags:8, WindowSize:16,
CheckSum: 16, UrgentPointer:16,
Payload/binary>> = SomeBinary.

Learn you some Erlang for great good!

Erlangがどのようなユースケースで利用されているかがなんとなく感じ取れる機能ですね。

Scalaの場合

怖いライブラリを使いましょう。

ホットコードローディング

Erlangの場合

Erlangは言語自体にホットデプロイするための仕組みが備わっています。

このようなプログラムを

-module(hotdeploy).
-export([start/0, loop/0, print/2, update/1]).

start() ->
    {ok, spawn(fun() -> loop() end)}.

loop() ->
    receive
        {print, Num} ->
            io:format("~p~n", [Num]),
            loop();
        update ->
            io:format("version update~n"),
            ?MODULE:loop()
    end.

print(Pid, Num) ->
    Pid ! {print, Num}.

update(Pid) ->
    Pid ! update.

1> c(hotdeploy).
{ok,hotdeploy}
2> {ok, Pid} = hotdeploy:start().
{ok,<0.39.0>}
3> hotdeploy:print(Pid, 5).
5
{print,5}

書き換えて、

loop() ->
    receive
        {print, Num} ->
            io:format("~p~n", [Num*2]),
            loop();
        update ->
            io:format("version update~n"),
            ?MODULE:loop()
    end.

再コンパイルすると、update用フックを呼び出したタイミングで新しい動作に切り替わります。

4> c(hotdeploy).
{ok,hotdeploy}
5> hotdeploy:print(Pid, 5).
5
{print,5}
6> hotdeploy:update(Pid).
version update
update
7> hotdeploy:print(Pid, 5).
10
{print,5}

Scalaの場合

このようなホットデプロイは私が把握している限りでは無理だと思います。
Akkaの場合であれば、クラスタを組んでおいて更新の際は一台ずつ切り離して再起動してから再join、みたいなことをすれば一応ホットデプロイはできそうな気がしますが、相当につらいロジックを組む必要が出てきそうです。

とはいえErlangのホットコードローディングを使えたとしても、かなり注意しないと壊れてしまうのは目に見えているので、使用するにはかなりの準備と勇気が必要そうです。
以前勉強会で、ドワンゴの方がこの機能を実戦投入していると発表していました。

Akka vs Erlang

ここからはAkkaとErlangの違いについてです。

アクターモデルの実装

プロセス(アクター)同士がメッセージパッシングでやりとりし、受け取ったメッセージに対する処理を記述することで処理を進めていく、という点についてはAkkaもErlangも同じです。

Erlangの場合

Erlangではプロセス(アクター)を以下のように定義・生成しますね。

-module(printer).
-export([start/0]).

start() ->
    {ok, spawn(fun() -> loop() end)}.

loop() ->
    receive
        {print, X} ->
            io:format("~p~n", [X]),
            loop();
        stop ->
            ok
    end.

1> c(printer).
{ok,printer}
2> {ok, Pid} = printer:start().
{ok,<0.39.0>}
3> Pid ! {print, hello}.
hello
{print,hello}
4> Pid ! {print, <<"Learn You Some Erlang for Great Good!">>}.
<<"Learn You Some Erlang for Great Good!">>
{print,<<"Learn You Some Erlang for Great Good!">>}
5> Pid ! stop.
stop
6> Pid ! {print, hello}.
{print,hello}

spawnされた関数が新たなプロセス上で動いている、という感じでしょうか。
関数なので、処理の継続を再帰で表現します。この関数の停止条件は「stop」メッセージを受け取ったときなので、「stop」を送信したあとはprintメッセージを受け取っても何も出力されなくなります。

Akkaの場合

一方、Akkaにおけるアクターは、あくまでScalaのオブジェクトです。

package sample

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

// classはJavaのクラスのようなものだと思ってください。
class AkkaActor extends Actor {
  override def receive: Receive = {
    case AkkaActor.Print(value) => println(value)
    case AkkaActor.Stop => context.stop(self)
  }
}

// objectはJavaのstaticメンバーを配置するところだと勘違いしていただければこの記事を読む分には問題ないです。
object AkkaActor {
  case class Print(value: String)
  case object Stop
  val system = ActorSystem("akka-actor")

  def create(): ActorRef = {
    system.actorOf(Props(classOf[AkkaActor]))
  }
}

scala> import sample.AkkaActor
import sample.AkkaActor

scala> val ref = AkkaActor.create()
ref: akka.actor.ActorRef = Actor[akka://akka-actor/user/$a#-1887968506]

scala> ref ! AkkaActor.Print("hello")
hello

scala> ref ! AkkaActor.Print("world")
world

scala> ref ! AkkaActor.Stop

scala> ref ! AkkaActor.Print("let it crash")

scala> [INFO] [12/22/2014 17:19:07.200] [akka-actor-akka.actor.default-dispatcher-4] [akka://akka-actor/user/$a] Message [sample.AkkaActor$Print] from Actor[akka://akka-actor/deadLetters] to Actor[akka://akka-actor/user/$a#-1887968506] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

受け取ったメッセージが巡り巡ってActorクラスのインスタンスのreceiveメンバーに引き渡される、といった動作となっています。
関数と違い、処理の終端が存在しないので、終了するときは明示的に指示する必要があります(context.stop(self)の部分)。
ActorSystem, ActorRef, Actorと似たクラスが並んでいますが、これらはそれぞれErlangのルートのプロセス、プロセス、spawnに引き渡す処理(関数)と読み替えていただければ大体あっていると思います。
最後に長いログが出力されていますが、これはデッドレターというものです。Akkaは、存在しないActorRefにメッセージを送信すると、そのイベントがグローバルなイベントマネージャ的なところにpublishされ、デフォルト設定ではログが出力されるようになっています。

アクターの状態

Erlangの場合

Erlangプロセスの実体は再帰関数です。したがって、状態は関数のパラメータを用いて表現することができます。

-module(kvs).
-export([start/0]).

start() ->
    {ok, spawn(fun() -> loop(maps:new()) end)}.

loop(Map) ->
    receive
        {get, Key} ->
            io:format("~p~n", [maps:find(Key, Map)]),
            loop(Map);
        {set, Key, Value} ->
            loop(maps:put(Key, Value, Map))
    end.

1> c(kvs).
{ok,kvs}
2> {ok, Pid} = kvs:start().
{ok,<0.39.0>}
3> Pid ! {get, "mofu"}.
error
{get,"mofu"}
4> Pid ! {set, "mofu", "poyo"}.
{set,"mofu","poyo"}
5> Pid ! {get, "mofu"}.
{ok,"poyo"}
{get,"mofu"}

Akkaの場合

一方、AkkaはScalaで実装されているので、変数やmutableなコレクションを用いることで、状態を表現することができます。

package sample

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

class KVS extends Actor {
  // varは再代入可能な変数。valは再代入不可。
  var map: Map[String, String] = Map.empty

  override def receive: Receive = {
    case KVS.Get(key) => println(map.get(key))
    case KVS.Set(key, value) => map = map.updated(key, value)
  }
}

object KVS {
  case class Get(key: String)
  case class Set(key: String, value: String)
  val system = ActorSystem("kvs-actor")

  def create(): ActorRef = {
    system.actorOf(Props(classOf[KVS]))
  }
}

scala> import sample.KVS
import sample.KVS

scala> val ref = KVS.create()
ref: akka.actor.ActorRef = Actor[akka://kvs-actor/user/$a#1068972862]

scala> ref ! KVS.Get("mofu")
None

scala> ref ! KVS.Set("mofu", "poyo")

scala> ref ! KVS.Get("mofu")
Some(poyo)

メッセージは１つずつ逐次的に処理されていくので、排他制御は不要です。
Actorはマルチスレッドで動いているのでメモリモデル的に同期化が必要な気もしますが、その辺もAkka側がよしなに面倒を見てくれます。
私はAkkaから入ったのでErlangのアプローチがしっくりこず、「状態を扱うならgen_serverとか使うといいよ」と言われて「なんでそれだけのためにサーバー立てなきゃいけないんだよ」と噛み付いてしまったのはよい思い出です。

返信

AkkaやErlangを書いていると、メッセージの返信をしたい、というケースが訪れます。
返信をするには、メッセージの送信元のActorを特定する必要があります。

Erlangの場合

プロセスIDをメッセージに含めることで、返信先を指定します。

-module(echo_server).
-export([start/0]).

start() ->
    {ok, spawn(fun() -> loop() end)}.

loop() ->
    receive
        {Pid, Message} -> Pid ! Message
    end.

1> c(echo_server).
{ok,echo_server}
2> {ok, Pid} = echo_server:start().
{ok,<0.39.0>}
3> Pid ! {self(), hello}.
{<0.32.0>,hello}
4> flush().
Shell got hello
ok

Akkaの場合

package sample

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

class EchoServer extends Actor {
  override def receive: Receive = {
    case EchoServer.Request(message) => sender() ! EchoServer.Response(message)
  }
}

class EchoClient(server: ActorRef) extends Actor {
  override def receive: Receive = {
    case message: String => server ! EchoServer.Request(message)
    case response: EchoServer.Response => println(response)
  }
}

object EchoServer {
  case class Request(message: String)
  case class Response(message: String)
  val system = ActorSystem("echo-actor")

  def create(): ActorRef = {
    val server = system.actorOf(Props(classOf[EchoServer]))
    system.actorOf(Props(classOf[EchoClient], server))
  }
}

scala> import sample.EchoServer
import sample.EchoServer

scala> val ref = EchoServer.create()
ref: akka.actor.ActorRef = Actor[akka://echo-actor/user/$b#-1754226359]

scala> ref ! "hello"
Response(hello)

Scalaの場合すべてがプロセス(Actor)、というわけにはいかないのでクライアントとなるActorも作成しました。
特筆すべきは「sender() ! EchoServer.Response(message)」の部分ですね。
クライアント側ではどのActorRefから送信したか一切明示的に指定していないのに、サーバ側からは「sender()」を呼び出すだけで送信元が特定できています。
Scalaには引数を指定しなくても、スコープ内に条件に合致するオブジェクトが存在した場合に勝手にメソッドにパラメータを引き渡してくれる闇の機能があり、それを利用しています。

ActorRef

def !(message: Any)(implicit sender: ActorRef = Actor.noSender): Unit

最初にErlangを触った時は、わざわざメッセージにプロセスIDを埋め込まなきゃいけない点に煩わしさをかんじました。
でもしばらくすると、むしろメッセージに返信先の情報が含まれていたほうが、プロトコル上返信する必要があるかどうかがメッセージを見ただけで見分けられて便利なことに気づきました。
senderメソッドはない方がよいのかもしれません。

メールボックス

Erlangの場合

メールボックスはFIFOなキューです。

-module(countdowner).
-export([start/0, countdown/2]).

start() ->
    {ok, spawn(fun() -> loop() end)}.

loop() ->
    receive
        Message ->
            io:format("~p~n", [Message]),
            loop()
    end.

countdown(_, 0) ->
    ok;
countdown(Pid, N) ->
    Pid ! N,
    countdown(Pid, N-1).

1> c(countdowner).
{ok,countdowner}
2> {ok, Pid} = countdowner:start().
{ok,<0.39.0>}
3> countdowner:countdown(Pid, 10).
10
9
ok
8
7
6
5
4
3
2
1

Akkaの場合

Akkaのメールボックスも、基本的にはやはりFIFOなキューです。

package sample

import akka.actor.{Actor, ActorRef, ActorSystem, Props}

class FIFOCountDowner extends Actor {
  override def receive: Receive = {
    case count: Int => println(count)
  }
}

object FIFOCountDowner {
  val system = ActorSystem("count-downer")

  def create(): ActorRef = {
    system.actorOf(Props(classOf[FIFOCountDowner]))
  }

  def countDown(ref: ActorRef, n: Int): Unit = {
    n match {
      case 0 =>
      case _ =>
        ref ! n
        countDown(ref, n - 1)
    }
  }
}

scala> import sample.FIFOCountDowner
import sample.FIFOCountDowner

scala> val ref = FIFOCountDowner.create()
ref: akka.actor.ActorRef = Actor[akka://count-downer/user/$a#-1406074768]

scala> FIFOCountDowner.countDown(ref, 10)
10
9
8
7
6
5
4
3
2
1

しかしながら、Akkaの場合「基本的でない」メールボックスを持ったActorを作ることもできます。

package sample

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import akka.dispatch.{PriorityGenerator, UnboundedPriorityMailbox}
import com.typesafe.config.{Config, ConfigFactory}

class TODOManager extends Actor {
  override def receive: Receive = {
    case task: TODOManager.Task =>
      println(task)
    case TODOManager.Register(tasks) =>
      tasks.foreach { task =>
        self ! task
      }
  }
}

object TODOManager {
  val Generator = PriorityGenerator {
    case Task(deadline) => deadline
    case _ => Int.MaxValue
  }
  class TaskPriorityMailBox(settings: ActorSystem.Settings, config: Config)
    extends UnboundedPriorityMailbox(Generator)

  case class Task(deadline: Int)
  case class Register(tasks: Seq[Task])

  val config = ConfigFactory.parseString(
    """{
      |  task-priority-mail-box {
      |    mailbox-type = "sample.TODOManager$TaskPriorityMailBox"
      |  }
      |}""".stripMargin)
  val system = ActorSystem("todo-manager-actor", config)

  def create(): ActorRef = {
    system.actorOf(Props(classOf[TODOManager]).withMailbox("task-priority-mail-box"))
  }
}

scala> import sample.TODOManager
import sample.TODOManager

scala> val ref = TODOManager.create()
ref: akka.actor.ActorRef = Actor[akka://todo-manager-actor/user/$a#2131552672]

scala> val tasks = Seq(1, 5, 3, 99, 8).map(deadline => TODOManager.Task(deadline))
tasks: Seq[sample.TODOManager.Task] = List(Task(1), Task(5), Task(3), Task(99), Task(8))

scala> ref ! TODOManager.Register(tasks)
Task(1)
Task(3)
Task(5)
Task(8)
Task(99)

プライオリティが高い(期日が近い)タスクが先に表示されました。

Erlangが得意なOさんは、「Akkaは言語レベルでなくライブラリとしてアクターモデルが実装されているから、自由度が高い実装が可能だと思う」とおっしゃっておりました。
メールボックス自前実装はその良い例だと思います。

まとめ

以上、AkkaとErlangを勉強して、見つけた違いを列挙してみました。
練習も兼ねてコードを書きながら執筆したので、無駄に長くなってしまいました。
本当はスーパーバイザー周りにも触れたかったのですが、そこまで踏み込むとアドヴェント破りをしてしまうこと間違いなしなので諦めます(◞‸◟)

「Akka vs Erlang」と銘打った以上決着をつけたいところではありますが、私の実力ではマサカリを避けきれないのでそれは遠慮させていただきます。
初心者の意見といたしましては、普通のプログラムが書きやすく、Webアプリケーションの開発環境も充実しているScala + Akkaの方が幾分取っ付き易いかなとは思いました。