distributed-process - 分散処理（Parallel and Concurrent Programming in Haskell Chapter 14)
英語の原文はこちらのページで読める
http://chimera.labs.oreilly.com/books/1230000000929/ch14.html
前の章についてまとめたものはこちら
http://jsapachehtml.hatenablog.com/entry/2015/03/17/072031
第１４章では分散プログラミングについて。
haskellではCloud Haskellというプロジェクトで分散プログラミング用のライブラリを開発している。パッケージ名はdistributed-process。
並行処理と分散処理の違いはプロセスが同じマシンで動くか違うマシンで動くかという点。分散処理は各プロセスが異なるマシン上で動いている状態を扱うため、並行処理とは違った以下のような問題点が出てくる。
- どれか一つのプロセスだけがマシンのトラブルなどで止まってしまう可能性があること。それを当然あり得るとして考慮する必要が出てくる。
- 各プロセスがやりとりする方法を考える必要がある。同じマシン上ならば共有メモリの値を読むことで実現できたが違うマシン上ではそうはいかず、ネットワーク越しにやりとりすることになる。なので情報の共有にとても時間がかかる。
- システム全体でグローバルな状態を共有するのが難しい。どのようにしてコンセンサスを取るのかについてアルゴリズムを考える必要がある。
haskellの分散処理ではメッセージパッシングを基本的なモデルとし、TChanをプリミティブとして使っている。
haskellでは言語そのものでは分散処理をサポートいない。代わりにdistributed-processというパッケージを使う。また、それとともにネットワークの処理を扱うパッケージを利用する必要がある。
トランスポートレイヤーの処理はdistributed-processとうまく分離されており、使いたい機能に合わせて付け替えることができる。この章ではトランスポートレイヤーとしてdistributed-process-simplelocalnetを使う。
ping/pong
masterプロセスからslaveプロセスへpingを投げるとpongを返すという例。
今回使うdistributed-processのAPIはこれら。
data Process -- instance Monad, MonadIO data NodeId -- instance Eq, Ord, Show, Typeable, Binary data ProcessId -- instance Eq, Ord, Show, Typeable, Binary -- 自身のプロセスIDやノードIDを取得 getSelfPid :: Process ProcessId getSelfNode :: Process NodeId -- 他のマシン上でプロセスを起動 spawn :: NodeId -> Closure (Process ()) -> Process ProcessId -- メッセージの送受信 send :: Serializable a => ProcessId -> a -> Process () expect :: Serializable a => Process a -- プロセスの終了 terminate :: Process a -- Processモナド内での出力 say :: String -> Process ()
プロセスというのはここで新しく出てきた概念で以下のように説明されている。
メッセージの型
master, slave間でやりとりするメッセージは以下の形で定義する。
{-# Language DeriveDataTypeable #-}
{-# Language DeriveGeneric #-}
data Message = Ping ProcessId
| Pong ProcessId
deriving (Typeable, Generic)
instance Binary Message Ping, PongにProcessIdを含めたのはそれぞれ誰から来たものかを受け取り側が知るため。
Messageがネットワークワーク越しに送信される際は一度バイトデータに変換され、受け取り側で再度haskellのデータに組み直される。(serialize, deserialize)
そのためプロセス間でやりとりされるデータはすべてBinaryのインスタンスにする必要がある。変換の方法はBinaryのインスタンスを自分で導出することで定義できるが基本的には特殊な形にする必要がないため自動導出してほしい。DeriveDataTypeable、DeriveGenericはそのために必要なプラグマ。
BinaryとTypableはSerializableとしてパッケージングされておりControl.Distributed.Process.Serializableにて以下のように定義されている。
class (Binary a, Typeable a) => Serializable a instance (Binary a, Typeable a) => Serializable a
slave側の処理
slave側でpingが届くのを待ち、それに対してpongを応答する処理。
pingServer :: Process ()
pingServer = do
Ping from <- expect
say $ printf "ping received from %s" (show from)
mypid <- getSelfPid
send from (Pong mypid)
$(remotable ['pingServer])まず、関数はProcessモナド内で処理していることに注意。expectによって届いた値をPing fromとしてパターンマッチで取りだす。
各プロセスはチャネルを一つずつ持っていてどんな型のメッセージでも受け取る。expectは型を指定してチャネルから値を取り出す関数である。
expectはreadと似ているが、文脈通りの型にparseできない場合に失敗するreadと違い、expectは型が異なった場合に処理をせずに飛ばす。そしてチャネル内で最初に見つかった指定の型の値を取り出す。型に合わないメッセージはチャネルに残りつづける。また、チャネルに指定の型がなかった場合、expectはブロックすることになるためデッドロックしないよう注意が必要である。
sendでProcessIdを指定してメッセージを送っている。型は以下。
send :: (Serializable a) => ProcessId -> a -> Process ()
最後のremotableは他のマシン上で指定した関数を実行するために必要な設定を提供している。
master側の処理
次にmaster側の処理について。
master :: Process () master = do node <- getSelfNode say $ printf "spawning on %s" (show node) pid <- spawn node $(mkStaticClosure 'pingServer) mypid <- getSelfPid say $ printf "sending ping to %s" (show pid) send pid (Ping mypid) Pong _ <- expect say "pong." terminate
masterではslaveノードの起動、pingの送信、pongの受信、プロセス終了という順で処理を行う。
spawnの型は以下。
spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
NodeIdはプロセスを立ちあげたい対象のノードのID。今回はgetSelfNodeで取得した値を使うことで自身のノードとしている。
2つ目の引数Closure (Process ())で行いたい処理を指定。本質的にはProcess ()を相手のノードに伝えればよいが簡単にはシリアライズすることができない。大量のデータへの参照があったりローカルノード内で共有するような値を持っていたり。そこでClosureでシリアライズの処理を行っている。ちなみにmkStaticClosureは引数なしの関数をClosureにしてくれる。
slaveノードを起動したあと、masterからpingをsendし、expectで返答の受信を待つ。
main関数
mainは以下のようになる。
main :: IO () main = distribMain (\_ -> master) Main.__remoteTable
distribMainはdistributed-process-simplelocalnetの機能を使ったutility関数でmasterやslaveの起動を行っている。
https://github.com/simonmar/parconc-examples/blob/master/DistribUtils.hs
Main.__remoteTableはremotableによって生成された関数で他のマシンで関数を実行するためのメタデータを持っている。
実行結果は以下のようになる。
$ ./ping pid://localhost:44444:0:3: spawning on nid://localhost:44444:0 pid://localhost:44444:0:3: sending ping to pid://localhost:44444:0:4 pid://localhost:44444:0:4: ping received from pid://localhost:44444:0:3 pid://localhost:44444:0:3: pong.
複数ノードでping/pong
先ほどの例を拡張して複数ノードとping/pongのやりとりをする例。
存在するすべてのノード上でプロセスを生成してping/pongする。
master :: [NodeId] -> Process ()
master peers = do
ps <- forM peers $ \nid -> do
say $ printf "spawning on %s" (show nid)
spawn nid $(mkStaticClosure 'pingServer)
mypid <- getSelfPid
forM_ ps $ \pid -> do
say $ printf "pinging %s" (show pid)
send pid (Ping mypid)
waitForPongs ps
say "All pongs successfully received"
terminate
waitForPongs :: [ProcessId] -> Process ()
waitForPongs [] = return ()
waitForPongs ps = do
m <- expect
case m of
Pong p -> waitForPongs (filter (/= p) ps)
_ -> say "MASTER received ping" >> terminateまず存在するノードの発見はdistributed-processのフレームワーク上で行ってくれる。masterはそこで見つかったノードを[NodeId]として引数にとる。
spawnやsendの先ほどとの違いはforMですべてのノードに対して行うという点だけ。
waitForPongsではメッセージの受信を立ちあげたすべてのプロセスから行うためのヘルパー。受け取ったメッセージの送り元のプロセスを除いていき、[ProcessId]のリストが空になれば全て受信し終えたので終了。
main関数は以下のようになる。
main :: IO () main = distribMain master Main.__remoteTable
[NodeId]をフレームワーク側から受け取るようにしたため_ -> masterの部分だけ修正した。
実行
複数ノードを立ち上げるには以下のようにする。
$ ./ping-multi slave 44445 & $ ./ping-multi slave 44446 &
distribMain関数内では引数の形に合わせてdistributed-process-simplelocalnetの関数(startMaster, startSlave)が呼ばれており、slaveの場合ポート番号を指定することでそこで新しいノードを立ち上げることができる。バックグラウンドプロセスにしないと応答が返ってこないので注意。詳しくはこちらで。
https://github.com/simonmar/parconc-examples/blob/master/DistribUtils.hs
実行結果はこんな感じ。
$ ./ping-multi pid://localhost:44444:0:3: spawning on nid://localhost:44445:0 pid://localhost:44444:0:3: spawning on nid://localhost:44446:0 pid://localhost:44444:0:3: pinging pid://localhost:44445:0:4 pid://localhost:44444:0:3: pinging pid://localhost:44446:0:4 pid://localhost:44446:0:4: ping received from pid://localhost:44444:0:3 pid://localhost:44445:0:4: ping received from pid://localhost:44444:0:3 pid://localhost:44444:0:3: All pongs successfully received
今回は同じホスト上で別のノードを立ちあげたが、別ホストにノードを立ち上げることもできる。その際はノード立ちあげ時にホストも指定するだけ。
$ ./ping-multi slave 192.168.100.2 44445 &
その他は何も変更なく指定したホスト上で実行できる。
typed channel
ping/pongの例では一つのチャネルに他の全てのプロセスからのメッセージを受信することになっていた。しかし、それだと以下のようなデメリットがある。
- expectでチャネルにあるメッセージの中で型が一致するものを毎回探すことになるため動作が遅い
- メッセージの中に明示的にどのプロセスから来たかの情報を含める必要がある
これらの問題を解決できる方法としてtyped channelがある。typed channelで用いるAPIはこちらのようになっている。
data SendPort a -- instance of Typeable, Binary data ReceivePort a newChan :: Serializable a => Process (SendPort a, ReceivePort a) sendChan :: Serializable a => SendPort a -> a -> Process () receiveChan :: Serializable a => ReceivePort a -> Process a
まず、送信用・受信用のポートがデータ型として用意されている。SendPort aはSerializableであることに注意。
newChanで新しい一組の送信受信用ポートを取得でき、SendPortから送信したメッセージは対になるReceivePortから受信することができる。sendChan, receiveChanでそれぞれのポートを指定してメッセージの送受信を実行する。
これを使ってやりたいことは以下。
- クライアント側（メッセージを送りたい側）でnewChanによりチャネルを生成
- サーバ側（メッセージを受け取りたい側）へSendPortと共にメッセージを送る
- サーバ側では送られてきたSendPortを使ってメッセージに返答する
- クライアント側ではReceivePortを見ることでメッセージを受信する
typed channelのよいところはこのチャネルに送られてくるメッセージが誰からのものかというが明らかな点である。ProcessIdなどをメッセージに含める必要はない。また、やりとりされるメッセージの型も指定されている。
ping/pongの例をtyped channelを使って書き直すと以下のようになる。
(もともと本にあるコードとは少し変えている）
data Message = Ping (SendPort String) deriving (Typeable, Generic)
instance Binary Message
pingServer :: Process ()
pingServer = do
Ping chan <- expect
mypid <- getSelfPid
sendChan chan $ "pong from " ++ show mypid
$(remotable ['pingServer])
master :: [NodeId] -> Process ()
master peers = do
ps <- forM peers $ \nid -> do
say $ printf "spawning on %s" (show nid)
spawn nid $(mkStaticClosure 'pingServer)
ports <- forM ps $ \pid -> do
say $ printf "pinging %s" (show pid)
(sp, rp) <- newChan
send pid $ Ping sp
return rp
forM_ ports $ \port -> do
m <- receiveChan port
say $ show m
return ()
say "All pongs received"
terminate
main :: IO ()
main = distribMain master Main.__remoteTable MessageにはProcessIdを含める必要がない。今回はStringを含めるようにした。
pingServerでは送られてくるSendPortをexpectで受け取り、それを使ってmasterプロセスに対して返答する。
masterの処理はspawnするまでは変更なし。各プロセスに対してメッセージを送る際にそれぞれチャネルを生成してSendPortを送る。相手からのメッセージを受信はreceiveChanでReceivePortに来る値を読むだけ。
ここで一つ気になるのはmasterからslaveへのメッセージ送信時はtyped channelを使わないのかという点。注意しなければいけないのはReceivePort aはSerializableになっていないということ。ReceivePortを他のプロセスに送れてしまうと送信したメッセージを複数の相手にルーティングする必要が出てくるため、実装が複雑化してしまうと考えられるためそれを防ぐためにできないようになっている。
なので、typed channelを使って双方向に通信したい場合、spawn実行時にslave側で新たにtyped channelを生成してそのSendPortを送り返してもらう必要がある。ただ、今回の例ではping/pongを行うために無駄に複雑な処理をする必要はないためmasterからslaveへの送信はsendを使ってダイレクトに行うようにしている。
エラーハンドリング
プロセス内でエラーが発生した場合の対処について。
ping/pongの例の中でexpectの際に以下のようにパターンマッチして値を取り出している。
Ping chan <- expect
ここで送られてきた値がPongだとするとパターンマッチに失敗してプロセスが死ぬ。その際、それをmaster側で検知したい。
それにはwithMonitorを使う。
withMonitor pid $ do
send pid $ Pong sp
receiveWait
[ match $ \(ProcessMonitorNotification _ deadpid reason) -> do
say $ printf "process %s died: %s" (show deadpid) (show reason)
terminate
] withMonitorはpidと監視のためのアクション(Process ())をとる。アクション内ではreceiveWaitとmatchによって検知したい状態を指定する。
receiveWaitやmatchの型は以下のようになっている。また、実はexpectはreceiveWaitを使って実装されている。
receiveWait :: [Match b] -> Process b match :: forall a b. Serializable a => (a -> Process b) -> Match b expect :: forall a. Serializable a => Process a expect = receiveWait [match return]
なのでreceiveWait内でexpectの代わりに正規の値を待つことも可能。
ProcessMonitorNotificationはエラーを起こしたプロセスのIDやその原因を含むデータ型。
data ProcessMonitorNotification = ProcessMonitorNotification MonitorRef ProcessId DiedReason data DiedReason = DiedNormal -- Normal termination | DiedException !String -- The process exited with an exception | DiedDisconnect -- We got disconnected from the process node | DiedNodeDown -- The process node died | DiedUnknownId -- Invalid (process/node/channel) identifier
DiedReasonとして原因が返ってくるのでなぜ死んだのかを判別する助けになる。ただ、これだけだとコードのどこを修正すればよいかわからないことも多いが。
typed channelを使った実装をやってみた際、なかなかうまくいかずにデバッグする方法を調べたのでそれについては別の記事にまとめた。
http://jsapachehtml.hatenablog.com/entry/2015/03/22/133345
分散チャットサーバ
第１２章で作成したチャットサーバを分散環境で使えるように拡張する。具体的には複数のマシンでチャットサーバを起動させてクライアントはどれにつなげても互いにチャットができるという形。
まずはping/pongの例でやったようにmaster/slaveの役割を持つ形で実装する。その後、各サーバが対等な関係になるよう手を加える。
第１２章での実装と大きく違ってくるのは以下の２点。
- ClientがRemoteClient, LocalClientに分離される
- サーバ間でのメッセージやりとりが存在する
修正点はすべて上記２点に関わる部分であり、そのためのデータ型定義とLocal, Remoteへのメッセージの投げ分けとなる。
サーバの定義についてはわかりづらい点があったのでそれだけ書いておく。
data Server = Server
{ clients :: TVar (Map ClientName Client)
, proxychan :: TChan (Process ())
, servers :: TVar [ProcessId]
, spid :: ProcessId
}第１２章では単にclientのMapを保持しているのみだったが今回は他にもいくつかのデータをもつ。わかりづらいのがproxychanであり、Process ()を持ったTChanである。これは他のサーバへのメッセージを一旦持っておくためのチャネル。
proxychanにデータを入れる関数はsendRemote。入ったデータを処理していくのはproxy。
sendRemote :: Server -> ProcessId -> PMessage -> STM ()
sendRemote Server{..} pid pmsg = writeTChan proxychan (send pid pmsg)
proxy :: Server -> Process ()
proxy Server{..} = forever $ join $ liftIO $ atomically $ readTChan proxychansendRemoteは名前の通りではなく、他のサーバへメッセージを投げるわけではない。ProcessIdとPMessageを指定してsendするアクションを自身のproxychanに入れておく。すると、proxyによって読み取られてProcess ()が実行され、ここで実際に他のサーバに対してメッセージが送られる。
サーバ間のメッセージ送信はdistributed-processの機能を用いて行うのでProcessモナド内の処理となる。しかし、サーバの処理自体は通常のforkIOしたスレッドで行われている。つまりIOモナド内で行われているためProcessモナドを実行できない。
そのため、一旦proxychanに溜めておいてProcessモナドを実行できるスレッドから読み取ることでそれを実行している。各スレッドはchatServer関数で作られる。
chatServer :: Int -> Process () chatServer port = do server <- newServer [] liftIO $ forkIO (socketListener server port) spawnLocal (proxy server) forever $ do m <- expect; handleRemoteMessage server m
socketListenerはクライアントからの接続を監視してaccept、talkという流れ処理をする（１２章での処理とほぼ同じで通常のIO処理）。spawnLocalで作成したスレッドがproxyを実行している。そして、メインスレッドはforeverで他のサーバから来たメッセージを受け取って処理していく。
コードのその他の部分については本 or こちらを参照。
https://github.com/simonmar/parconc-examples/blob/master/distrib-chat/chat.hs
ちなみに本の通りに実装しただけだとまだ動かないので注意。Clientが２つ分離されたりしたことによりtalkやcheckAddClientなどいくつかの関数を修正する必要があった。ただ、コンパイラの出すエラーのおかげで修正すべき部分はわかるので↑のコードを見つつやれば問題ない。
- 作者: Simon Marlow
- 出版社/メーカー: O'Reilly Media
- 発売日: 2013/07/12
- メディア: Kindle版
- この商品を含むブログ (2件) を見る
- 作者: Simon Marlow,山下伸夫,山本和彦,田中英行
- 出版社/メーカー: オライリージャパン
- 発売日: 2014/08/21
- メディア: 大型本
- この商品を含むブログ (2件) を見る
