この記事は ACCESS Advent Calendar 2015 19日目の記事です。
ACCESSの @asukamirai です。
FRP（Functional reactive programming）をサーバのロジックとして使うのはどうなのだろう、ということで試しに感覚を掴むためにやってみました。
使用したFRPのライブラリは、ICFP2015に登場した論文のFRPNow。言語はHaskellです。
確認環境など
- stack 0.1.10.1
- lts-3.18
- frpnow-0.18
- OS: Windows7 64bit
動機
FRPというと、なんとなくGUI的なものを扱うようなイメージが強かったのですが、その一方で、データフローでプログラミングを行うというのは、ストリーミング処理系のライブラリとも共通する部分がありそうだと思っていました。
先月行われた関数型ストリーム処理勉強会でも、ほんのわずかですがFRPという言葉が聞こえた気がします。
では、ストリーミング処理系をよく使いそうな場所のひとつであるサーバのロジックとしてFRPは役に立ったりするんだろうか、という疑問がはじまりです。
今回使用したFRPNowも基本的にはGUIを想定しているようで、サーバと組み合わせるためのコードは自作する必要がありました。サーバはwarpを使用しています。
なお、FRPの存在自体はだいぶ以前から知っていたのですが、それなりに手を出したのは今回が初めてです。
作成してみたもの
動作
数を数えるだけのサーバです。
- http://localhost:3000/start を叩くと、1秒毎に内部の数が1ずつあがりはじめます。
- http://localhost:3000/stop を叩くと、動作を停止します。
- http://localhost:3000/add/XXX (XXXは数字)を叩くと数字を内部の数字に加算することができます。
- http://localhost:3000/get を叩くと内部の数字をレスポンスとして返します。
- http://localhost:3000/shutdown を叩くと、内部のFRP処理を終了します。以降他のコマンドが動作しなくなることが確認できます。
実験目的なのでAPI設計はとても適当です。
コード
作成したコードはgithubにあります。あまりちゃんとは整備していませんが。（以下に貼り付けてあるコードとは僅かに違う部分があります）
2つのファイルに分かれています。
{-# LANGUAGE MultiWayIf #-}
{-# LANGUAGE OverloadedStrings #-}
module FRPNow.Counter where
import qualified Control.Concurrent as CC
import qualified Control.Concurrent.Async as AS
import qualified Control.Concurrent.Chan as Chan
import qualified Control.Concurrent.MVar as MV
import qualified Control.Exception as E
import qualified Control.FRPNow as FRP
import Control.FRPNow (Behavior, Event, EvStream, Now)
import qualified Control.Monad as M
type InputIF a = a -> IO ()
data Command =
StartCount
| StopCount
| AddCount Int
| GetCount (InputIF Int)
| Shutdown
instance Show Command where
show StartCount = "StartCount"
show StopCount = "StopCount"
show (AddCount a) = "AddCount " ++ show a
show (GetCount _) = "GetCount"
show Shutdown = "Shutdown"
getGetCount :: Command -> Maybe (InputIF Int)
getGetCount (GetCount inIF) = Just inIF
getGetCount _ = Nothing
getCountState :: Command -> Maybe Bool
getCountState StartCount = Just True
getCountState StopCount = Just False
getCountState _ = Nothing
getAddCount :: Command -> Maybe Int
getAddCount (AddCount x) = Just x
getAddCount _ = Nothing
getShutdown :: Command -> Maybe ()
getShutdown Shutdown = Just ()
getShutdown _ = Nothing
-- | カウンターをセットアップする。
setupCounter :: EvStream Command -> Now (Event ())
setupCounter inputStream = do
-- inputStreamを４つのEvStreamに分ける
-- StartCount/StopCountはTrue/Falseに変換してひとつのEvStreamに。
let addCountStream = getAddCount `FRP.filterMapEs` inputStream
countStateStream = getCountState `FRP.filterMapEs` inputStream
getCountStream = getGetCount `FRP.filterMapEs` inputStream
shutdownStream = getShutdown `FRP.filterMapEs` inputStream
-- 秒カウントのEvStreamを生成
secStream <- genSecStream
-- countStateStreamの最後のイベントの値を表すBehaviorを作成
countingState <- FRP.sampleNow $ False `FRP.fromChanges` countStateStream
-- countStateがTrueの間だけ、秒カウントするEvStreamを作成
let secCountStream = secStream `FRP.during` countingState
-- addCountStreamとsecCountStreamのイベントをマージしたEvStreamを作成
let countUpStream = secCountStream `FRP.merge` addCountStream
-- countUpStreamの内容を合算した状態を表すBehaviorを作成
countState <- FRP.sampleNow $ FRP.foldEs (+) 0 countUpStream
-- GetCountイベントに、そのときのcounterの値を応答させる
respondToGetCount countState `FRP.callStream` getCountStream
-- counterの値が変化するたびに、その値をコンソールに出力する
(putStrLn . ("count " ++) . show) `FRP.callIOStream` FRP.toChanges countState
-- 最初に来たShutdownイベントを取り出して返す
-- このイベントが届くとFRPの実行が終了する
FRP.sampleNow $ FRP.next shutdownStream
where
-- | GetCountコマンドに応答する
respondToGetCount :: Behavior Int -> [InputIF Int] -> Now ()
respondToGetCount countState inIFs = do
-- counterの値を採取する。
count <- FRP.sampleNow countState
-- コンソールに表示してGetCountイベントに応答
FRP.sync $ do
putStrLn $ "respond to GetCount : " ++ show count
-- 「同時」に届いたGetCountイベントはリストになって渡されるので全てに同じ値を渡す
M.forM_ inIFs ($ count)
-- | １秒ごとにイベント（内容はIntの１）を発行するEvStreamを生成する。
-- 終了をサポートしていないので、放置するとリークする。今は気にしない。
genSecStream :: Now (EvStream Int)
genSecStream = do
(evs, emitEv) <- FRP.callbackStream
FRP.sync . M.void . AS.async . M.forever $ do
emitEv 1
CC.threadDelay 1000000
return evs
-- | 入力EvStreamを受け取るNowモナドを実行する。
-- 返り値は、入力EvStreamへの入力アクション。
runNow :: (EvStream a -> Now (Event x)) -> IO (InputIF a)
runNow now = do
-- frpnow側が多数のスレッドからの同時イベント発行に対応していない
-- ように見えるので、チャネルを設けてタイミング制御を行っている。
-- が、これで正しく動く保証はない。
mvEmitEv <- MV.newEmptyMVar
mvWait <- MV.newEmptyMVar
M.void . AS.async . FRP.runNowMaster $ do
(inputStream, emitEv) <- FRP.callbackStream
M.void . FRP.sync $ mvEmitEv `MV.putMVar` emitEv
MV.putMVar mvWait `FRP.callIOStream` inputStream
now inputStream
emitEv <- MV.takeMVar mvEmitEv
chan <- Chan.newChan
M.void . AS.async . E.handle handleException . M.forever $ do
a <- Chan.readChan chan
emitEv a
M.void $ MV.takeMVar mvWait
return $ Chan.writeChan chan
where
handleException (E.SomeException _err) = return ()
runCounter :: IO (InputIF Command)
runCounter = runNow setupCounter
だいたいの流れはコメントに書いています。
setupCounterがメインのロジックです。最初に実行されて回路のようなものが構築され、以降はその回路が動く、というイメージです。
詳しくおいかけていくとイベントのストリームを組み合わせたりして処理を作っているのがなんとなくわかるんじゃないかと思います。
runNowはFRPライブラリのrunNowMasterに入力のイベントストリームを渡す手段を提供するための仕組みを加えたようなものです。返り値のInputIFにCommandを与えると、それが構築した回路の入力になります。サーバ以外の用途でも、外から入力を受け取りたい場合には使いまわせる関数です。
コメントにも書いてある通り、これでいつでもちゃんと動くという確証はないのでお気をつけください。
{-# LANGUAGE OverloadedStrings #-}
module Main where
import qualified FRPNow.Counter as FRP
import qualified Network.Wai as Wai
import qualified Network.Wai.Handler.Warp as Warp
import qualified Network.HTTP.Types.Status as Status
import qualified Control.Concurrent.MVar as MVar
import qualified Data.ByteString.Lazy.Char8 as LBS
import Data.Monoid ((<>))
import qualified Data.Text as TXT
import System.Timeout (timeout)
main :: IO ()
main = do
let port = 3000
let setting = Warp.setPort port Warp.defaultSettings
putStrLn $ "start server port=" ++ show port
input <- FRP.runCounter
Warp.runSettings setting $ simpleApp input
simpleApp :: FRP.InputIF FRP.Command -> Wai.Application
simpleApp input req respond = do
let path = Wai.pathInfo req
putStrLn $ "----- " ++ show path
case path of
["add", numTxt] -> do
let numStr = TXT.unpack numTxt
num = read numStr
numBS = LBS.pack numStr
input $ FRP.AddCount num
respond $ Wai.responseLBS Status.status200 [] $ "AddCount " <> numBS <> " accepted"
["start"] -> do
input FRP.StartCount
respond $ Wai.responseLBS Status.status200 [] "StartCount accepted"
["stop"] -> do
input FRP.StopCount
respond $ Wai.responseLBS Status.status200 [] "StopCount accepted"
["shutdown"] -> do
input FRP.Shutdown
respond $ Wai.responseLBS Status.status200 [] "Shutdown accepted"
["get"] -> do
mv <- MVar.newEmptyMVar
input $ FRP.GetCount $ MVar.putMVar mv
mcount <- timeout 1000000 $ MVar.takeMVar mv
case mcount of
Just count -> respond . Wai.responseLBS Status.status200 [] . LBS.pack $ show count
Nothing -> respond $ Wai.responseLBS Status.status200 [] "timeout"
_ -> do
respond $ Wai.responseLBS Status.status200 [] "not supported command"
こちらはCounter.hsで実装したコードとサーバをくっつける部分のコードです。
runCounterを実行し、そこにサーバへのリクエストに応じたCommandを投げているだけです。
感想
この内容だとそれなりにきれいに書けた気がしますが、タイミング制御をきちんとやろうとするととたんに面倒になります。
例えば上記のコードでは、秒をカウントするイベントストリームは、内部的にはstartのコマンドを与える以前から動作し続けていて、そのイベントストリームのイベントがstart状態でのみ通るようになっています。
startコマンドを受け取った時点からきっちり1秒を数え始めようと思うと、コマンドを受け取った段階で初めて秒カウントのイベントストリームを生成し、stopコマンドが着たら破棄して、次のstartでまたイベントストリームを作り直す、ということが必要になります。実のところこれをするコードも試しに書いてみたのですが、非常に見難いコードになってしまいました。もしかするとFRPに慣れればうまく書けるのかもしれませんが。
また、そんなにたいした量のコードではありませんが、書くのにけっこうな時間がかかっています。
ドキュメント（例：EvStream）を見ても、どういうときにどういう関数を使えばいいのか一見しただけではわからないものも多く、慣れるまではだいぶ時間がかかりそうに感じました。
関数がもっと体系化されていればわかりやすいのかもしれません。この観点ではFRPNowではなく、 Sodiumあたりがうまくまとまっていそうなイメージをもっていますが、私は触れていません。
さて、では当初の疑問である「サーバのロジックとして使えるのか」というと、正直これだけだとまだなんともいえません。こういうものがうまくハマる場所はあるかもしれませんが、その場合でも慣れるまでに時間がかかる覚悟を持っておいたほうがよさそうな感触を持っています。また、他のライブラリの使用も検討したほうがよさそうです。
ACCESS Advent Calendar、明日は @_mirer さんです。