python - ZeroMQ 轮询线程安全

Question

我有一个正在 ZMQ 轮询器上轮询的线程:

poller.poll(timeout)

该线程也是通过在轮询器中注册的套接字接收和发回消息的线程。

然后我有另一个线程，最终可能会创建一个新的套接字并注册它以轮询输入事件:

socket = context.socket(...)
socket.bind/connect(...)
poller.register(socket, zmq.POLLIN)

一旦套接字被注册，后面的线程就不会再碰它。

这样安全吗？

更新

我得到的答案/评论是关于我不应该这样做的。或者哪些是指南的建议(我已经知道)。但这并没有真正回答我的问题。

更具体地说，我正在使用 ZeroMQ 的 pyzmq Python 绑定(bind)。

现在，虽然 ZeroMQ 套接字不是线程安全的，但确实可以将它们从一个线程传输到另一个线程，只要在传输过程中存在完整的内存屏障。

所以第一个问题是:我需要在那里设置一个显式的内存屏障吗？请注意，有一个线程创建并绑定(bind)/连接套接字，然后注册它，但它不会再次使用该线程。有实际冲突吗？有什么时候我应该明确阻止两个线程访问套接字吗？

那么第二个问题是:在轮询器中注册套接字是线程安全的吗？大多数时候，执行轮询的线程正忙于做其他事情，但它也可能正在轮询等待超时。在那种情况下，我是否需要使用锁来防止对轮询器的并发访问？或者在另一个线程轮询时在轮询器中注册新套接字是否安全？

更新二

我正在使用 Pyro4处理和配置远程进程(即:它们的 ZeroMQ 连接及其行为)。可以使用 Pyro Proxy 完成初始配置非常容易。但是，当我启动该过程时，我实际上是在使用一个一直运行的专用线程 ( Pyro oneway call ) 运行主循环，但是如果我再次使用 Pyro 代理访问该对象，那么此访问来自另一个线程。

所以这个想法是避免修改远程对象的类，但仍然允许使用 Pyro 来配置远程对象，即使它们正在运行。只要新套接字的创建+绑定(bind)/连接+注册对于另一个线程是安全的，我就很好。

Answer 1

Once the socket is registered, the latter thread will not touch it again. Is this safe?

没有。

不仅需要安全解决方案，而且还向供应商一方输出实际证明稳定和有保证的系统行为的责任的行业(无论是由于明智的祖父、对 QA/TQM 的坚定信念还是由于强加的法规)在 MIL/GOV/aerospace/healthcare/pharma/automotive 等分割供应商管理上)只会直接拒绝。

为什么？

“……不会再碰了。”只是一个 promise 。

经过安全交叉验证的系统设计不会满足于避免碰撞的证明。

---

让我引用 Pieter HINTJENS 的一本可爱的书“Code Connected, Vol.1” - ZeroMQ 的必读文章:

Some widely used models, despite being the basis for entire industries, are fundamentally broken, and shared state concurrency is one of them. Code that wants to scale without limit does it like the Internet does, by sending messages and sharing nothing except a common contempt for broken programming models.

You should follow some rules to write happy multithreaded code with ØMQ:

• Isolate data privately within its thread and never share data in multiple threads. The only exception to this are ØMQ contexts, which are threadsafe.
• Stay away from the classic concurrency mechanisms like as mutexes, critical sections, semaphores, etc. These are an anti-pattern in ØMQ applications.
• Create one ØMQ context at the start of your process, and pass that to all threads that you want to connect via inproc sockets.
• Use attached threads to create structure within your application, and connect these to their parent threads using PAIR sockets over inproc. The pattern is: bind parent socket, then create child thread which connects its socket.
• Use detached threads to simulate independent tasks, with their own contexts. Connect these over tcp. Later you can move these to stand-alone processes without changing the code significantly.
• All interaction between threads happens as ØMQ messages, which you can define more or less formally.
• Don’t share ØMQ sockets between threads. ØMQ sockets are not threadsafe. Technically it’s possible to migrate a socket from one thread to another but it demands skill. The only place where it’s remotely sane to share sockets between threads are in language bindings that need to do magic like garbage collection on sockets.

If you need to start more than one proxy in an application, for example, you will want to run each in their own thread. It is easy to make the error of creating the proxy frontend and backend sockets in one thread, and then passing the sockets to the proxy in another thread. This may appear to work at first but will fail randomly in real use. Remember: Do not use or close sockets except in the thread that created them.

If you follow these rules, you can quite easily build elegant multithreaded applications, and later split off threads into separate processes as you need to. Application logic can sit in threads, processes, or nodes: whatever your scale needs.

ØMQ uses native OS threads rather than virtual “green” threads. The advantage is that you don’t need to learn any new threading API, and that ØMQ threads map cleanly to your operating system. You can use standard tools like Intel’s ThreadChecker to see what your application is doing. The disadvantages are that native threading APIs are not always portable, and that if you have a huge number of threads (in the thousands), some operating systems will get stressed.

---

如果您要跨线程共享套接字，不要。它会导致随机异常和崩溃。

---

We could assume "light" conditions: system not stressed, high-watermark never reached, no big congestions. There is just a single thread running the application (polling and executing tasks on input). So most of the time (99.99%) there is no concurrency. Now, concurrency only occurs when a second thread appears just to add a socket to the pool. There will never be more than 2 threads being executed. And the second thread will be always restricted to adding new sockets to the pool (once added the socket is transferred to the main thread). Is this enough for boundary conditions? – Peque

在update-II中添加的原理图用例细节越多，专业的解决方案就不会浪费时间，并且应该通过使用线程清理设计来避免任何隐患。

#T1 a poller-maintainer -has Context() instance control
                        -has graceful .close() + .term() responsibility
                        -has POLLER instance under it's own control
                        -has PAIR  .bind(    "inproc://worker2poller" )
                        -has PAIR  .recv() <add_socket>-request processing responsibility

#T2 a worker-process:   -has PAIR  .connect( "inproc://worker2poller" )
                        -has PAIR  .send() privilege to ask T1 to add a socket & include it into POLLER

---

虽然 GIL 无论如何都避免了找到 python 线程运行的机会 PARALLEL ，纯 OOP 设计是保持架构清晰和职责分离并保持正式通信模式完全可扩展的动力。