やさしいFunctional reactive programming（Event編）

Haskell FRP

前回はFRPのBehaviorとEventという重要な概念と、Haskellでの代表的なライブラリを簡単に紹介しました。今回は紹介したものの中からreactiveというライブラリを取り上げます。

なお、reactiveを理解するにあたってLess Sugar/More Meat » Blog Archive » Introducing Reactive: Eventsというチュートリアルがとても役に立ったことを書いておきます。この後出てくる解説も冒頭部分はこのチュートリアルからとってきています。それと、動作環境はLinux上のGHC 6.12.1でコンパイルしたreactive 0.11.4を使っています。

reactiveの特徴

reactiveはConal Elliottが作った最新のFRPライブラリで、ライブラリのユーザ側から見た大きな特徴は、Arrowを使わない古典的FRPをFunctorにApplicative、あるいはMonoidやMonadを使って再定義しているところです。これにより古典的FRPに似た関数的なインタフェースを、より親しみやすい形で利用できるようになっています。

reactiveがArrowを使わない理由はConal ElliottのConal Elliott » Blog Archive » Why classic FRP does not fit interactive behaviorを読む限り、

  • Arrow式では複数のbehaviorを一つにまとめる必要があり、これが効率的な評価を妨げているようであること。
  • BehaviorとEventの区別ができなくなってしまうこと。表現力や正確な意味を損ねることなくBehaviorとEventを統合させたいと思っている。
  • Arrowを使ったプログラミングはArrow記法なしではかなり不格好で、なおかつArrow記法は彼の好みより逐次的なスタイルになってしまうこと。古典的FRPの関数的な書き方が好みらしい。

というもののようです。

reactiveにおけるEventとBehavior

reactiveにおけるEventとBehaviorは意味的には次のように考えられます。

詳しく見ていきましょう。Event は時間と値のペアのリストと考えられます。ここでについている^はがからまでの値域であることを表しています。具体的には次のような例が考えられるでしょう。

  • [(1.0, 'h'), (2.0, 'e'), (3.0, 'l'), (3.1, 'l'), (4.0, 'o')]というユーザからのキー入力
  • [(50.0, FSharp), (50.0, ASharp), (50.0, CSharp)]というギターの和音
  • [(, ()), (0, ()), (5, ())]のように無限大も許されます

ただし時間は昇順でないとならないので、

  • [(1, ()), (0, ())]

はEventではありません。

注意しなければならないのは、reactiveの実装が上のようになっているわけではないということです。実装はもっと効率的な形になっていますが、その意味や性質が上記のようになっているということです。

続いてBehaviorについても見てみましょう。Behavior は時間の関数になっています。時間をとって値を返す関数です。前回触れた連続的な変化というのはこの関数のことを表しています。具体例を挙げると

  • 時間を受け取り時間を返す関数（idみたいなもの）
  • 時間とともに正弦値を返す関数（sin関数みたいなもの）
  • 時間にかかわらず定数を返す関数（constみたいなもの）

など、様々な関数が考えられます。これ以外にもBehaviorは不連続な関数を扱うこともできます。

それでは、EventとBehaviorをそれぞれサンプルコードを交えて、その挙動や使い方を見ていきます。

Eventで遊んでみる（BellMachine.hs）

Eventの意味は時間と値のペアのリストであると説明しました。では実際にEventを作るにはどうしたらよいのでしょうか。reactiveのドキュメントを見るとEventを生成する関数がいくつも並んでいます。これらを使ってEventの動きを実際に試してみましょう。

最初に、プログラム起動後、所定の時間に"BEEP!"と出力するタイマーを作ってみます。reactiveを使うときは必ず最初にFRP.Reactiveをimportします。

例えば「3秒後に発火する*1」ことをEventを使って表すには次のようにします。

bellTimer :: Event ()
bellTimer = atTime 3   -- 意味的には[(3, ())]というイベントを作っている

コメントにあるように、atTimeを使うと3秒後に()という値を生成するEventを作れます。また「毎秒発火する」ようなEventは次のように書けます。

bellTimer' :: Event ()
bellTimer' = atTimes [1..] -- 意味的には[(1, ()), (2, ()), ...]というイベントを作っている

Eventの発火を受けて"BEEP!"と出力する部分はどうしましょう。bellActionはEventが生成する()を受け取り、"BEEP!"と出力するIO()を返す関数と考えられます。すなわち

bellAction :: a -> IO ()
bellAction = const $ putStrLn "BEEP!" -- 何を受け取るのかにかかわらず"BEEP!"と出力する

と定義できるでしょう。

