GHC (Haskell) では、軽量スレッドを利用することで並行処理のプログラムを非常に手軽に記述できます。しかし、この強力な機能を素直に利用すると、思わぬ落とし穴にはまることがあります。

特に注意が必要なのが、Control.Concurrent モジュールの killThread :: ThreadId -> IO () や、 Control.Concurrent.Async モジュールの cancel :: Async a -> IO ()x の利用です。これらは対象の軽量スレッドに対して非同期例外を発生させ、実行中の IO アクションを強制的に中断させます。

本記事では、ネットワークサーバーのような「入出力を伴うセッションの実行」を例に取り、ナイーブな非同期例外による中断がどのような問題を引き起こすのかを確認し、その危険性を回避するためのアイデアを紹介します。

なお、この記事で例に挙げている ghci で対話的に試すことができるコードは、Github-Gist*1 にアップロード済みです。

ナイーブな非同期例外が引き起こす問題

次のような、入力の Mock と入出力セッションの実行器（sessionRunner）を考えてみます。この実行器は、入力が終わると入出力アプリケーションの IO アクションを 非同期例外で終了 させてしまうという特徴を持っています。

import Control.Concurrent
import Control.Concurrent.Async
import Control.Concurrent.STM
import Control.Applicative
import Control.Monad

import Data.IORef

import Data.Set (Set)
import qualified Data.Set as Set

msleep :: Int -> IO ()
msleep mills = threadDelay (mills * 1000)

data Input
  = Req Int
  | EndOfInput
  deriving (Eq, Show)

data Output
  = Resp Int
  deriving (Eq, Show)

-- 入力の Mock
genGetInput :: Int -> IO (IO Input)
genGetInput count = do
  ref <- newIORef 0
  return $ do
    i <- succ <$> readIORef ref
    if i <= count
      then writeIORef ref i >> return (Req i)
      else return EndOfInput

-- 入出力のセッションを実行する
sessionRunner
  :: (IO Input -> IO ())
  -> ((Output -> IO ()) -> IO ())
  -> IO ()
sessionRunner receiver sender = do
  getInput0 <- genGetInput 10
  done <- newEmptyMVar
  let getInput = do
        inp <- getInput0
        case inp of
          EndOfInput -> putMVar done ()
          Req {}     -> return ()
        return inp

  r <- async $ receiver $ do
    msleep 10
    getInput

  s <- async $ sender $ \_out -> do
    return ()

  takeMVar done
  -- 入力が終わると非同期に終了してしまう
  cancel s
  cancel r

そして、この実行器で動かすアプリケーションのコードとして、以下のような実装を用意します。リクエストを受け取り、少し遅延を入れてからレスポンスを返す処理を並列に実行する、単純な処理です。

getApplication :: IO ( IO Input -> IO ()
                     , (Output -> IO ()) -> IO ()
                     , IO ())
