multithreading - 使用 TChan 进行类似管道的操作

Question

我想在两个线程之间实现一个管道。我有线程 A 获取数据、处理数据并将其发送到线程 B。我有一个 MVar 检查数据是否已完全处理

但是，我有一个异常*** Exception: thread indefinitely blocked in an STM transaction

为什么我的线程被阻塞了？我虽然比当第一个线程写入 channel 时，当 channel 上有数据时，第二个线程可以读取它

fstPipe :: (a -> b) -> TChan b -> MVar () -> [a] -> IO ()
fstPipe f chIn m xs = do
    ( mapM_(\x-> atomically $ writeTChan chIn $ f x) xs) >> putMVar m ()

pipelineDone channel mIn = do
    isDone <- fmap isJust $ tryTakeMVar mIn
    isEmpty <- atomically $ isEmptyTChan channel
    return $ isDone && isEmpty

lastPipe f chIn mIn = iter 
    where iter = do
        atomically $ fmap f $ readTChan chIn
        isDone <- pipelineDone chIn mIn
        unless isDone $ iter

pipeline = do
    chIn <- atomically newTChan
    m <- newEmptyMVar
    first <- async $ fstPipe reverse chIn m $ replicate 10 [1..500]
    last <- async $ lastPipe print chIn m
    wait first
    wait last

Answer 1

在同一个代码块中使用 STM 和 信号量对我来说似乎很奇怪......为什么不在 STM 中完成所有事情？

特别是，为什么不是 TChan (Maybe x)，Nothing 表示序列的结尾？

另外，请注意您的 fstPipe 可能只是生成一堆未计算的 thunk 并立即将它们放入 TChan，而实际上没有计算任何东西。您可能需要一个 seq 或类似的东西来强制在该线程上进行一些实际的工作。