しかしこのままではEventに関する関数であるbellTimerと単なるIOであるbellActionを結びつけられません。これを結びつけるためにApplicative functorを使います。

bellEvent :: Event a -> Event (IO ())
bellEvent = bellAction <$> bellTimer

Applicative functorの力を借りて、bellActionという（Eventに対しては）pureな計算を、（Eventに対して）effectfulな計算に持ち上げています。ここら辺の感覚がよくわからない方は、是非ともTypeclassopedia（邦訳）を読まれることをお勧めします。

これで準備は整いました。後はEvent (IO ())を実際のIOに変換するだけです。これにはFRP.Reactive.LegacyAdaptersが提供するadaptEという関数を使います。プログラムの全体像を見てみましょう。

import FRP.Reactive                ( Event, atTime, atTimes )
import FRP.Reactive.LegacyAdapters ( Action, adaptE )
import Control.Applicative         ( (<$>) )

bellTimer :: Event ()
bellTimer = atTime 3

bellTimer' :: Event ()
bellTimer' = atTimes [1..]

bellAction :: () -> Action
bellAction = const $ putStrLn "BEEP!"

main :: IO ()
main = adaptE $ bellAction <$> bellTimer'

Actionという型は先のLegacyAdapters内で定義された型シノニムで、単にIO ()を置き換えているだけです。実際に走らせてみましょう。

maoe@maoe ~/coding/haskell/reactive-examples
$ ghc --make BellMachine.hs
[1 of 1] Compiling Main             ( BellMachine.hs, BellMachine.o )
Linking BellMachine ...
maoe@maoe ~/coding/haskell/reactive-examples
$ ./BellMachine 
BEEP!
BEEP!
BEEP!

一秒ごとに"BEEP!"と出力されました。

reactiveでは基本的にどんなアプリケーションも

  • FRP.Reactive
  • FRP.Reactive.LegacyAdapters

の2つをインポートすれば動くように設計されています。今後のサンプルコードもこの2つがインポートされていると考えてください。

インタラクティブなEvent（BellMachineI.hs）

次に、ユーザからの入力によって挙動が変わるインタラクティブなEventについて考えてみます。

ユーザの入力はEventとして見えます。インタラクティブなEventとは、Eventを受け取りEventを返す関数です。これを使って先ほどの例を拡張して、

  • 2秒ごとに"BEEP!"と出力
  • さらに、ユーザのキー入力に併せて"BEEP!"と出力

となるようにしてみましょう。まずEventを受け取りEventを返すという型をBellMachineと名付けます。

type BellMachine = Event () -> Event ()

ユーザからの入力Eventを受け取り、Eventを生成するBellMachineは恒等関数idで書けます。

bell :: BellMachine
bell = id

さらに2秒に1回Eventを生成するBellMachineは、入力には依存しないためconstを使って書けます。

bellTimer :: BellMachine
bellTimer = const $ atTimes [0, 2..]

ではbellとbellTimerをどうやって組み合わせたらいいでしょうか。複雑なタイマー制御が必要でしょうか？イベントループが必要になるでしょうか？実はどちらも要りません。ここではMonoidを使います。Eventは意味的に時間と値の組の「リスト」であると言いました。リストがMonoidであるのと同様にEventもまたMonoidとなっています。これにより、Event同士はMonoidのメソッドで結合することができます*2。Monoidについて詳しく知りたい方には、やはりTypeclassopediaをお勧めします。

nifty :: BellMachine
nifty = bellTimer `mappend` bell

I/Oへの接続はどうでしょうか。bellActionは先ほどと同様に使えます。問題はキー入力をEvent ()に変える方法です。先にコードの全体を見せます。

import FRP.Reactive                ( Event, atTimes )
import FRP.Reactive.LegacyAdapters ( makeClock, Action, makeEvent, adaptE )
import Control.Applicative         ( Applicative(pure), (<$>) )
import Control.Concurrent          ( forkIO )
import Control.Monad               ( forever )
import Data.Monoid                 ( Monoid(mappend) )
import System.IO                   ( stdout, stdin, hSetBuffering, hSetEcho
                                   , BufferMode(NoBuffering) )

type BellMachine = Event () -> Event ()

bell :: BellMachine
bell = id

bellTimer :: BellMachine
bellTimer = const $ atTimes [0, 2..]

nifty :: BellMachine
nifty = bellTimer `mappend` bell

bellAction :: a -> Action
bellAction = const $ putStrLn "BEEP!"