getApplication = do
  pendings       <- newTVarIO Set.empty               :: IO (TVar (Set Int))
  results        <- newTQueueIO                       :: IO (TQueue Output)
  (eof, setEof)  <- do eofVar <- newEmptyTMVarIO      :: IO (TMVar ())
                       return (readTMVar eofVar, putTMVar eofVar ())

  let idispatch (Req n)     = do
        atomically (modifyTVar' pendings (Set.insert n))
        msleep ((n `rem` 5 + 1)  * 10)
        atomically (writeTQueue results (Resp n))
      idispatch  EndOfInput = atomically setEof
      runI inp = void $ forkIO (idispatch inp)

      receiver getInput = loop
        where loop = do
                inp <- getInput
                case inp of
                  Req {}     -> runI inp >> loop
                  EndOfInput -> runI inp

      runO out@(Resp n) = do
        print out
        atomically (modifyTVar' pendings (Set.delete n))
      -- sender の終了条件の判定
      done = do
        eof                                       -- 入力の終わりに到達 かつ
        guard . Set.null =<< readTVar pendings  -- 処理中のリクエストが無い
      sender send = loop
        where loop = do
                next <- atomically $
                        done *> return (return ())  -- 終了条件が成立するまではループを抜けない
                        <|>
                        do out <- readTQueue results
                           return (runO out >> send out >> loop)
                next

  let printResult = atomically (readTVar pendings) >>= \p ->  putStrLn ("pendings: " ++ show (Set.toList p))
  return (receiver, sender, printResult)

このアプリケーションのアクションを、実行器 sessionRunner にそのまま渡した場合どうなるでしょうか。

example1 :: IO ()
example1 = do
  (receiver, sender, printResult) <- getApplication
  sessionRunner receiver sender
  printResult

実行結果は以下のようになります。

ghci>  example1
Resp 1
Resp 2
Resp 5
Resp 3
Resp 6
Resp 4
Resp 7
pendings: [8,9,10]

入力が終了した時点で sessionRunner が即座に cancel を呼ぶため、アプリケーション側の sender が強制的に中断されてしまいます。その結果、バックグラウンドで処理中だった Req 8, Req 9, Req 10 の処理が完了せずに残ってしまいました。

解決策: 境界スレッドと STM による切り離し

このように、実行器がアプリケーションの IO アクションに対して直接非同期例外を投げてくるナイーブな設計は、処理の取りこぼしなどのバグを生みやすくなります。

そこで、この問題を回避するためのアーキテクチャを提案します。方針は 実行器に渡す IO アクションと、実際のアプリケーションの IO アクションを切り離す ことです。 (そして、この切り離された IO アクションは別のスレッドとなります)

具体的には、sessionRunner に渡された IO アクションが中断されることを前提とした上で、実行器と直接やり取りするための別のアクション（境界スレッド用の breceiver, bsender）を定義します。そして、アプリケーション本体との情報の受け渡しは、非同期例外の影響を受けない STM 変数（TQueue など）経由で行うようにします。

example2 :: IO ()
example2 = do
  input  <- newTQueueIO  :: IO (TQueue Input)
  -- sessionRunner に渡すための境界スレッド用入力 IO action
  let breceiver getInput = loop
        where loop = do
                inp <- getInput
                case inp of
                  Req {}     -> atomically (writeTQueue input inp) >> loop
                  EndOfInput -> atomically (writeTQueue input inp)
  output <- newTQueueIO  :: IO (TQueue Output)
  -- sessionRunner に渡すための境界スレッド用出力 IO action
  let bsender send = loop
        where loop = do
                send =<< atomically (readTQueue output) :: IO ()
                loop
  asess <- async $ sessionRunner breceiver bsender

  (receiver, sender, printResult) <- getApplication

  arecv <- async $ receiver (atomically (readTQueue input))
  asend <- async $ sender (\out -> atomically (writeTQueue output out))

  wait asess
  wait arecv
  wait asend

  printResult

実行結果は以下の通りです。

ghci> example2
Resp 1
Resp 2
Resp 5
Resp 3
Resp 6
Resp 4
Resp 7
Resp 10
Resp 8
Resp 9
pendings: []

sessionRunner にアプリケーションの IO アクションを直接渡さなくなったため、アプリケーション側は sessionRunner からの非同期例外の影響を直接受けなくなりました。不意の中断が起こらなくなったことで、すべてのリクエスト処理が安全に完了するようになっています。

DNSフルリゾルバ bowline での改善

筆者が開発している DNS フルリゾルバ bowline*2 においても、これと全く同じ構造の課題に直面しています。

bowline が利用している TLS や HTTP2 のトランスポートライブラリは、セッションの終了時に非同期例外を発行する構造になっています。 DoT*3 や DoH*4 の IO アクションに対して非同期例外が発行されることになるため、ライブラリ側の終了条件と、bowline 側のセッションの終了条件が異なることがあると、難しい不具合が起きやすい構造を抱えてしまっています。

今後、紹介したような「非同期例外を境界スレッドで切り離す」パターンを適用することで、このセッション管理の堅牢性を改善していく予定です。