multithreading - 使用forkIO在Haskell "Concurrently, Asynchronously, Parallel, Non-Blocking"中进行Network.Socket编程

Question

请我想了解 GHC 的 RTS 是否同时处理阻塞读/写操作，即假设我假设的服务器只能并行处理 CPU 1000 个线程，如果所有 1000 个 forkIO 由于持久客户端套接字阻塞读/写而被阻塞，并且还有另一个需要处理 500 个请求。

Option-A> 另一个 500 请求是否必须等到 1000 forkIO 进程完成。

Option-B> Haskell 通过有效地使用所有 1000 个 CPU 线程，在内部处理(即并发、异步、并行、非阻塞)1000 + 500 个 forkIO。

仅供引用，我已经阅读了 C 和 Haskell(Network.Socket) 的许多套接字教程(和博客)，以了解 Haskell(forkIO) 和 C 的处理方式(即并发、异步、并行、非阻塞)，但我对 Haskell 是如何做到这一点并没有清楚的了解。

引用 :

https://github.com/lpeterse/haskell-socket/issues/15#issuecomment-224382491

Answer 1

请注意 forkIO产生一个绿色线程，而不是一个 O/S 线程，所以如果你产生 1000 forkIO处理同时请求的线程，这不对应于 1000 个单独的 O/S(或“CPU”)线程。

无论如何，是的，RTS 可以同时处理多个阻塞读/写调用，而不会占用 O/S 线程。事实上，默认情况下，当您编译和运行没有 -threaded 的 Haskell 程序时标志或不带-N RTS 选项，它在 中运行所有内容单 O/S 线程，甚至是你用 forkIO “ fork ”的东西，并且多个绿色线程仍然可以阻塞而不阻塞(唯一的)O/S线程。

您也永远不会在 1000 个单独的 O/S 线程上运行 Haskell 服务器。 RTS 可以在一个小的 O/S 线程池上调度数千个绿色线程，比在其自己的 O/S 线程上运行每个绿色线程更有效，因为在绿色线程之间切换不需要昂贵的 O/S 上下文切换。

您可以找到 this 2012 article about the Warp web server library有帮助。特别是，他们比较和对比了多种可能的架构:每个请求一个 CPU 线程(即您想象的设计)、事件驱动架构、每个内核一个事件处理进程的混合架构，以及轻量级的 Haskell 模型在 O/S 线程池上运行的线程。请注意，Warp 使用 Network.Socket在引擎盖下。在他们的基准测试设置中，他们使用 1-10 个 O/S 线程同时处理 1000 个客户端的请求。

如果你想要一些具体的证明 forkIO线程不会阻塞，这是一个玩具程序:

import Control.Monad
import Control.Concurrent
import Network.Socket hiding (recv)
import Network.Socket.ByteString
import qualified Data.ByteString.Char8 as BS

forks = 10

main = withSocketsDo $ do
  s <- socket AF_INET Datagram defaultProtocol
  bind s (SockAddrInet 6667 (tupleToHostAddress (127,0,0,1)))
  replicateM_ forks $ forkIO (BS.putStrLn =<< recv s 4096)
  let loop = (putStrLn =<< getLine) >> loop
  loop

如果您在没有线程的情况下编译并运行它:

$ stack ghc -- -O2 Socket.hs && ./Socket

它将产生 10 个等待 forkIO线程，然后输入 getLine/ putStrLn循环将您的输入回显给您。同时，您可以使用 netcat 或您喜欢的网络工具向等待线程发送请求:

$ echo -n 'request' | nc -uw0 localhost 6667

这也将由服务器回显。 10 次请求后，您将耗尽等待线程，它将不再响应网络请求。

然后你可以用 fork = 10000 增加线程。创建 10000 个等待线程。在他们等待的时候，主要的 getLine/ putStrLn循环将继续运行而不会出现故障。

所有这些都发生在单个 O/S 线程中，您可以通过查看 ps -Lf 来验证。管他呢。

如果同时使用多个套接字并且程序使用 -threaded 编译，是否需要更多线程的评论出现在评论中。 .以下测试程序:

import Control.Monad
import Control.Concurrent
import Network.Socket hiding (recv)
import Network.Socket.ByteString
import qualified Data.ByteString.Char8 as BS

forks = 50

main = withSocketsDo $ do
  forM_ [0..forks-1] $ \i -> forkIO $ do
    s <- socket AF_INET Datagram defaultProtocol
    bind s (SockAddrInet (6667+i) (tupleToHostAddress (127,0,0,1)))
    BS.putStrLn =<< recv s 4096
  let loop = (putStrLn =<< getLine) >> loop
  loop

在端口 6667 到 6716 上创建 50 个单独的套接字并等待它们。如果在没有线程的情况下编译，它可以毫无困难地在一个 O/S 线程中运行。如果使用线程编译并提供大于 1 的功能计数，如下所示:

$ stack ghc -- -O2 -threaded Socket.hs
$ ./Socket +RTS -N4

它似乎运行了 11 个工作线程(我认为这是一个“主线程”，四个功能，加上由 RTS 源中的常量 MAX_SPARE_WORKERS 指定的六个备用工作线程)，它们在这 50 个套接字上共享等待。

顺便说一句，这是在 Network.Socket 中完成的。代码是 recv例如，调用最终实现为:

throwSocketErrorWaitRead sock "..." $
    c_recv s (castPtr ptr) (fromIntegral nbytes) 0

与 throwSocketErrorWaitRead包装器定义为:

throwSocketErrorWaitRead :: (Eq a, Num a) => 
     Socket -> String -> IO a -> IO a
throwSocketErrorWaitRead sock name io =
    throwSocketErrorIfMinus1RetryMayBlock name
        (threadWaitRead $ fromIntegral $ fdSocket sock)
        io

和 throwSocketErrorIfMinus1RetryMayBlock记录如下:

throwSocketErrorIfMinus1RetryMayBlock
    :: (Eq a, Num a)
    => String  -- ^ textual description of the location
    -> IO b    -- ^ action to execute before retrying if an
               --   immediate retry would block
    -> IO a    -- ^ the 'IO' operation to be executed
    -> IO a

这有点复杂，但结果是包装器调用 c_recv是实际的 recv系统调用。它永远不会阻塞，因为套接字被配置为非阻塞，如果它返回一个错误代码表明它会阻塞， threadWaitRead call 用于提醒 RTS 这个绿色线程应该休眠，直到数据可供读取。