main :: IO ()
main = do
  hSetBuffering stdin NoBuffering
  hSetBuffering stdout NoBuffering
  hSetEcho stdin False
  (sink, event) <- makeEvent =<< makeClock
  forkIO $ forever $ getChar >> sink ()
  adaptE $ bellAction <$> nifty event

mainの中のsinkとeventがポイントです。reactiveでは外部ソース（キー入力など）からEventを生成するのにSinkという仕組みを使います。SinkはmakeEvent関数でEventとともに作ります。sink aとしてSinkに値を渡すと、eventではEvent aという値を取り出せるようになります。

Sinkへ値を流し込むのと、イベント処理を並行して行うため、例ではスレッドを使っています。たいていのアプリケーションでこのようなスタイルになるでしょう。hSetBufferingとhSetEchoはそれぞれ標準入出力のバッファリングを無効にしているのと、キー入力を端末に表示する機能を無効化しています。

実際に走らせてみましょう。

maoe@maoe ~/coding/haskell/reactive-examples
$ ./BellMachineI 
BEEP!
BEEP!
BEEP!

何も入力しないと2秒に1回BEEP!と出力されます。さらに何かしらキー入力しても同様にBEEP!と出力されるのがわかると思います。

値をもつEvent（EchoMachine.hs）

もう少し複雑な例を見てみましょう。今までEventはすべてEvent ()でしたが、今度は()以外の値を扱ってみます。

EchoMachineという標準入力からの文字列を入力として別の値を返すEventを考えます。

type EchoMachine a = Event String -> Event a

文字列をそのまま返すecho、文字列中の文字を1文字ずつ進めるsuccEcho、文字列をひっくり返すreverseEchoはそれぞれ次のように定義できるでしょう。

echo :: EchoMachine String
echo = id

succEcho :: EchoMachine String
succEcho = fmap (map succ)

reverseEcho :: EchoMachine String
reverseEcho = fmap reverse

注目は、String -> Stringな関数をfmapでEcho String -> Echo Stringへと変換しているところです。Functorを使ってpureな関数をEvent上の関数に持ち上げています。

さらに、入力にかかわらず2秒ごとに"bot"と出力するbotEchoはこのように定義できます。

botEcho :: EchoMachine String
botEcho = const $ listE $ zip [0, 2 ..] $ repeat "bot"

BellMachineと同じようにI/Oを繋ぐ関数を用意します。mainは次のように書きます。

main :: IO ()
main = do
  let machine = mconcat [ echo, succEcho, reverseEcho, botEcho ]
  runMachine machine

走らせてみましょう。

maoe@maoe ~/coding/haskell/reactive-examples
$ ./EchoMachine 
"bot"
a
"a"
"b"
"a"
"bot"
maoe
"nbpf"
"maoe"
"eoam"
"bot"

二重引用符でくくられてない文字列が入力です。2秒ごとの"bot"という出力と、入力文字列をそれぞれ

  • そのまま返す
  • ー文字ずつ進める
  • ひっくり返す

という操作をしていることがわかります。なお内部でスレッドを使っているためか、出力順は不定になるようです。

例えば、起動後何秒たったのかを追加したかったらwithTimeEを使うようmainを書き換えます。

main :: IO ()
main = do
  let machine = mconcat [ echo, succEcho, reverseEcho, botEcho ]
  runMachine $ withTimeE <$> machine

出力は次のようになります。

maoe@maoe ~/coding/haskell/reactive-examples
$ ./EchoMachine 
("bot",0.0)
("bot",2.0)
a
("a",3.009475)
("b",3.009475)
("a",3.009475)
("bot",4.0)
maoe
("maoe",4.607139)
("nbpf",4.607139)
("eoam",4.607139)

さらに何個目のイベントかを知りたかったらcountEを足せばよいでしょう。

main :: IO ()
main = do
  let machine = mconcat [ echo, succEcho, reverseEcho, botEcho ]
  runMachine $ countE . withTimeE <$> machine

maoe@maoe ~/coding/haskell/reactive-examples
$ ./EchoMachine 
(("bot",0.0),1)
a
(("a",1.10522),2)
(("a",1.10522),3)
(("b",1.10522),4)
(("bot",2.0),5)
maoe
(("maoe",2.902198),6)
(("nbpf",2.902198),7)
(("eoam",2.902198),8)

以上のようにEventを使うことで、イベントドリブンなプログラムが高度にモジュール化され再利用可能なパーツの組み合わせとして記述できるようになることがわかります。

echoサーバ

最後に、EchoMachineを拡張してechoサーバを作ってみましょう。ちょっと例が長くなるので、最初にFRPではないechoサーバを作り、徐々にFRPを適用していきます。

通常のechoサーバ（SimpleEchoServer.hs）

