haskell - MonadBaseControl:如何解除ThreadGroup

Question

在模块threads中的Control.Concurrent.Thread.Group包中，有一个函数forkIO：

forkIO :: ThreadGroup -> IO α -> IO (ThreadId, IO (Result α))

我想使用 monad-control中的 MonadBaseControl解除它。这是我的尝试：

fork :: (MonadBase IO m) => TG.ThreadGroup -> m α -> m (ThreadId, m (Result α))
fork tg action = control (\runInBase -> TG.forkIO tg (runInBase action))

这是错误消息：

Couldn't match type `(ThreadId, IO (Result (StM m α)))'
              with `StM m (ThreadId, m (Result α))'
Expected type: IO (StM m (ThreadId, m (Result α)))
  Actual type: IO (ThreadId, IO (Result (StM m α)))
In the return type of a call of `TG.forkIO'
In the expression: TG.forkIO tg (runInBase action)
In the first argument of `control', namely
  `(\ runInBase -> TG.forkIO tg (runInBase action))'

进行哪些更改以使类型匹配？

Answer 1

主要问题是IO a的forkIO参数。要在m a中派生IO操作，我们需要一种将m a运行到IO a的方法。为此，我们可以尝试使类具有runBase :: MonadBase b m => m a -> b a方法的monad，但是很少有有趣的转换器可以提供。例如，如果考虑使用StateT变压器，那么它首先可以找到如何使用runStateT在基本monad中运行某些内容的方法，前提是它首先有机会观察其自身的状态。

