使用 Control.Concurrent.MonadIO 进行管道和分叉答案

【问题标题】：pipes and fork with Control.Concurrent.MonadIO使用 Control.Concurrent.MonadIO 进行管道和分叉
【发布时间】：2017-09-08 21:57:18
【问题描述】：

在下面的代码中，我试图将 2 个Producers 合并为 1 个。所有这些都具有相同的类型。它们将由 2 个输入 Producers 中的每一个在单独的线程中运行并被 Consumer 消耗，后者将值放入 unagi chan（我使用 unagi chan 来提高性能）。返回一个从 chan 读取的生产者。我希望能够使用任何Producer，只要它可以运行IO，所以我将monad的类限制为MonadIO和HasFork。

import Pipes (Producer, Consumer, (>->), yield, await, runEffect)
import Control.Concurrent.Chan.Unagi (InChan, OutChan, newChan, readChan, writeChan)
import Control.Monad (forever, void)
import Control.Concurrent.MonadIO (MonadIO, HasFork, liftIO, fork)

combine :: (MonadIO m, HasFork m) => Producer a m r -> Producer a m r -> Producer a m r
combine p1 p2 = do
  (inChan, outChan) <- liftIO $ newChan
  t1 <- fork . void . runEffect $ p1 >-> (consumer inChan)
  t2 <- fork . void . runEffect $ p2 >-> (consumer inChan)
  producer outChan

producer :: (MonadIO m) => OutChan a -> Producer a m r
producer outChan = forever $ do
  msg <- liftIO $ readChan outChan
  yield msg

consumer :: (MonadIO m) => InChan a -> Consumer a m r
consumer inChan = forever $ do
  msg <- await
  liftIO $ writeChan inChan msg

但是我收到以下错误：

    • Couldn't match type ‘m’
                 with ‘Pipes.Internal.Proxy Pipes.Internal.X () () a m’
  ‘m’ is a rigid type variable bound by
    the type signature for:
      combine :: forall (m :: * -> *) a r.
                 (MonadIO m, HasFork m) =>
                 Producer a m r -> Producer a m r -> Producer a m r
    at src/Pipes/Unagi.hs:8:12
  Expected type: Pipes.Internal.Proxy
                   Pipes.Internal.X () () a m GHC.Conc.Sync.ThreadId
    Actual type: m GHC.Conc.Sync.ThreadId

我正在尝试做的事情可能吗？

【问题讨论】：

在fork前面添加lift。 fork stuff :: m ThreadId，lift :: m r -> Producer a m r。虽然我不认为有很多有趣的类型可以实际派生。
很好用。你能解释一下它为什么起作用吗？也许您应该发布一个答案，以便我将其标记为正确并获得更多积分？
另外，我想使用fork 的原因是，只要有 IO，我就可以使用任何 monad 堆栈。没关系IO 是您唯一可以使用的类型fork

标签： haskell monad-transformers haskell-pipes

【解决方案1】：

在fork前面添加lift。

t1 <- lift . fork . ...

在您的函数中，fork . ... 的类型为 m ThreadId，但 do 块位于 monad Producer a m 中。

另外，lifted-base 有一个更新的fork。

【讨论】：

啊，是的，最初我尝试使用提升异步来执行此操作，但我无法解决。我找不到任何关于我如何“运行”MonadBase 的信息。或者我该如何运行(MonadIO m, MonadBase m) => Producer a m r？
MonadBaseControl IO m 就像HasFork m。没有可应用的“运行”功能。