ここではネットワークプログラミングについて詳しく触れません。これから作るechoサーバは次の要件を満たします。

  • TCP接続して端末に何か入力すると、入力した文字列が出力される
  • 複数のコネクションを扱える（つまり複数クライアントからの同時接続を受け付ける）
  • "quit"と入力するとコネクションを閉じる

コードにするとこのようになるでしょう。

import Control.Applicative
import Control.Concurrent
import Network
import System.IO
import Control.Monad

-- Normal Echo Server
main :: IO ()
main = withSocketsDo $
  listenOn (PortNumber 10000) >>= handler

handler :: Socket -> IO ()
handler s = forever $ do
  (h, _, _) <- accept s
  hSetBuffering h NoBuffering
  forkIO $ echo h

echo :: Handle -> IO ()
echo h = do
  text <- filter (/= '\r') <$> hGetLine h
  if text == "quit"
     then hClose h
     else hPutStrLn h text >> echo h

1コネクションに、forkIOで作られた1つのHaskellスレッドが対応するようになっています。実行してtelnetで接続してみましょう。

maoe@maoe ~
$ telnet localhost 10000
Trying 127.0.0.1...
Connected to maoe.
Escape character is '^]'.
test
test
hoge
hoge
quit
Connection closed by foreign host.

ちゃんと動いていますね。

Reactiveなechoサーバ（EchoServer1.hs）

それではreactiveを使ってFRPなechoサーバに変えてみます。

大まかな方針として

  • Eent String -> Event String: 入力文字列に応じて出力文字列を決める
  • Event String -> Event Action: 入力文字列に応じて振る舞いを変える

という2つの処理に分けて考えます。

import FRP.Reactive
import FRP.Reactive.LegacyAdapters
import Control.Applicative
import Control.Concurrent
import Control.Monad
import Data.Monoid
import Network
import System.IO

main :: IO ()
main = withSocketsDo $ listenOn (PortNumber 10000) >>= handler

handler :: Socket -> IO ()
handler s = forever $ do
  (h, _, _) <- accept s
  hSetBuffering h NoBuffering
  forkIO $ runEcho h

runEcho :: Handle -> IO ()
runEcho h = do
  (sink, event) <- makeEvent =<< makeClock
  forkIO $ forever $ hGetLine h >>= sink
  adaptE $ echoServer h event

type EchoServer = Event String -> Event Action

-- a combined reactor
echoServer :: Handle -> EchoServer
echoServer h = mconcat [ hPutStrLnE h . fmap reverse
                       , putStrLnE
                       , quitE h
                       ] . stripE

-- pure reactors
stripE :: Event String -> Event String
stripE = fmap $ filter (/= '\r')

reverseE :: Event String -> Event String
reverseE = fmap reverse

-- Action-valued reactors
quitE :: Handle -> EchoServer
quitE = fmap . quit
  where quit :: Handle -> String -> Action
        quit h "quit" = hClose h
        quit _ _      = return ()

hPutStrLnE :: Handle -> EchoServer
hPutStrLnE = fmap . hPutStrLn

putStrLnE :: EchoServer
putStrLnE  = hPutStrLnE stdout

この例ではFRPの利点をよりわかりやすく示すために、いくつかの機能を追加しています。

  • クライアント側のみでなく、サーバ側にも入力文字列を表示
  • クライアント側では入力を反転した文字列を表示

また、先に示したようにコード中のEvent処理は2種類に分けられます

  • 入力文字列から出力文字列を生成する関数
    • stripE: \rを取り除く
    • reverseE: 文字列をひっくり返す
  • 入力文字列から振る舞い（I/O）を生成する関数
    • quitE: 入力が"quit"の場合、Handleを閉じる
    • hPutStrLnE: Handleに入力文字列を書き込む
    • putStrLnE: 標準出欲に入力文字列を書き込む

注目すべき関数はechoServerです。上の追加機能を別々の関数として定義し、それらをmconcatで繋げています。同じ機能をFRPでないコードに実装しようと思ったら、echo関数の中にhPutStrLnやreverseなどを埋め込む必要があるでしょう。これはFRPによって得られるcomposability*3や再利用性の向上を端的に示している例といえます。

動かしてみましょう。先ほどと同じようにtelnetで繋いでみると、サーバ側には入力文字列が、クライアント側では入力文字列がひっくり返されたものが出力されることがわかります。

