Cloud Haskellについて調べてみた

Cloud Haskellは["Towards Haskell in the Cloud"](https://www.microsoft.com/en-us/research/wp-content/uploads/2016/07/remote.pdf) by Jeff Epstein, Andrew Black, and Simon Peyton Jones という論文で提案された、Haskellで分散環境で動くプログラムを書くためのDSLです。その実装として[distributed-process](https://hackage.haskell.org/package/distributed-process)というライブラリがあります。論文の著者であるJeffを中心に今も活発に開発が続けられており、歴史については[公式によくまとまっています](http://haskell-distributed.github.io/team.html)。

[Cloud Haskell 公式サイト
<img width=600 src="https://qiita-image-store.s3.amazonaws.com/0/17134/9d397a81-f13a-7fda-a37e-da888018dd33.png">](http://haskell-distributed.github.io/)

## Basic messaging

Cloud Haskellが採用しているのはErlangに影響されたメッセージパッシング型の通信モデルです。Erlangとの大きな違いはメッセージに型がついているところでしょう。

```hs
ping :: Process ()
ping = do
  Pong partner <- expect
  self <- getSelfPid
  send partner (Ping self)
  ping


data Ping = Ping ProcessId deriving (Generic, Typeable)

data Pong = Pong ProcessId deriving (Generic, Typeable)

instance Binary Ping
instance Binary Pong
```

この例は論文から引用してきたものです。`ping`という関数は`Pong`のメッセージが送られてくるのを待ち、受け取ったら送り元に`Ping`のメッセージを返すという動作を繰り返すプロセスの例です。

```hs
expect :: Serializable a => Process a
```

`expect`の型は上記のようになっており、他のプロセスから送られてきたメッセージを受け取ることができます。注目したいのは受け取るメッセージの型が決められているところで、上の例の場合`a`として`Pong`が推論されるため`Pong`メッセージが来るまで待ち続けます。つまり`expect`でブロックしているタイミングで`Ping`メッセージが来ても処理が進むことはなく、確実に`Pong`のメッセージを受け取ることができます。（複数のメッセージを受け取る方法は後で紹介します。）

```hs
send :: Serializable a => ProcessId -> a -> Process ()
```

`send`の型はこの様になっていて、`Serializable`のインスタンスであれば何でもメッセージとして送ることができます。

```hs
class (Binary a, Typeable a) => Serializable a

instance (Binary a, Typeable a) => Serializable a
```

`Serializable`はこのように定義されており、`Binary`と`Typeable`のインスタンスであれば`Serializable`のインスタンスとなる事がわかります。

本来であればプロセス間の通信やノード間の通信にどういったプロトコルを使うかはバイナリのレベルでちゃんと設計して実装しなければいけませんが、`Cloud Haskell`はそこを抽象化してくれており、開発者は`Process`モナドに包まれた処理を記述して`Serializable`のインスタンスとなるようなHaskellの値をメッセージとして送りあうコードを書くだけで、複数のプロセス・複数のノードで分散して動くようなプログラムを簡単に実装することができるのです。

以下にCloud Haskellで提供されているインターフェースの一覧を載せておきます（論文より抜粋、一部`distributed-process`に合わせて修正）。

```hs
-----------------------------
-- Basic messaging
-----------------------------

instance Monad Process

instance MonadIO Process

send :: Serializable a => ProcessId -> a -> Process ()

expect :: Serializable a => Process a

-----------------------------
-- Advanced messaging
-----------------------------

data Match b

receiveWait :: [Match b] -> Process b

receiveTimeout :: Int -> [Match b] -> Process (Maybe b)

match :: Serializable a => (a -> Process b) -> Match b

matchIf :: Serializable a => (a -> Bool) -> (a -> Process b) -> Match b

matchUnknown :: Process b -> Match b

-----------------------------
-- Channels
-----------------------------

newChan :: Serializable a => Process (SendPort a, ReceivePort a)

sendChan :: Serializable a => SendPort a -> a -> Process ()

receiveChan :: Serializable a => ReceivePort a -> Process a

mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

-----------------------------
-- Process management
-----------------------------

spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

terminate :: Process a

getSelfPid :: Process ProcessId

getSelfNode :: Process NodeId

-----------------------------
-- Process monitoring
-----------------------------

link :: ProcessId -> Process ()

monitor :: ProcessId -> Process MonitorRef

```

## Advanced messaging

`match`と`receiveWait`を組み合わせれば複数の型のメッセージを同時に待つことができます。例えば足し算と掛け算とその結果を表す型を以下のように定義します。

```hs
data Add = Add ProcessId Double Double deriving (Generic, Typeable)

data Multiply = Multiply ProcessId Double Double deriving (Generic, Typeable)

data Result = Result Double deriving (Generic, Typeable)

instance Binary Add
instance Binary Multiply
instance Binary Result
```

この`Add`と`Multiply`のメッセージを待って結果を返すようなプロセスは以下のように書くことができます。

```hs
calc :: Process ()
calc = forever $
  receiveWait [ match $ \(Add      pid x y) -> send pid (Result (x + y))
              , match $ \(Multiply pid x y) -> send pid (Result (x * y))
              ]
```

`matchIf`を使えば受け取ったメッセージが第一引数の条件を満たすときだけ処理を行うといったことも可能です。

Cloud Haskellではこのようにメッセージごとに型を定義します。例えば`Add`と`Mutilply`は、

```hs
data MathOp = Add ...
            | Multiply ...
```

のように和型として書けるかもしれませんが、`expect`や`receiveWait`が型で受け取るメッセージを振り分けていることを考えると得策ではありません。更に和型に新しいメッセージを追加したとすると分散している全てのノードのプログラムを一斉に書き換える必要が出てくるため変更に対して弱くなってしまいます。

## Channels
前述の`Ping`や`Add`などのメッセージでは別のメッセージを送り返すために送り元の`ProcessId`がメッセージに含まれていました。Cloud Haskellにはチャネルという概念があり送信チャネルと受信チャネルと作ることで`ProcessId`を引き回す工夫をする必要がなくなり、更にメッセージの型を固定することもできます。

例えば`ping`の例は

```hs
ping2 :: SendPort Ping2 -> ReceivePort Pong2 -> Process ()
ping2 pingout pongin = do
  receiveChan pongin
  sendChan pingout Ping2
  ping2 pingout pongin

data Ping2 = Ping2 deriving (Generic, Typeable)

data Pong2 = Pong2 deriving (Generic, Typeable)

instance Binary Ping2
instance Binary Pong2
```

のように書くことができるでしょう。

ポイントはメッセージに `ProcessId` を含める必要がなくなったことと、`SendPort Ping2`, `ReceivePort Pong2`のように書く送受信チャネルでどの種類のメッセージが送られてくるのかを方で表現することができるところです。

また`SendPort`は`Serializable`のインスタンスになっているため、メッセージに含めて他のプロセスに送信することも可能です。

## Process management

`spawn`は`Process`の値を実行するための関数です。

```hs
spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
```

改めて`spawn`の型を見ると、実行するプロセスは`Closure`という型に包まれています。実は`spawn`は`NodeId`に別のノードのIDを指定することができて、つまり自分のノードで作成したプロセスを別のノードに実行させることが可能です。この時に、作成したプロセスの中に自分の環境への参照（例えばプロセスの外で定義ている変数）が残されたままだと別のノード上でうまく実行することができません。このようなプロセスが参照している環境をまとめて詰め込んだものが`Closure`というわけです。

実際に`Closure`の値を作るときはTemplate Haskellを使います。

```hs
ping :: Process ()
ping = do
  Pong partner <- expect
  self <- getSelfPid
  send partner (Ping self)
  ping


remotable ['ping]


main = do
  ...
    ...
    spawn nid $(mkStaticClosure 'ping)
```

別のノードで実行したい関数を予め[`remotable`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Closure.html#v:remotable)で宣言しておくのがポイントです。今回の例では`ping`は引数を取らなかったので[`mkStaticClosure`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Closure.html#v:mkStaticClosure)を使いましたが、引数を取るプロセスを`Closure`に包むときは[`mkClosure`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Closure.html#v:mkClosure)を使います。

### Control.Distributed.Process.Node
ところで上の例では`...`として誤魔化しましたが、実際に`Process`を実行する処理を一回書いてみましょう。[`Control.Distributed.Process.Node`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Node.html)を使えば、ローカルのノードでプロセスを立ち上げることができます。以下にping/pongの完全なExampleを載せます。

```hs
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TemplateHaskell #-}

import Control.Concurrent (threadDelay)
import Data.Binary
import Data.Typeable (Typeable)
import GHC.Generics (Generic)

import Control.Distributed.Process
import Control.Distributed.Process.Node
import Network.Transport.TCP (createTransport, defaultTCPParameters)

-------------------------
-- Messages
-------------------------

data Ping = Ping ProcessId deriving (Generic, Typeable)

data Pong = Pong ProcessId deriving (Generic, Typeable)

instance Binary Ping
instance Binary Pong

-------------------------
-- Processes
-------------------------

ping :: Process ()
ping = do
  Pong partner <- expect
  say "received Pong"
  liftIO $ threadDelay 2000000
  self <- getSelfPid
  send partner (Ping self)
  ping


pong :: Process ()
pong = do
  Ping partner <- expect
  say "received Ping"
  self <- getSelfPid
  send partner (Pong self)
  pong


main :: IO ()
main = do
  let (host, port) = ("127.0.0.1", "8000")
  Right transport <- createTransport host port ((,) host) defaultTCPParameters
  localNode <- newLocalNode transport initRemoteTable
  runProcess localNode $ do
    pingPID <- spawnLocal ping
    pongPID <- spawnLocal pong
    send pingPID (Ping pongPID)
    receiveWait []
```

[`newLocalNode`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Node.html#v:newLocalNode)を使ってノードを立ち上げ[`runProcess`](https://hackage.haskell.org/package/distributed-process-0.7.4/docs/Control-Distributed-Process-Node.html#v:runProcess)を使ってそのノード上でプロセスを実行しています。

## distributed-process-simplelocalnet
[`distributed-process-simplelocalnet`](http://hackage.haskell.org/package/distributed-process-simplelocalnet)は`distributed-process`に依存したライブラリで、ローカルネットにSlaveとなるノードを立てておくとそれらのノードに対してマスターノードからプロセスを送って分散実行することができる仕組みを簡単に実装することができます。UDPマルチキャストによるローカルネットでのノードの探索とマスタースレイブの仕組みを提供してくれているライブラリです。

```hs
master :: Backend -> [NodeId] -> Process ()
master backend slaves = do
  -- Do something interesting with the slaves
  liftIO . putStrLn $ "Slaves: " ++ show slaves
  -- Terminate the slaves when the master terminates (this is optional)
  terminateAllSlaves backend

main :: IO ()
main = do
  args <- getArgs

  case args of
    ["master", host, port] -> do
      backend <- initializeBackend host port initRemoteTable
      startMaster backend (master backend)
    ["slave", host, port] -> do
      backend <- initializeBackend host port initRemoteTable
      startSlave backend
```

これは[Control.Distributed.Process.Backend.SimpleLocalnet](http://hackage.haskell.org/package/distributed-process-simplelocalnet-0.2.4/docs/Control-Distributed-Process-Backend-SimpleLocalnet.html)に記載されている例です。Slaveとして実行されたときは[`startSlave`](http://hackage.haskell.org/package/distributed-process-simplelocalnet-0.2.4/docs/Control-Distributed-Process-Backend-SimpleLocalnet.html#v:startSlave)を実行し、単純な待機状態となります。Slaveをいくつか立ち上げた後にMasterを実行すると、`master`関数の引数にSlaveの`NodeId`が渡され、その`NodeId`を使ってSlaveノードにプロセスを`spawn`したりすることができます。例では単純に`terminateAllSlaves`を実行して待機状態のSlaveをkillしています。

## まとめ
CloudHaskellを使えば分散環境で動作するプログラムを簡単に作れることがわかりました。`distributed-process`系のライブラリは他にも[distributed-process-client-server](https://hackage.haskell.org/package/distributed-process-client-server)や[distributed-process-p2p](https://hackage.haskell.org/package/distributed-process-p2p)などがあり名前の通りのネットワークが楽に作れるようになっています。

ところで、この記事を書いてる途中で検証していて`distributed-process-simplelocalnet`を使ってSlaveノードにプロセスを実行させるのには成功したのですが、自分でforkIOを使って複数のLocalNodeを立ててプロセスをspawnさせるのには失敗してしまいました。そもそもリモートのノードに任意のプロセスを実行させる機能は信頼できないP2P環境などであればOffにしたいと思うのでこの挙動はありがたいのですが、どういう条件であればspawnさせることができるのか誰か詳しい人がいたらコメント等で教えていただけると幸いです :pray: