Pyzmq: Can't timeout a REQ/REP connection

[hrunting]

If I try to connect to non-existent socket, I can't timeout that connection. Is there no way to do this? I am writing some test code and tests won't complete if the service I'm testing (which may be broken) isn't available. Any way to implement a timeout? Here is a test case that shows the problem (requires two Ctrl-C's to exit). The alarm appears to fire, but strace shows a secondary thread repeatedly trying to connect, poll, close.

import signal
import sys
import zmq

def alarm_handler(signal, frame):
  raise IOError("Timeout processing auth request")

previous_alarm_handler = signal.signal(signal.SIGALRM, alarm_handler)

try:
  signal.alarm(5)
  context = zmq.Context()
  socket = context.socket(zmq.REQ)
  socket.connect("tcp://127.0.0.1:12346")
  signal.alarm(0)

  signal.alarm(10)
  socket.send_json({"msg": "testmsg"})
  msg = socket.recv_json()
  signal.alarm(0)

except Exception, e:
  raise e

finally:
  signal.signal(signal.SIGALRM, previous_alarm_handler)
  socket.close()
  context.term()

sys.exit(0)

[minrk]

Short answer: add socket.setsockopt(zmq.LINGER, 0), and you should be set.

You need to set LINGER to something other than -1 (the default) if you want the socket to stop trying to send. The gist is that you should be able to do:

    socket.send('foo')
    socket.close()

and close() will block until the socket is done sending messages. Since your peer never arrives, the socket keeps waiting. LINGER is the amount of time (in ms) that it will block. The default is -1, or forever. setsockopt(LINGER, 0) means close() will drop everything and return right away.

You don't need any of the alarm bits, so you should be able to reduce your test to:

import sys
import zmq

# None of these operations will block, regardless of peer:
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.setsockopt(zmq.LINGER, 0)
socket.connect("tcp://127.0.0.1:12346")
socket.send_json({"msg": "testmsg"}) # send can block on other socket types, so keep track
# use poll for timeouts:
poller = zmq.Poller()
poller.register(socket, zmq.POLLIN)
if poller.poll(10*1000): # 10s timeout in milliseconds
    msg = socket.recv_json()
else:
    raise IOError("Timeout processing auth request")

# these are not necessary, but still good practice:
socket.close()
context.term()

sys.exit(0)

See the zeromq linger doc for more.

[hrunting]

Thanks minrk, that works perfectly. I knew about poll, but since that wasn't working either without linger, I reduced it even further.

Could the linger information be added to the pyzmq documentation. Currently, socket.setsockopt reads this for the "option" parameter:

The name of the option to set. Can be any of: SUBSCRIBE, UNSUBSCRIBE, IDENTITY, HWM, SWAP, AFFINITY, RATE, RECOVERY_IVL, MCAST_LOOP, SNDBUF, RCVBUF.

[minrk]

Good point, that docstring hasn't been updated in quite some time.

[minrk]

I don't plan to document general zeromq usage inside pyzmq, because I just don't have time, and pyzmq can behave differently based on what version of zeromq you link against. If you want to know about how to use zeromq itself (in any language, Python included), the zeromq docs are the place to go. The pyzmq documentation only covers extensions to the basic zeromq behavior.