Haskell、通道、STM、线程、消息传递

我正在尝试使用 Channels/STM 在 Haskell 中实现消息传递。也许这是一个糟糕的想法，并且有更好的方法在 Haskell 中实现/使用消息传递。如果是这种情况，请告诉我；然而，我的探索提出了一些关于并发 Haskell 的基本问题。

我听说过有关 STM 的伟大事情，特别是在 Haskell 中的实现。由于它支持读取和写入，并且具有一些安全优势，因此我认为应该从这里开始。这提出了我最大的问题：

msg <- atomically $ readTChan chan

其中 chan 是 TChan Int，导致等待通道上有值吗？

考虑以下程序：

p chan = do
    atomically $ writeTChan chan 1
    atomically $ writeTChan chan 2

q chan = do
    msg1 <- atomically $ readTChan chan 
    msg2 <- atomically $ readTChan chan
    -- for testing purposes
    putStrLn $ show msg1 
    putStrLn $ show msg2

main = do
    chan <- atomically $ newTChan
    p chan
    q chan

使用 ghc --make -threaded 编译它，然后运行该程序，实际上您会得到 1 后跟 2 打印到控制台。现在，假设我们这样做

main = do 
    chan <- atomically $ newTChan
    forkIO $ p chan 
    forkIO $ q chan

反而。现在，如果我们使用 -threaded，它要么什么也不打印，要么打印 1，要么打印 1 后跟 2 到终端；但是，如果不使用 -threaded 进行编译，它总是打印 1 后跟 2。 问题 2：-threaded 和非 -threaded 之间有什么区别？我想它们并不是真正作为并发事物运行，它们只是一个接一个地运行。这与下面的内容是一致的。

现在，在我的想法中，如果我让 p 和 q 同时运行；也就是说，我对它们进行了分叉，它们应该能够以相反的顺序运行。假如

main = do
    chan <- atomically newTChan
    forkIO $ q chan
    forkIO $ p chan

现在，如果我在没有 -threaded 的情况下编译它，我永远不会在控制台上打印任何内容。如果我使用 -threaded 进行编译，有时会这样做。尽管如此，获得 1 后又获得 2 的情况非常罕见——通常只有 1 或什么都没有。我也用 Control.Concurrent.Chan 尝试过这一点，并得到了一致的结果。

第二个大问题：通道和分叉如何一起发挥作用，上面的程序中发生了什么？

无论如何，看来我不能这么天真地用STM来模拟消息传递。也许 Cloud Haskell 是解决这些问题的一个选择——我真的不知道。任何有关如何在序列化~~>写入套接字~~>从套接字读取~~>反序列化的情况下传递消息的任何信息都将受到极大的赞赏。

不，你的想法是对的——这就是什么TChans 代表 - 你只是错过了一个小点forkIO:

问题是你的主线程不会等待用创建的线程终止forkIO (请参阅此处以供参考 https://hackage.haskell.org/package/base-4.8.0.0/docs/Control-Concurrent.html#g:12)

所以如果我使用hint参考文献中给出：

import Control.Concurrent
import Control.Concurrent.STM

p :: Num a => TChan a -> IO ()
p chan = do
    atomically $ writeTChan chan 1
    atomically $ writeTChan chan 2

q chan = do
    msg1 <- atomically $ readTChan chan 
    msg2 <- atomically $ readTChan chan
    -- for testing purposes
    putStrLn $ show msg1 
    putStrLn $ show msg2

main :: IO ()
main = do
    children <- newMVar []
    chan <- atomically $ newTChan
    _ <- forkChild children $ p chan
    _ <- forkChild children $ q chan
    waitForChildren children
    return ()

waitForChildren :: MVar [MVar ()] -> IO ()
waitForChildren children = do
  cs <- takeMVar children
  case cs of
    []   -> return ()
    m:ms -> do
      putMVar children ms
      takeMVar m
      waitForChildren children

forkChild :: MVar [MVar ()] -> IO () -> IO ThreadId
forkChild children io = do
  mvar <- newEmptyMVar
  childs <- takeMVar children
  putMVar children (mvar:childs)
  forkFinally io (\_ -> putMVar mvar ())

它按预期工作：

d:/Temp $ ghc --make -threaded tchan.hs
[1 of 1] Compiling Main             ( tchan.hs, tchan.o )
Linking tchan.exe ...
d:/Temp $ ./tchan.exe 
1
2
d:/Temp $

当然，如果您将呼叫切换到，它会继续工作p and q too