I am very new to ZeroMQ and trying to construct a very basic messaging system. The code is very much based on examples from here with some twists
For some reason following a last message arriving to a frontend socket (hbm ,tx) the code throws errors that I'm not sure of it's origin
Sorry for the long code/output I felt it is important to stress what is the issue and help in debug
Below is my code + output :
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()
Running this code results in a following output :
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
Process finished with exit code 1
Q : "... throws error ... I'm not sure of it's origin "
ZMQError: Operation cannot be accomplished in current state
The origin is that in each use-case, where REQ-Archetype was used, the code was not meeting the native API conditions :
...
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 has a hardwired expectation of a mandatory sequencing of being kept :
.send() - .recv() - .send() - .recv() - .send() - .recv() - ... ad infinitum
Any violation of this flip/flop-changing its internal Finite-State-Automaton (FSA) internal-state causes the said ZMQError: Operation cannot be accomplished in current state