
はいどうもー

引き続き Haskell の話題です. ちょっと Haskell で TCP ソケットを使ってみたくなったので, まず簡単なものから実装してみます.

TCP ソケットのチュートリアルといえば echo サーバですね！クライアントからの入力をそのまま返すサーバです.

せっかくなのできちんと複数クライアントとの同時通信を可能にしましょう.

# Network.Socket

Haskell で TCP ソケットを使うには, `Network.Socket`を使うようです. [Network.Socket](http://hackage.haskell.org/package/network-2.6.2.1/docs/Network-Socket.html)

ドキュメントには, 低レベル API が`Network.Socket`で, 高レベル API が`Network`と書いてあるのですが, `Network`モジュールのドキュメントには, 互換性のために残してあるけどこれから使う人は`Network.Socket`を使ってくれみたいなことが書いてあります.

適当にググると`Network`モジュールを使ったサンプルが散見されますが, ここはドキュメントにしたがって, `Network.Socket`を使用することにします.

# ソケットの用意

TCP のサーバ側は, ソケットの作成 -> ソケットをポート番号指定で bind -> 接続受付を開始(listen) -> 接続を受け付ける(accept) というステップを踏む必要があります.

というわけでまずは指定したポート番号に bind されたソケットを用意するアクションを定義します.

```
import Network.Socket

serveSocket :: PortNumber -> Socket
serveSocket port = do
    soc <- socket AF\_INET Stream defaultProtocol
    addr <- inet\_addr "0.0.0.0"
    bind soc (SockAddrInet port addr)
    return soc

```

これで引数に渡したポート番号に bind されたソケットが作成されます.

## accept

複数クライアントとの同時通信を実現するためには, `Control.Concurrent`のちからを借ります.

今回は `forkIO` を使って, 各コネクションごとにスレッドを起動していくことにします. (非同期版も作れるのかな？つくれたらつくります)

というわけで次は `accept`して`forkIO`するという処理を繰り返し行うアクションを定義します.

`forkIO` した後に実行するアクション(`echoLoop`)についてはとりあえず`undefined`とします.

Haskell の`undefined`, とても便利ですね. 型で考えるっていうスタイルが実行しやすくなっているのは, ソースコード上にトップレベル関数の型指定を書きやすい Haskell の文法と`undefined`のおかげって感じがします.

```
echoLoop :: Socket -> IO ()
echoLoop = undefined

-- import Control.Concurrent
-- import Control.Monad
-- が必要
acceptLoop :: Socket -> IO ()
acceptLoop soc = forever $ do
    (conn, \_addr) <- accept soc
    forkIO $ echoLoop conn

```

`forever :: Monad m => m a -> m b`は引数に IO アクションを受け取り, それを無限に繰り返し実行し続けます. (無限にくりかえすので返り値の型変数`b`は不定)

`forever`の引数には, `accept`して`forkIO`するアクションを渡しています.

# echo

最後にソケットから読み込み, そのまま書き出す`echoLoop`部分を作ります.

```
-- import Control.Exception が必要
echoLoop :: Socket -> IO ()
echoLoop conn = do
    sequence\_ $ repeat $ do
        (str, \_, \_) <- recvFrom soc 64
        send soc str

```

`recvFrom :: Socket -> Int -> IO (String, Int, SockAddr)`は, `recvFrom conn n` で, `conn`から最大で`n`文字まで読み込みます.
返り値は, `(読み込んだ文字列, 読み込んだ文字数, 読み込み元のアドレス??)` を返します.

そして, `send :: Socket -> String -> IO ()` でソケットに読み込んだ文字列をそのまま書き込みます.

このように, 読み込んでそのままｍ書き込むというアクションを `repeat`でつなげています. `repeat :: a -> [a]`は無限リストを作る関数です. `repeat 0`で`[0, 0, 0, 0, ...`というリストが作成されます.

このままでは `[IO ()]`型なので, これを`sequence_ :: Monad m => [m a] -> m ()`を使って１つの IO アクションにまとめ上げます.

Haskell が遅延評価だから出来る芸当ですね. 無限に繰り返す感あふれるコードになっている気がします. (`forever`使ったほうがいいと思います)

# 例外処理

`recvFrom`は相手側がコネクションを切断すると`End of file`の例外を投げます. `forkIO`しているので, １つのスレッドが例外で落ちてもサーバ全体は動き続けますが, ソケットのクローズも出来ませんし, 標準エラーになんかでてきてよろしくないので修正します.

```
-- import Control.Exceptionが必要
echoLoop :: Socket -> IO ()
echoLoop conn = do
   sequence\_ $ repeat $ do
       (str, \_, \_) <- recvFrom conn 64
       send conn str
   `catch` (\(SomeException e) -> return ())
   `finally` close conn

```

`catch`と`finally`を追加しています.

どちらも Java とかのそれと同じように動きます.

`SomeException`はすべての例外を補足することが出来ますが, ほんとはあんまり良くないですね. ここでは EOF に達した(コネクションが切断された)という場合だけを補足したいので. (どの関数がどういう場合にどんな例外を投げるのかっていうドキュメントがわからなかったのでこのままにしておきました)

そして, 例外が発生してもしなくても, 最後にかならずソケットのクローズをするよう`finally`を使います.

`SomeException`ですべての例外が捕捉出来るのって不思議じゃないですか？Haskell にはオブジェクト指向っぽい型の階層関係なんてないのに.
[Haskell の多相性 - あどけない話](http://d.hatena.ne.jp/kazu-yamamoto/20081024/1224819961)このへんが関係しているっぽいなという感じがしますが詳しいことはよくわかりませんでした...

# 全体

```
module Main where

import Network.Socket
import Control.Monad
import Control.Concurrent
import Control.Exception

main :: IO ()
main = do
    soc <- serveSocket 8080
    listen soc 5
    acceptLoop soc `finally` close soc

serveSocket :: PortNumber -> IO Socket
serveSocket port = do
    soc <- socket AF\_INET Stream defaultProtocol
    addr <- inet\_addr "0.0.0.0"
    bind soc (SockAddrInet port addr)
    return soc

acceptLoop :: Socket -> IO ()
acceptLoop soc = forever $ do
    (conn, addr) <- accept soc
    forkIO $ echoLoop conn

echoLoop :: Socket -> IO ()
echoLoop conn = do
    sequence\_ $ repeat $ do
      (str, \_, \_) <- recvFrom conn 64
      send conn str
    `catch` (\(SomeException e) -> return ())
    `finally` close conn

```

`main`内で `listen`するのを忘れずに！また, `acceptLoop`中に例外が発生してもソケットをクローズするように`finally`を使っています. (まぁプログラム終了するのでいらない気もします)

# 動作確認

`telnet`コマンドでテストします.

```
% telnet localhost 8080
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
test
test
aaa
aaa
hooooogle
hooooogle
```

ちょっとわかりづらいですが, 入力した文字列が即座にそのまま帰ってきていることがわかります. バッファリングの関係で, 一行ずつになっていますが.

# まとめ

Haskell で echo サーバ, 意外とすんなりかけましたね. 例外関係があまりよく理解できていない感じがしますが...

非同期版が気になります. 調べてみます.

---

---


Haskellでechoサーバ


どうもこんにちは.

前回([Haskell の Concurrent について調べてまとめる (IORef 編) - プログラミングのメモ帳 ➚](http://agtn.hatenablog.com/entry/2015/07/21/234658))の続きです.

今回はスレッド間協調のために`MVar`を使う方法について調べたので, まとめたいと思います.

## MVar

Haskell にかかわらず, 最近の並行処理はメッセージパッシングでやれみたいなのが流行ってますね (Scala の Akka や golang の chan など).

`MVar`は Haskell における, 容量 1 のメッセージボックスのようなものです. `MVar`を使うことで, スレッド間でメッセージのやり取りを協調的に行うことができます.

複数のスレッドが１つの`MVar`に対して, メッセージを入れたり取り出したりすることでスレッド間協調を行います.

基本となる API はこのような感じ

```
newEmptyMVar :: IO (MVar a)
newMVar :: a -> IO (MVar a)
takeMVar :: MVar a -> IO a
putMVar :: MVar a -> a -> IO ()
readMVar :: MVar a -> IO a
```

型を見ればなんとなく使い方もわかる気がしますね.

`MVar`を作るには`newEmptyMVar`か`newMVar`を使用します. `newEmptyMVar`は空のメッセージボックスを作り, `newMVar`は第一引数を初期値としてもつメッセージボックスを作ります.

`MVar`にメッセージを格納するには, `putMVar`を使います. `putMVar mvar msg` で, `msg`を`mvar`に格納します.

この際, もし`MVar`にすでにメッセージが格納されている場合, `MVar`は容量 1 のボックスなので, `putMVar`がブロックされます. 他のスレッドが`MVar`からメッセージを取り出して空にするまで待ってから, メッセージを格納します.

一方, `MVar`からメッセージを読み取るには, `takeMVar`か`readMVar`を使用します.

`takeMVar`はメッセージを読み取り, その`MVar`を空にします. `readMVar`はメッセージを読み取りますが, `MVar`の中のメッセージはそのまま残します.

ここで, `put`の時と同様に, `takeMVar`も`readMVar`も`MVar`にメッセージが格納されていなかった場合, 他のスレッドが`MVar`にメッセージを格納するまでブロックします.

というわけで簡単なサンプルコード

```
module Main where

import Control.Concurrent (forkIO, threadDelay)
import Control.Concurrent.MVar

main :: IO ()
main = do
    mvar <- newEmptyMVar
    forkIO $ do
        msg <- takeMVar mvar
        putStrLn $ "recv: " ++ msg
        threadDelay $ 1 * 10 ^ 6
        putMVar mvar "B"
    putStrLn "sleep 1"
    threadDelay $ 1 * 10 ^ 6
    putStrLn "wake up"
    putMVar mvar "A"
    takeMVar mvar >>= print
```

実行結果

```
sleep 1
wake up
recv: A
"B"
```

確かにメッセージが格納されるまで `takeMVar`がブロックしていることがわかります

## 共有変数としての MVar

さて, `MVar`にはもうひとつの使い方があります. 共有変数としての`MVar`です.

`MVar`の特徴として, 誰かが`take`してから`put`するまでの間は, 他のスレッドはだれも`MVar`の中身に触れないという点が挙げられます.

```
main = do
    mvar <- newMVar 0
    forkIO $ do
        val <- takeMVar mvar
        -- 他のスレッドはMVarの中身に触れない
        putMVar mvar $ val + 1
    ...
```

この特徴はまさにロックの特徴といえます. ロックを取得し解放するまでは, 他のスレッドは同じロックで保護された区間にははいれません.

というわけで`MVar`は型レベルでロックがついた共有変数とみなすことができますね！(このへんは Rust の Mutex に似た空気を感じます. どちらも型レベルでロックとそれが保護する中身がつながっています)

型レベルでロックがくっついているので, 中身にアクセスするには必ずロックをとる(`takeMVar`)必要があり, ロックの取得忘れがありません.

さらに, Haskell は基本的に破壊的操作があまり登場しない言語であることもこの`MVar`ロックにプラスに働きます.

例えば, 連想配列をスレッド間で共有することを考えます. また, ここでは連想配列の実装として, hashtable ではなく`Data.Map`を使用するとします(`Data.Map`は immutable な構造になっていて, lookup は O(log n)ですが, immutable なので Haskell 上で扱いやすいというメリットがあります).

`Data.Map`は immutable なので, 一度`MVar`から取得してしまえばそれ以降変更される可能性もないため, ロックを保持し続ける必要がありません. そこで, 単なる読み込みの場合は, `takeMVar`してすぐに`putMVar`で戻すだとか, `readMVar`で読み込むだけにすることで, ロックの粒度を小さくできます.

`MVar`の中身を書き換えたい場合は, 単純にロックを取得し, 書き換え後の値を`putMVar`します.

```
module Main where

import Control.Concurrent (forkIO, threadDelay)
import Control.Concurrent.MVar
import qualified Data.Map.Strict as M

main :: IO ()
main = do
    mvar <- newMVar M.empty
    forkIO $ do
        table <- takeMVar mvar
        putMVar mvar table
        -- tableを使用する操作
    forkIO $ do
         table <- readMVar mvar
         -- tableを使用する操作
    forkIO $ do
        table <- takeMVar mvar
        -- tableを変更する操作
        let newTable = ...
        putMVar mvar newTable
```

このように`MVar`と immutable なデータ構造を組み合わせることで, 粒度の小さいロックを実現することができます.

一方, `MVar`と mutable なデータ構造(`IORef`など)を組み合わせる場合は, たとえ読み込みしかしない場合であっても操作が終わるまではロックを保持しておく必要があることに注意しなければなりません (`IORef`には前回紹介したように`atomicModifyIORef`があるのでなかなかこういう状況は起こりませんね)

また, Rust の Mutex と違い, `MVar`によるロックの模倣(?)はロックの解放を自動的には行いません. したがって例外が送出された場合にロックを開放し忘れるケースがあるので, 注意が必要です.

## 一旦まとめ

というわけで今回は`MVar`について紹介しました. `MVar`でロックを実現する方に関しては, 散々言われているロックの問題点をそのまま持ってきてしまうのであまり使えないかもしれませんね...

`MVar`は容量 1 のメッセージボックスでしたが, Haskell には`Chan`というものもあります. こちらは golang の chan にかなり近いもので, 容量の制限がないキューのように働かせることができます. `Chan`のよみとり専用のスレッドを１つ立てておき, 他の複数のスレッドがタスクを`Chan`に書き込んでいくといったユースケースが考えられますね. こっちのほうが便利そうな気がしてきました.

ロックはいろいろ厄介で, デッドロックとか解放忘れとかの問題がついて回ります. それを解決する１つの方法として`STM`があるようなので, 次はそれについて調べてみようと思います.

---

---


HaskellのConcurrentについて調べてまとめる (MVar編)


こんばんは. Haskell(GHC)で並行処理を必要とするアプリケーションを書いてみようと思ったのですが, 並列処理に関するいろいろについてよくわかっていない部分が多かったので, 調べたついでにまとめておこうと思います.

もし間違い等ありましたらコメントいただけるとありがたいです

## Concurrent v.s. Parallel

Concurrent は並行, Parallel は並列と訳されます.

Concurrent は論理的に同時に実行されることで, 実際に複数のタスクが物理的に同時に実行している必要はありません. 実際どうであれ, 同時に実行しているように見えれば OK で, 複数のタスクで CPU を細かく交代で使用しながら実行していくといった実行モデルも Concurrent であるといえます.

Parallel は物理的に同時に実行されることです. 必然的に複数の CPU が必要になります. 物理的に同時に実行されているタスクは, 論理的にも同時に実行しているとみなせるので, Parallel であれば Concurrent です.

この記事では Concurrent について言及しているつもりです.

## Haskell のスレッド

Haskell は処理系実装の軽量スレッドを持ちます. OS が提供するネイティブスレッドと違い, コンテキストスイッチ(スレッドの切り替え)のオーバーヘッドが少なく, より気軽に扱えるスレッドのようです. 軽量スレッドといえば Erlang や Golang が思い浮かびますね.(Erlang は軽量プロセスっていうんでしたっけ)

実際に Haskell で軽量スレッドを立ち上げてみます.

まずはスレッドを立ち上げない場合です. (threadDelay は指定したマイクロ秒分スレッドをスリープします)

```
module Main where

import Control.Concurrent (threadDelay)

sleepN :: Int -> IO ()
sleepN n = do
    putStrLn $ "sleep " ++ show n
    threadDelay $ n * 10 ^ 6
    putStrLn $ "wake up " ++ show n

main :: IO ()
main = do
    sleepN 3
    threadDelay $ 2 * 10 ^ 6
    putStrLn "sleep 2 and wakeup"
    threadDelay $ 2 * 10 ^ 6
    putStrLn "end"
```

実行結果

```
sleep 3
wake up 3
sleep 2 and wakeup
end
```

全体として, 3 秒->2 秒->2 秒とスリープするので 7 秒ほどの実行時間になります.

つぎにスレッドを立ち上げる場合です

Haskell でスレッドを立ち上げるには `forkIO :: IO () -> IO ThreadId` を使用します. `IO ()`を渡すと, それを新しく立ち上げたスレッド上で実行してくれます.

(`forkOS :: IO () -> IO ThreadId` というものもありますが, こちらは Haskell の軽量スレッドではなく, ネイティブスレッドを立ち上げます)

```
module Main where

import Control.Concurrent (forkIO, threadDelay)

sleepN :: Int -> IO ()
sleepN n = do
    putStrLn $ "sleep " ++ show n
    threadDelay $ n * 10 ^ 6
    putStrLn $ "wake up " ++ show n

main :: IO ()
main = do
    forkIO $ sleepN 3
    threadDelay $ 2 * 10 ^ 6
    putStrLn "sleep 2 and wakeup"
    threadDelay $ 2 * 10 ^ 6
    putStrLn "end"
```

実行結果

```
sleep 3
sleep 2 and wakeup
wake up 3
end
```

１つのスレッドが 3 秒スリープしている間に, もう一つのスレッドのスリープが始まるので, 全体で 4 秒ほどの実行時間になります.

## 共有変数

複数スレッドによる並行実行を扱うと, どうしても共有変数的なものが欲しくなる場合があります. Haskell でスレッド間共有をしたい場合はいくつかの方法があるようです.

もっとも直感的(手続きプログラミング出身者にとって)で馴染みやすいのは `Data.IORef` かと思います. `IO` の世界の内側でのみ読み書きができる"変数"です.

まずは単一スレッドで実際に使ってみます.(以後 import などは省略します)

```
add1 :: IORef Int -> IO ()
add1 v = do
    modifyIORef v (+1)

main :: IO ()
main = do
    ref <- newIORef 0
    v <- readIORef ref
    print v
    add1 ref
    v' <- readIORef ref
    print v'
```

実行結果

```
0
1
```

このように変数として中身を書き換えることができます.

これは変数なので, ひとつのスレッドで行った書き換えが他のスレッドにも影響を及ぼします.(State モナドのように変数を模倣しているだけではこれはできない)

```
add1 :: IORef Int -> IO ()
add1 v = modifyIORef v (+1)

spawn :: IORef Int -> IO ()
spawn ref = do
    forkIO $ add1 ref
    return ()

main :: IO ()
main = do
    ref <- newIORef 0
    spawn ref
    spawn ref
    spawn ref
    threadDelay 1000000
    v <- readIORef ref
    print v
```

## データ競合

一方これを複数スレッドで並列に動かすことを考えます. `modifyIORef` はアトミックではないので,

```
v の中身を読む
v の中身 + 1 を計算する
v にその結果を入れる

```

というそれぞれの計算の間に別のスレッドでの計算が割り込まれる可能性がある.

上の例で, `spawn ref >> spawn ref >> spawn ref` という部分は複数スレッドから一つの変数を同時に変更しようとしている. そのため, 変更が競合し意図しない動作になる可能性がある.

IORef で競合を防ぐ方法としては `atomicModifyIORef :: IORef a -> (a -> (a, b)) -> IO b` を使用する方法がある.

`atomicModifyIORef` の第 2 引数は `a -> (a, b)` である. これは `IORef` の中身を引数にとって, `(変更後の値, atomicModifyIORefの返り値にしたい値)` を返す関数である.

```
inc :: IORef Int -> IO Int
inc ref = atomicModifyIORef ref (\n -> (n + 1, n))

main :: IO ()
main = do
    ref <- newIORef 0
    res <- inc ref
    v <- readIORef ref
    print res
    print v
```

`inc` は C 言語の`i++;`のような動きをする. 加算する前の値を返し, 変数をインクリメントする.

`atomicModifyIORef` は名前の通り atomic な操作であり, 分割不可能になるため他のスレッドと処理が競合することがなくなる.

## 一旦まとめ

長くなってきた & 疲れてきたので一旦きります.

今回はスレッド間共有変数のために `IORef` を使用し, その変更に `atomicModifyIORef` を使用することでデータ競合を防ぐ方法を紹介した.

`MVar` や `STM` を使用する方法もあり, そっちのほうが良い場合もあるっぽいのでそっちについてもまとめたいと思います.

---

---


↗ agatan blog ↗

#Haskell

Haskellでechoサーバ

HaskellのConcurrentについて調べてまとめる (MVar編)

HaskellのConcurrentについて調べてまとめる (IORef編)