networking - 为什么我的并发 Haskell 程序过早终止？

Question

我有一个 UDP 服务器，它反射(reflect)它收到的每个 ping 消息(我认为这很有效)。我是客户端，然后我想做两件事:

确保我发送了 N(例如 10000)条消息，以及

计算正确接收响应的数量。

似乎是因为 UDP 的性质或因为 forkIO事情，我下面的客户端代码过早结束/根本不做任何计数。

我也很惊讶地看到函数 tryOnePing返回 Int 4 的 250 倍。为什么会这样？

main = withSocketsDo $ do
        s <- socket AF_INET Datagram defaultProtocol
        hostAddr <- inet_addr host
        thread <- forkIO $ receiveMessages s
        -- is there any better way to eg to run that in parallel and make sure
        -- that sending/receiving are asynchronous? 


        -- forM_ [0 .. 10000] $ \i -> do
              -- sendTo s "ping" (SockAddrInet port hostAddr)
        -- actually this would be preferred since I can discard the Int 4 that
        -- it returns but forM or forM_ are out of scope here?

        let tryOnePing i = sendTo s "ping" (SockAddrInet port hostAddr)
        pings <- mapM tryOnePing [0 .. 1000]
        let c = length $ filter (\x -> x==4) pings

        -- killThread thread
        -- took that out to make sure the function receiveMessages does not
        -- end prematurely. still seems that it does

        sClose s
        print c
        -- return()

receiveMessages :: Socket -> IO ()
receiveMessages socket = forever $ do
        -- also tried here forM etc. instead of forever but no joy
        let recOnePing i = recv socket 1024
        msg <- mapM recOnePing [0 .. 1000]
        let r = length $ filter (\x -> x=="PING") msg
        print r
        print "END"

Answer 1

这里的主要问题是当您的主线程完成时，所有其他线程都会自动终止。你必须让主线程等待 receiveMessages thread ，或者它很可能会在收到任何响应之前简单地完成。一种简单的方法是使用 MVar .

安 MVar是一个同步单元格，可以是空的，也可以只保存一个值。如果当前线程试图从空 MVar 中获取数据，它将被阻塞或插入一个完整的。
在这种情况下，我们不关心值本身，所以我们只存储一个 ()在里面。

我们将从 MVar 开始空的。然后主线程将 fork 接收者线程，发送所有数据包，并尝试从 MVar 中获取值。 .

import Control.Concurrent.MVar

main = withSocketsDo $ do
    -- prepare socket, same as before

    done <- newEmptyMVar

    -- we need to pass the MVar to the receiver thread so that
    -- it can use it to signal us when it's done
    forkIO $ receiveMessages sock done

    -- send pings, same as before

    takeMVar done    -- blocks until receiver thread is done

在接收者线程中，我们会收到所有的消息，然后放一个 ()在 MVar表示我们已完成接收。

receiveMessages socket done = do
    -- receive messages, same as before

    putMVar done ()  -- allows the main thread to be unblocked

这解决了主要问题，并且该程序在我的 Ubuntu 笔记本电脑上运行良好，但是您还需要处理一些其他事情。

sendTo不保证会发送整个字符串。您必须检查返回值以查看发送了多少，如果不是全部发送，则重试。即使是像 "ping" 这样的短消息也可能发生这种情况。如果发送缓冲区已满。

recv需要一个连接的套接字。你会想要使用 recvFrom反而。 (尽管由于某种未知原因它仍然可以在我的 PC 上运行)。

打印到标准输出是不同步的，所以你可能想要改变它，以便 MVar将用于传达接收到的数据包的数量，而不仅仅是 () .这样，您就可以完成主线程的所有输出。或者，使用另一个 MVar作为互斥锁来控制对标准输出的访问。

最后，我推荐阅读 Network.Socket 的文档。 , Control.Concurrent和 Control.Concurrent.MVar小心。我的大部分答案都是从那里找到的信息拼接在一起的。