PR

orElseを使った代替処理の記述

 前節では,ミューテックスを使い,「他のスレッドによって危険領域が実行されている間,処理を止めて待つ」という制御を行いました。しかし,場合によっては処理を止めてしまうより,別の代替処理を行ったり,処理自体を行わないほうがよいこともあります。

 例えば,後で取り消される可能性がある処理の場合,すべての部分で愚直に実行待ちをするよりも,処理が必要ないとわかった時点で処理を取りやめたほうが賢明でしょう。特に,実時間(realtime,リアルタイム)の処理が必要なアプリケーションでは,個々の処理を完遂することよりも,望んだときに望んだ反応をすぐに返すことのほうが重要です。必ずしも必要ではない処理なのに,それが終わってないからといってずっと他の処理が待たされるようなプログラムを書いてはプログラマ失格でしょう。

 STMは,こうした柔軟性を提供するための機能として,「ある処理が行えないときに,その代替となる処理を行うプログラム」を記述するためのorElse関数を用意しています。

Prelude Control.Concurrent.STM> :t orElse
orElse :: STM a -> STM a -> STM a

 orElseを使えば,あるSTMアクションとその代替となる複数のSTMアクションを合成して,一つのSTMアクションを作成できます。こうして合成したアクションは,第1引数のアクションが成功せずに再試行が必要になったとき,orElseを使って第2引数に連結した別のアクションを実行します。そのアクションも失敗した場合には,アクション全体を再試行します。

 …と言葉だけで説明してもわからないかもしれません。実際にorElseを使ってみましょう。I/O処理がロックされていたら,そのまま何もせずに終了する処理は,以下のように記述できます。

write n str res = replicateM_ n $ nonBlockAct n (writeFile text str) res

nonBlockAct :: Int -> IO a -> TVar Lock -> IO ()
nonBlockAct n doAct res = do
    randomDelay
    act <- atomically $
      (do
          v <- readTVar res
          check (not $ isLock v)
          lock res
          return True)
       `orElse`
       (return False)
    when act
      (do
          doAct
          atomically $ unlock res)

 whenはContorl.Monadで定義されている関数で,第1引数にTrueが与えられた場合には第2引数のアクションを実行し,そうでないときには何もしない関数です。Falseが与えられたときにしか処理をしないunlessという逆バージョンの関数があります。

when              :: (Monad m) => Bool -> m () -> m ()
when p s          =  if p then s else return ()

unless            :: (Monad m) => Bool -> m () -> m ()
unless p s        =  if p then return () else s

 nonBlockActでは,orElseを使って処理が成功した場合にはTure,処理がロックされている場合にはFalseを返すようにしています。これにより,処理がロックされている場合には当該アクションを実行しないというプログラムを定義しています。

 それでは,writeのwriteFile text strの部分をprint strに置き換えて実行してみましょう。実際に実行される処理が減少しているのがわかると思います。

*Main> :main
"IO 1"
"IO 2"
"IO 1"
"IO 1"
"IO 2"
"IO 2"

 また,orElseの第2引数をretryに置き換えると,全体の処理が再実行されるため,blockActと同じ動作になるのがわかります。

*Main> :main
"IO 1"
"IO 2"
"IO 2"
"IO 1"
"IO 1"
"IO 2"
"IO 2"
"IO 1"
"IO 2"
"IO 1"
"IO 2"
"IO 1"

 同様に,`orElse` retryを間に挟んだ場合には,再試行時にすぐさまretryが呼ばれ,2度目の再試行時には2番目のorElseの第2引数である3番目の処理が行われるため,`orElse` retryを挟まない場合と同じ動作になるのがわかるでしょう。

nonBlockAct n doAct res = do
    randomDelay
    act <- atomically $
      (do
          v <- readTVar res
          check (not $ isLock v)
          lock res
          return True)
       `orElse` retry
       `orElse`
       (return False)
    when act
      (do
          doAct
          atomically $ unlock res)

*Main> :main
"IO 1"
"IO 2"
"IO 1"
"IO 2"
"IO 1"
"IO 1"

 この例は処理をすぐに取り消すだけのものですが,応用することで様々な複雑な処理を実現できます。例えば,「I/O処理がロックされていてすぐに実行できない場合にはタスク・リストに登録し,その処理を行うための条件が満たされなくなったり,別の処理によって取り消された場合にはタスク・リストから除去する」といった仕組みを実現できるでしょう。

 そして,そうした複雑な処理を記述するときにこそ,デッドロックが起こらないためモジュール性(modularity)が高いSTMの真価が発揮されると思います。

 最後に,STMアクションの再試行とorElseの間に成り立つ法則を示しておきます。

  1. M1 `orElse` (M2 `orElse` M3) = (M1 `orElse` M2) `orElse` M3
  2. retry `orElse` M = M
  3. M `orElse` retry = M

 この法則とretryの動作そのものを併せて考えると,第4回 「取り出し可能な値」のみを持つListモナドで説明したMonadPlusの法則が成り立っていることがわかります。このため,retryをmzero,orElseをmplusにしたMonadPlusクラスのインスタンスも定義されています。

instance MonadPlus STM where
  mzero = retry
  mplus = orElse

著者紹介 shelarcy

 2007年6月15日に発売された「Beautiful Code: Leading Programmers Explain How They Think」という本を買いました。この本には,前回と今回で扱ってきた並行プログラミングやSTMについて書いている"Beautiful concurrency"という論文が収録されています。

 この論文やソースコードはWebで公開されているので,自分の知識を試すために読んでみるのもよいかもしれません。STM (IO ( ))やIO (IO ( ))の内部のIOが実行されないことを利用したり,foldを使ってorElseを組み合わせるような少々技巧的なコードになっていますが,その分,新しい発見があって面白いと思います。今回使用したrandomDelayは,この論文から持ってきました。

 なお,この論文で取り上げられているサンタクロース問題に対して,いくつかの別解や別の言語による回答がWebで公開されています(参考リンク)。それらと見比べてみるのも面白いかもしれませんね。