python - pyzmq:套接字中断为 "ZMQError: Operation cannot be accomplished in current state"

Question

我对 ZeroMQ 非常陌生，并试图构建一个非常基本的消息传递系统。该代码很大程度上基于来自 here 的示例。有一些曲折

出于某种原因，在最后一条消息到达 frontend 之后socket ( hbm , tx ) 代码抛出错误，我不确定它的来源

抱歉，代码/输出很长，我觉得强调问题所在并帮助调试很重要

下面是我的代码+输出:

s_frontend = "ipc://frontend.ipc"
s_backend  = "ipc://backend.ipc"

import time

def hbm():
    """Worker task, using a REQ socket to do load-balancing."""
    socket = zmq.Context().socket(zmq.REQ)
    socket.identity = u"hbm".encode("ascii")
    socket.connect(s_frontend)
    # Tell broker we're ready for work
    socket.send(b"READY")
    while True:
        msgs = socket.recv_multipart()
        print("hbm got something", msgs)
        if "BE READY" in msgs:
            print("hbm:BE is ready for some work")
        time.sleep(3)

def tx():
    """Worker task, using a REQ socket to do load-balancing."""
    socket = zmq.Context().socket(zmq.REQ)
    socket.identity = u"txtx".encode("ascii")
    socket.connect(s_frontend)
    # Tell broker we're ready for work
    socket.send(b"READY")
    while True:
        msgs = socket.recv_multipart()
        print("tx got something ", msgs)
        if "BE READY" in msgs:
            print("tx:BE is ready for some work")
        time.sleep(3)


def dev():
    """Worker task, using a REQ socket to do load-balancing."""
    socket = zmq.Context().socket(zmq.REQ)
    socket.identity = u"dev".encode("ascii")
    socket.connect(s_backend)
    # Tell broker we're ready for work
    socket.send(b"READY")


def main():
    # Prepare context and sockets
    context = zmq.Context.instance()
    frontend = context.socket(zmq.ROUTER)
    frontend.bind( s_frontend )                 # ( "ipc://frontend.ipc" )
    backend = context.socket(zmq.ROUTER)
    backend.bind( s_backend )                   # ( "ipc://backend.ipc" )

    def start(task, *args):
        process = multiprocessing.Process(target=task, args=args)
        process.daemon = True
        process.start()

    start(hbm)
    start(tx)


    time.sleep(1)
    start(dev)

    clients = []
    poller = zmq.Poller()
    poller.register(backend, zmq.POLLIN)
    #poller.register(frontend, zmq.POLLIN)
    all_is_ready = False
    while True:
        sockets = dict(poller.poll(timeout=1))
        #print(sockets)
        soc = None
        if backend in sockets:
            print("got something from backend")
            msg = backend.recv_multipart()
            print(msg)
            print("adding frontend to poller")
            poller.register(frontend, zmq.POLLIN)
            print("backend is ready, notify frontend")

        elif frontend in sockets:
            print("got something from frontend")
            msg = frontend.recv_multipart()
            print(msg)
            clients.append(bytes(msg[0]))

        elif len(clients) == 2 and all_is_ready is False:
            all_is_ready = True
            for c in clients:
                print("sending response to", c)
                time.sleep(0.1) # just to prevent print overlap
                frontend.send_multipart([c, b"", b"BE READY"])
        else:
            print("so much work, no rest, sleeping for 3")
            time.sleep(3)


    # Clean up
    backend.close()
    frontend.close()
    context.term()


if __name__ == "__main__":
    main()

运行此代码会产生以下输出:

so much work, no rest, sleeping for 3
got something from backend
['dev', '', 'READY']
adding frontend to poller
backend is ready, notify frontend
got something from frontend
['hbm', '', 'READY']
got something from frontend
['txtx', '', 'READY']
sending response to hbm
sending response to txtx
hbm got something ['BE READY']
hbm:BE is ready for some work
tx got something  ['BE READY']
tx:BE is ready for some work
so much work, no rest, sleeping for 3
Process Process-1:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/vm/repo/pyzmq_examples/hal_example/example1.py", line 104, in hbm
    msgs = socket.recv_multipart()
  File "/usr/local/lib/python2.7/dist-packages/zmq/sugar/socket.py", line 475, in recv_multipart
    parts = [self.recv(flags, copy=copy, track=track)]
  File "zmq/backend/cython/socket.pyx", line 791, in zmq.backend.cython.socket.Socket.recv
  File "zmq/backend/cython/socket.pyx", line 827, in zmq.backend.cython.socket.Socket.recv
  File "zmq/backend/cython/socket.pyx", line 191, in zmq.backend.cython.socket._recv_copy
  File "zmq/backend/cython/socket.pyx", line 186, in zmq.backend.cython.socket._recv_copy
  File "zmq/backend/cython/checkrc.pxd", line 25, in zmq.backend.cython.checkrc._check_rc
    raise ZMQError(errno)
ZMQError: Operation cannot be accomplished in current state
so much work, no rest, sleeping for 3
Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/process.py", line 258, in _bootstrap
    self.run()
  File "/usr/lib/python2.7/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "/home/vm/repo/pyzmq_examples/hal_example/example1.py", line 130, in tx
    msgs = socket.recv_multipart()
  File "/usr/local/lib/python2.7/dist-packages/zmq/sugar/socket.py", line 475, in recv_multipart
    parts = [self.recv(flags, copy=copy, track=track)]
  File "zmq/backend/cython/socket.pyx", line 791, in zmq.backend.cython.socket.Socket.recv
  File "zmq/backend/cython/socket.pyx", line 827, in zmq.backend.cython.socket.Socket.recv
  File "zmq/backend/cython/socket.pyx", line 191, in zmq.backend.cython.socket._recv_copy
  File "zmq/backend/cython/socket.pyx", line 186, in zmq.backend.cython.socket._recv_copy
  File "zmq/backend/cython/checkrc.pxd", line 25, in zmq.backend.cython.checkrc._check_rc
    raise ZMQError(errno)
ZMQError: Operation cannot be accomplished in current state
so much work, no rest, sleeping for 3
Traceback (most recent call last):
  File "/home/vm/repo/pyzmq_examples/hal_example/example1.py", line 244, in <module>
    main()
  File "/home/vm/repo/pyzmq_examples/hal_example/example1.py", line 233, in main
    time.sleep(3)
KeyboardInterrupt

进程以退出代码 1 结束

Answer 1

Q : "... throws error ... I'm not sure of it's origin "

ZMQError: Operation cannot be accomplished in current state

起源是在每个用例中，其中 REQ -Archetype 被使用，代码不符合原生 API 条件:

...
socket = zmq.Context().socket( zmq.REQ ) #--------------------- REQ socket Archetype
socket.identity = u"hbm".encode( "ascii" )
socket.connect( s_frontend )
socket.send( b"READY" ) #-------------------------------------- REQ.send()-s
while True:             # ..................................... REQ next can .recv()
    #      socket-FSA( of a type of REQ ) can execute a .recv() iff.send() preceded.....?
    msgs = socket.recv_multipart() #- - - - - - - - - - - - - - REQ.recv_multipart()-s
    #.......................................................... REQ next can .send()
    continue                       #-?-?-?-?-?-?: DID IT TRY TO REQ.send()? NO, NEVER...!

Do [you] mean by that that REQ must do iterative recv send and cannot do any other way ? (send recv recv) – LordTitiKaka

REQ -Archetype 对保留的强制性顺序有一个硬性的期望:
.send() - .recv() - .send() - .recv() - .send() - .recv() - ...无限循环
任何违反此触发器/触发器更改其内部有限状态自动机 (FSA) 内部状态的行为都会导致上述 ZMQError: Operation cannot be accomplished in current state