maoe@maoe ~/coding/haskell/reactive-examples
$ ./EchoServer1 
echo
hoge
quit
EchoServer1: <socket: 4>: hGetLine: invalid argument (Bad file descriptor)
maoe@maoe ~
$ telnet localhost 10000
Trying 127.0.0.1...
Connected to maoe.
Escape character is '^]'.
echo
ohce
hoge
egoh
quit
tiuq
Connection closed by foreign host.

quitしたときにエラーメッセージが出ているのは、hCloseで閉じたHandleに対してhGetLineして例外が発生しているためです。forkIOを使って同じHandleに対して同時に操作している弊害です。killThreadなどを使えばこのエラーメッセージを消すことはできますが、どうも綺麗に書く方法がよくわからないので、このまま載せておきます。safe lazy I/Oとかiteratee I/Oなどを使うとうまく書けるのかもしれません。

このコードではネットワーク処理の部分、つまりmainからhandlerに書けての部分ではFRPを使っていませんが、ここもFRPを用いて書き直すことができます。

Reactiveなechoサーバ その2（EchoServer2.hs）

Handleの生成もEventとして扱ってみます。

import FRP.Reactive
import FRP.Reactive.LegacyAdapters
import Control.Applicative
import Control.Concurrent
import Control.Monad
import Data.Monoid
import Network
import System.IO

main :: IO ()
main = runEchoServer echoServer

runEchoServer :: (Handle -> EchoServer) -> IO ()
runEchoServer e = socketServer >>= adaptE . fmap (forkIO_ . handleConnection e)

socketServer :: IO (Event Handle)
socketServer = withSocketsDo $ do
  (sink, event) <- makeEvent =<< makeClock
  sock <- listenOn (PortNumber 10000)
  forkIO $ forever $ accept sock >>= \(h, _, _) -> hSetBuffering h NoBuffering >> sink h
  return event

handleConnection :: (Handle -> EchoServer) -> Handle -> Action
handleConnection srv h = do
  (sink, event) <- makeEvent =<< makeClock
  forkIO $ forever $ hGetLine h >>= sink
  adaptE $ srv h event

type EchoServer = Event String -> Event Action

-- a combined reactor
echoServer :: Handle -> EchoServer
echoServer h = mconcat [ hPutStrLnE h . fmap reverse
                       , putStrLnE
                       , quitE h
                       ] . stripE

-- pure reactors
stripE :: Event String -> Event String
stripE = fmap $ filter (/= '\r')

reverseE :: Event String -> Event String
reverseE = fmap reverse

-- Action-valued reactors
quitE :: Handle -> EchoServer
quitE = fmap . quit
  where quit :: Handle -> String -> Action
        quit h "quit" = hClose h
        quit _ _      = return ()

hPutStrLnE :: Handle -> EchoServer
hPutStrLnE = fmap . hPutStrLn

putStrLnE :: EchoServer
putStrLnE  = hPutStrLnE stdout


-- utilities
forkIO_ :: IO () -> IO ()
forkIO_ act = () <$ forkIO act

socketServerはacceptするごとにEventが発生するアクションです。これで得られたそれぞれのHandleに対してhandleConnectionをつかってechoServerの処理を行います。この例ではネットワーク処理部分のFRP化のうまみが表れていませんが、acceptしたHandleの種類に応じてなにがしかの処理を行う場合は、FRPの恩恵を受けられるようになるかもしれません。

まとめ

ちょっと長くなりましたがreactiveのEventについてまとめました。

  • reactiveの実際の使い方についてサンプルコードで示しました
    • 単純なEventの使い方
    • インタラクティブなEventの使い方
    • EventをIOと繋げる方法（SinkとadaptE）
  • Eventを使うと、イベントドリブンなプログラムをモジュール性・再利用性が高い形で記述できることを示しました
  • FRPなネットワークプログラミングを示しました
    • 前回の冒頭で取り上げたI/Oまみれのネットワークプログラミングは、Eventを上手に使って主要部分をFRPで書いてしまえば、IOモナドの呪縛から逃れられそうに見えます

次回は連続的に変化する値を表現するBehaviorについて書きたいと思います。

参考情報

  • サンプルコードはすべてgithubにあがっています
  • reactiveのチュートリアル
    • Less Sugar/More Meat » Blog Archive » Introducing Reactive: Events
    • 数少ないreactiveのチュートリアルです。Eventを理解する上でとても役に立つ内容ですが、記事中のコードを既存のI/Oとどう結びつけるかについては全く書かれていません。実行可能なサンプルコードもないので、手元で動かすのには一苦労します。今回の基本編を書いた動機がこれです。

*1:ここでは、「イベントが起こること」を発火すると呼んでいます

*2:この例では直接的にはa -> bがMonoidになっています

*3:訳は結合性とか？可結合性？