从不同线程写入套接字（Haskell 中的并发 UDP 服务器）答案

【问题标题】：Writing to a socket from different threads (a concurrent UDP server in Haskell)从不同线程写入套接字（Haskell 中的并发 UDP 服务器）
【发布时间】：2016-01-28 15:57:22
【问题描述】：

编辑：

我正在用 Haskell 编写一个 UDP BitTorrent 跟踪器。状态基于将我的数据类型 ServerState 传递到 runUDPServer、acceptConnections、handleConnection 和 handleRequestData 的 STM（两个 TVar 地图）时间>。客户端将请求启动“连接”、宣布或抓取。每次有人向服务器发送消息时，他们都应该收到一条消息。（协议在这里：http://www.rasterbar.com/products/libtorrent/udp_tracker_protocol.html）。

我将进行一些二进制解析，在 IO monad 中进行一些处理（实际上只是 STM），然后将二进制编码的消息发送回发送者。最初，我在想我可以在它自己的线程中运行这样的每个请求，但我想我可以只分叉几个线程，让它们来完成工作。一个问题可能是整个服务器（所有线程）会被 n 人以非常慢的速度发送 UDP 包阻塞（但实际上这可能是不可能的）。

我想我可以更清楚地定义我的问题：如果我只是分叉 n 个同时运行 handleConnection 的线程，那会以某种方式弄乱套接字吗？另外，（如何）理想情况下，我能否以某种方式为每个收到的数据包生成一个新线程？

我的意思是，当我分叉几个线程并写入标准输出时，输出将是乱码/从单独线程打印的内容之间的混合。 Network.accept 实际上提供了一个句柄，各个线程并不真正需要了解套接字，但我不能使用 accept。 我不会假设同时从多个线程写入套接字是安全的。

{-# LANGUAGE OverloadedStrings #-}

import Control.Exception (bracket)
import qualified Data.ByteString.Char8 as BS
import qualified Network.Socket as S hiding (send, sendTo, recv, recvFrom)
import qualified Network.Socket.ByteString as S

runUDPServer serverState port =
    S.withSocketsDo $ bracket (createSocket port) S.close (acceptConnections serverState)

    where
        createSocket port = do
            serverAddr <- fmap head $ S.getAddrInfo
                (Just (S.defaultHints {S.addrFlags = [S.AI_PASSIVE]}))
                Nothing
                (Just port)

            socket <- S.socket (S.addrFamily serverAddr) S.Datagram S.defaultProtocol
            S.bind socket $ S.addrAddress serverAddr

            return socket

        acceptConnections serverState socket = do
            handleConnection serverState socket
            acceptConnections socket

        handleConnection serverState socket = do
            (requestData, remoteAddress) <- S.recvFrom socket 2048
            responseData <- handleRequestData serverState requestData remoteAddress
            S.sendTo socket responseData remoteAddress

        handleRequestData :: ServerState -> BS.ByteString -> S.SockAddr -> IO BS.ByteString
        handleRequestData serverState requestData remoteAddress = do
            putStrLn "-----"

            putStrLn $ "Received UDP message"
            putStrLn $ "Address: " ++ show remoteAddress

            -- (left out code here)
            return "Dummy ByteString"

如果有任何提示、指点等，我将不胜感激。

【问题讨论】：

“我宁愿没有一个线程池，每个线程都运行 recvFrom，但我想我可以。我不确定如何以合理的方式从多个线程写入套接字。 "为什么不？这一切都应该在posix上是原子的，对吧？如果是这样，您可以replicateM n (forkIO $ forever $ handleConnection socket)。我不知道 UDP 套接字上的争用是否会对性能产生影响。如果是这样，让一个线程读取数据并将其穿梭到阻塞的 FIFO 队列中可能会更快。
@jberryman 您可以动态创建线程，而不是线程池，每个连接一个线程。
这个问题没有包含足够的信息来提供完整的答案。例如，您想在某个地方收集所有客户提供的信息吗？您需要由客户收集它们吗？您是否确认您当前的方法不够快/“并发”？ UDP 是否适合您的用例？
@jberryman 如果这在 POSIX 上是原子的，那是个好消息。我试试看。
@Zeta 我已经编辑了这个问题，我希望现在更清楚了。是的，我正在收集诸如 IP 地址以及实际请求之类的内容。不，我不确定我是否需要让服务器比现在更快，但是我正在做这个项目来学习如何使用 Haskell 和多线程等，所以我有点想尝试一下方法。 UDP 是我需要支持的协议。

标签： multithreading sockets haskell server udp

【解决方案1】：

兄弟你需要学习一些并发性！

您需要学习的第一件事是forkIO :: IO () -> IO ThreadId。这是所有并发开始的地方。你给它一个 IO 动作，它就会启动一个线程来运行那个 IO 动作，就像魔术师一样！一切，包括你所说的accept，都可以追溯到forkIO！由于 Haskell 数据是不可变的，所以使用起来非常安全（如果你不使用锁，没问题（然后死锁是不可能的）。）

在 Haskell 中学习并发的其余部分是如何使用forkIO，以及基于forkIO（以及一些相关原语）的库。首先，阅读Control.Concurrent。然后阅读Parallel and Concurrent Haskell（免费电子书）。要了解这本书对您的情况有何帮助，请转至Chapter 12。

Haskell much much much 在处理并发方面比任何命令式和/或不纯的语言都要好，除非专门为并发而构建（提示非Haskellers的反对意见）。拥抱它。

好的，为此，您将使用某种渠道。 MVar 实际上就足够了（只要您的写作速度比生产速度快）。见this。如果你写得比写得快，请参阅this。

stm 包具有类似的结构。

【讨论】：

正如你所说的IO，“Haskell 是不可变的，死锁是不可能的，而且要好得多”没有任何意义 - 你可以把 f+++ 提高到可怕的程度就像这里的任何其他语言一样
@Carsten 我说过如果你不使用锁，那么死锁是不可能的。当然，在 Haskell 中你可能会出错，但只有在显式使用可变变量的情况下，你才能更好地使用，因为你可以控制什么是可变的。
感谢您的链接！我只是不确定从多个线程同时写入套接字的语义。此外，您链接到的第 12 章中的示例使用 Network.accept。
@J.F.有一个线程接受连接，但随后该线程将派生新的线程来处理句柄。
@PyRulez 好吧，问题是我不能使用 Network.accept 因为它只适用于 TCP。 UDP是无连接的。但是，是的，我可以通过在 recvFrom 之后分叉来构建类似的东西。

【解决方案2】：

我个人有一个用于读取的线程，它只是在一个循环中调用recv，然后将请求扔到Chan，然后另一个用于写入，假脱机Chan。但是，我实际上确实怀疑，尽管无法确认，您可以像在您描述的体系结构中那样让多个线程直接执行此操作。我认为这可能没问题的原因是 IO 通过处理多路复用的 IO 管理器进行传输。虽然不包含最新进展，但我认为MIO paper 应该涵盖当前实现方式的基础知识。

【讨论】：