runFork :: Monad m => StateT s m a -> StateT s m (m b)
runFork x = do
    s <- get
    return $ do
        (a, s') <- runStateT x s
        return a

这建议使用类型 runForkBase :: MonadBase b m => m a -> m (b a)，我们将在下面的类型类中使用它。

{-# LANGUAGE MultiParamTypeClasses #-}
{-# LANGUAGE FunctionalDependencies #-}

import Control.Monad.Base

class (MonadBase b m) => MonadRunForkBase b m | m -> b where
    runForkBase :: m a -> m (b a)

我在名称中添加了 Fork一词，以强调未来状态更改通常不会在两个未来之间共享。因此，可以提供 WriterT的少数有趣的转换器（例如 runBase）仅提供无趣的 runBase；它们产生的副作用是无法观察到的。

对于 fork实例提供的有限形式的降低，我们可以为任何内容编写类似 MonadRunForkBase IO m的内容。我要从基址到 lift普通的 forkIO，而不是从 threads的那个，您可以使用相同的方法。

{-# LANGUAGE FlexibleContexts #-}

import Control.Concurrent

forkInIO :: (MonadRunForkBase IO m) => m () -> m ThreadId
forkInIO action = runForkBase action >>= liftBase . forkIO

执行个体

这就提出了一个问题：“我们可以为 MonadRunForkBase实例提供哪些变压器？”马上，我们可以轻松地为具有 MonadBase实例的任何基本monad提供它们

import Control.Monad.Trans.Identity
import GHC.Conc.Sync (STM)

instance MonadRunForkBase [] [] where runForkBase = return 
instance MonadRunForkBase IO IO where runForkBase = return
instance MonadRunForkBase STM STM where runForkBase = return
instance MonadRunForkBase Maybe Maybe where runForkBase = return
instance MonadRunForkBase Identity Identity where runForkBase = return

对于变压器，逐步建立这样的功能通常更容易。这是可以在紧邻的monad中运行fork的转换器的类别。

import Control.Monad.Trans.Class

class (MonadTrans t) => MonadTransRunFork t where
    runFork :: Monad m => t m a -> t m (m a)

我们可以提供一个默认实现，用于在基础环境中一直运行

runForkBaseDefault :: (Monad (t m), MonadTransRunFork t, MonadRunForkBase b m) =>
                      t m a -> t m (b a)
runForkBaseDefault = (>>= lift . runForkBase) . runFork

这使我们可以分两个步骤完成 MonadRunForkBase的 StateT实例。首先，我们将使用上面的 runFork来创建一个 MonadTransRunFork实例

import Control.Monad
import qualified Control.Monad.Trans.State.Lazy as State

instance MonadTransRunFork (State.StateT s) where
    runFork x = State.get >>= return . liftM fst . State.runStateT x

然后，我们将使用默认值提供一个 MonadRunForkBase实例。

{-# LANGUAGE FlexibleInstances #-}
{-# LANGUAGE UndecidableInstances #-}

instance (MonadRunForkBase b m) => MonadRunForkBase b (State.StateT s m) where
    runForkBase = runForkBaseDefault

我们可以对 RWS做同样的事情

import qualified Control.Monad.Trans.RWS.Lazy as RWS

instance (Monoid w) => MonadTransRunFork (RWS.RWST r w s) where
    runFork x = do
        r <- RWS.ask
        s <- RWS.get
        return $ do 
            (a, s', w') <- RWS.runRWST x r s
            return a

instance (MonadRunForkBase b m, Monoid w) => MonadRunForkBase b (RWS.RWST r w s m) where
    runForkBase = runForkBaseDefault

MonadBaseControl

与我们在前两节中开发的 MonadRunForkBase不同， monad-control中的 MonadBaseControl并没有基于“未来状态通常不会在两个期货之间共享”的假设。 MonadBaseContol和 control努力通过使用 restoreM :: StM m a -> m a从控制结构中的分支还原状态。对于base的 forkIO来说，这并不存在问题；使用 forkIO是 MonadBaseControl文档中提供的示例。对于 forkIO from threads，这将是一个小问题，因为返回了额外的 m (Result a)。

我们想要的 m (Result a)实际上将作为 IO (Result (StM m a))返回。我们可以摆脱 IO并用 m替换为 liftBase，而剩下 m (Result (StM m a))。我们可以将 StM m a转换成 m a来恢复状态，然后用 a返回 restoreM，但是它被卡在 Result ~ Either SomeException中。 Either l是函子，因此我们可以在其中的任何地方应用 restoreM，将类型简化为 m (Result (m a))。 Either l也是 Traversable，对于任何 Traversable t，我们始终可以将其替换为 Monad或 Applicative中的 sequenceA :: t (f a) -> f (t a)。在这种情况下，我们可以使用特殊用途的 mapM，它是 fmap和 sequenceA的组合，只有 Monad约束。这将给出 m (m (Result a))，而 m将通过在Monad中的联接或仅使用 >>=来展平。这引起了

{-# LANGUAGE FlexibleContexts #-}

import Control.Concurrent
import Control.Concurrent.Thread
import qualified Control.Concurrent.Thread.Group as TG

import Control.Monad.Base
import Control.Monad.Trans.Control

import Data.Functor
import Data.Traversable
import Prelude hiding (mapM)

fork :: (MonadBaseControl IO m) =>
        TG.ThreadGroup -> m a -> m (ThreadId, m (Result a))
fork tg action = do
    (tid, r) <- liftBaseWith (\runInBase -> TG.forkIO tg (runInBase action))    
    return (tid, liftBase r >>= mapM restoreM)

当我们在原始线程中运行 m (Result a)时，它将把状态从分支线程复制到原始线程，这可能会很有用。如果要在读取 Result之后恢复主线程的状态，则需要首先捕获它。 checkpoint将捕获整个状态，并返回一个操作来还原它。

checkpoint :: MonadBaseControl b m => m (m ())
checkpoint = liftBaseWith (\runInBase -> runInBase (return ()))
             >>= return . restoreM

一个完整的示例将显示两个线程对状态的处理方式。两个线程都从 fork发生时获取状态，而无需考虑修改另一个线程中的状态。当我们等待主线程中的结果时，主线程中的状态将设置为来自派生线程的状态。我们可以通过运行 checkpoint创建的操作来获取主线程的状态。

import Control.Monad.State hiding (mapM)

example :: (MonadState String m, MonadBase IO m, MonadBaseControl IO m) => m ()
example = do    
    get >>= liftBase . putStrLn
    tg <- liftBase TG.new
    (_, getResult) <- fork tg (get >>= put . ("In Fork:" ++)  >> return 7)
    get >>= put . ("In Main:" ++) 
    revert <- checkpoint
    result <- getResult
    (liftBase . print) result
    get >>= liftBase . putStrLn
    revert
    get >>= liftBase . putStrLn

main = do
    runStateT example "Initial"
    return ()

这个输出

Initial
Right 7
In Fork:Initial
In Main:Initial