在下面的代码中,我尝试将 2 个生产者合并为 1 个。所有生产者都具有相同的类型。它们将由在单独线程中运行的 2 个输入Producer
中的每一个组合而成,并由将值放入 unagi chan 的 Consumer
消耗(我是使用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
。
t1 <- lift . fork . ...
在您的函数中,fork 。 ...
的类型为 m ThreadId
,但 do
block 位于 monad Producer a m
中。
此外,lifted-base
有一个更新的fork
。
关于haskell - 使用 Control.Concurrent.MonadIO 进行管道和 fork ,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46125030/