zeromq 100% CPU 使用率 带有 0 个侦听器的推送套接字

Question

我目前正在使用 ZeroMQ 实现多个进程之间的通信管道，所有进程都使用 Push/Pull 机制。管道从生成任务的“呼吸机”开始，这也是我的问题开始的地方：当没有连接任何工作人员时，ZeroMQ 似乎正在使用 100% 的 CPU 负载。

这是有问题的代码，它试图只发送一条消息：

module Main where

import System.ZMQ4.Monadic
import Data.ByteString.Char8 (pack)

main :: IO ()
main = do
     runZMQ $ do
            publisher <- socket Push
            bind publisher "tcp://*:10150"

            send publisher [] (pack "foo")

            close publisher

如您所见，这段代码非常简单，只是尝试将消息“foo”发送给任何订阅者。我希望这段代码在后台排队这条消息，但它似乎进入了send 命令的永无止境的循环。在套接字上设置高水位线无效。

zguide 中有一个与我想要实现的示例类似的示例：https://github.com/imatix/zguide/blob/master/examples/Haskell/taskvent.hs

在这个例子中，他们明确要求用户输入开始发送（具体来说，'当工作人员准备好时按 Enter'）——这是他们解决这个问题的方法吗？

谁能告诉我我在这里做错了什么，或者解决这个问题的最佳方法是什么？

编辑

详细来说，以下程序（带有连接的侦听器）可以完美运行：

module Main where

import System.ZMQ4.Monadic
import Data.ByteString.Char8 (pack, unpack)
import Control.Applicative ((<$>))

main :: IO ()
main = do
     runZMQ $ do
            publisher <- socket Push
            receiver  <- socket Pull

            bind    publisher "tcp://*:10150"
            connect receiver "tcp://127.0.0.1:10150"

            send publisher [] (pack "foo")    
            message <- unpack <$> receive receiver

            liftIO $ putStrLn ("received data: " ++ message)

这会按预期打印出接收到的数据。

编辑 2

使用 strace，我能够破译 zeromq 显然处于轮询/选择无限循环中：

poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
--- SIGVTALRM {si_signo=SIGVTALRM, si_code=SI_TIMER, si_pid=0, si_uid=0, si_value=0} ---
rt_sigreturn()                          = 1
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
poll([{fd=8, events=POLLIN}], 1, 0)     = 0 (Timeout)
select(9, [], [8], NULL, NULL)          = 1 (out [8])
--- SIGVTALRM {si_signo=SIGVTALRM, si_code=SI_TIMER, si_pid=0, si_uid=0, si_value=0} ---
rt_sigreturn()                          = 1

这种模式会不断重复。

Answer 1

ZeroMQ 架构不仅仅是另一个套接字包装器

ZeroMQ 系列的共同之父 Pieter Hintjens 建议的第一件事是忘记迄今为止在处理套接字时可能使用的所有内容。

这有很多原因。

首先，ZeroMQ 为您创造了一个新的、抽象的世界，您应该理解“原则上”宁愿与之和平相处而不是与之抗争——两者都是 微型COSMOS （内部机制，人们应该以某种方式尊重并与之相适应）和一个 MACRO-cosmos，它构建了一组非常强大的 Scaleable-Formal-Communication-Patterns，可以进一步利用并集成到更高阶的分布式处理系统中。

那么？

由于micro-COSMOS，您可以选择从低延迟/高性能实践中受益，而不是在代码的早期某个位置启动ZMQ Context 实例，同样assign / setup / bind / connect 想要的 ZMQ 基元元素的实例（可行的 ZMQ 基元原型的套接字 ...PUSH/PULL 在您的情况下给出），而不是设置/关闭/临时设置/关闭资源，在无限循环中越少。

ZeroMQ 实例不是一次性的，而是系统的资产。重新思考架构，micro-COSMOS 和 MACRO-cosmos 都将为您提供巨大的工作。

下一步

由于 MACRO-cosmos 规则，PUSH / PULL（分布式）形式通信模式假定消息保留在 PUSH-er 的（内部）队列中，直到 PULL-呃（不可见且不受对手代码控制）握手并取回。

这也意味着，如果你的代码试图

main :: IO ()
main = do
     runZMQ $ do
            -- PUSH side
            -- ^... beware the ZMQ uses a keyword PUB ( publish )
            --      BUT in a different pattern, very different pattern
            ...
            close publisher

代码遇到了 micro-COSMOS 内部性，其中 ZMQ 套接字的 ZMQ_LINGER（默认 == -1）参数的默认值导致close 尝试无限等待，直到任何其他进程检索到已在 PUSH-er 中排队的任何和所有未使用的消息。

仅此一点就解释了最初的反对意见和 EDIT1 中的临时观察，即在“已连接”PULL-er 的情况下，不会出现无限的 ZMQ_LINGER 等待循环。 EDIT2 也只是可视化了在低级 {select|poll} 循环中忽略这样做会导致什么。 Q.E.D.

最后说明

绝对值得花几天时间和精力通读 Pieter Hintjens 关于 ZeroMQ 的两本书。

那里有大量的宝石和最佳实践。

比在试错循环中敲几行代码要好得多

（由于您使代码陷入了主要死胡同，因此“问题”在您的代码中，而不是在 ZMQ-haskel 绑定中）

Answer 2

库作者已确认 100% CPU 使用率是一个错误。它已在此提交中修复：

https://github.com/twittner/zeromq-haskell/commit/4a6bc238dcc81cee4c8407c32edb018e371ab1e4