ソースコード

STM モナドについて説明がおわった所で、カウンターを作りたいと思います。

• 子スレッドはカウンターを親から貰い、処理が終ったら +1 する。
• 親スレッドはカウンターが一定の値になるまで待つ。

という動作にします。

import Control.Concurrent
import Control.Concurrent.STM
import Control.Monad (unless, forever)

newtype Counter = Counter (TVar Int)

modifyTVar :: (a -> a) -> TVar a -> STM ()
modifyTVar f tv = readTVar tv >>= writeTVar tv . f

-- create a new counter and set its value 0.
newCounter :: IO Counter
newCounter = do tv <- newTVarIO 0
                return $ Counter tv

-- Increments the counter value.
incCounter :: Counter -> IO ()
incCounter (Counter c) = atomically $ modifyTVar (+1) c

-- Wait until the counter value becomes `n'
waitCounter :: Counter -> Int -> IO ()
waitCounter (Counter c) n = atomically $
                            do v <- readTVar c
                               unless (v == n) retry

-- forkIO with Counter.
fork :: Counter -> IO () -> IO ThreadId
fork counter act = forkIO $ act >> incCounter counter

-- threadDelay in milliseconds.
-- `delay 1000' means that this thread sleeps 1 sec at least.
delay :: Int -> IO ()
delay = threadDelay . (* 1000)

-- Do the first argument action every `time' ms infinitely.
every :: Int -> IO a -> IO ()
every time io = forever $ delay time >> io

main :: IO ()
main = do c <- newCounter
          fork c $ every 1000 $ putStrLn "a"
          fork c $ every 500 $ putStrLn "b"
          waitCounter c 2

その1から処理自体はあまり変化していません。 every , dalay の内容も同じです。

modifyTVar は TVar を変更するユーティリティーです。 よく使うと思うのですが、何故か標準モジュールにありません。

waitCounter の中で、例の retry を使っています。 カウンターの値が n でなければ処理をやりなおします。 ですが retry の特性のため、カウンターの値が変化するまで スレッドが起きることはありません。 これにより、無駄なループが起らないようになっています。

fork もユーティリティー関数です。 受け取ったアクションの後に incCounter するように合成してから forkIO をします。 ただし、このコードには問題があります。 エラーが起った時にカウンターが増えません。 サンプルなので省略していますが、エラー処理を入れないと危険です。

書いていて気付いたのですが、このサンプルだと、スレッドが永遠と終了しないので、 waitCounterの実感が沸きにくいですね…すみません。

コンパイル

-package stm を付けてください。

$ ghc -package stm -o mt mt.hs 

実行結果

マルチスレッドプログラムなので、環境によって出力が違う可能性があります。

$ ./mt
b
a
b
b
a
b
b