MonadBaseControl:如何提升ThreadGroup

Question

在模块Control.Concurrent.Thread.Group的threads包中，有一个函数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

主要问题是forkIO的IO a参数。要在IO中派生m a操作，我们需要一种方法来运行到IO a的m a。为此，我们可以尝试创建具有runBase :: MonadBase b m => m a -> b a方法的monads类，但很少有有趣的转换器可以提供该方法。例如，如果我们考虑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；它们产生的副作用永远不会被观察到。

我们可以为任何具有MonadRunForkBase IO m实例提供的有限降低形式的东西编写类似于fork的东西。我将从base中lift普通的forkIO，而不是从threads中，你可以用同样的方法。

{-# LANGUAGE FlexibleContexts #-}

import Control.Concurrent

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

实例

这就提出了一个问题，“我们可以为哪些转换器提供MonadRunForkBase实例”？我们可以直接为任何具有MonadBase实例的基础monads提供它们

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

对于transformers来说，像这样一步一步地构建功能通常更容易。下面是可以在紧随其后的monad中运行分支的转换器类。

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

这使我们可以在两个步骤中完成StateT的MonadRunForkBase实例。首先，我们将使用上面的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恢复控制结构中的分支状态。这对于从基础开始的forkIO来说不是问题；使用forkIO就是MonadBaseControl文档中提供的一个示例。这对于forkIO from threads来说是一个小问题，因为返回了额外的m (Result a)。

我们想要的m (Result a)实际上将以IO (Result (StM m a))的形式返回。我们可以去掉IO，代之以带有liftBase的m，只剩下m (Result (StM m a))。我们可以将StM m a转换为恢复状态的m a，然后使用restoreM返回a，但它被困在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