Limiting queue length with PyZMQ

Question

I want to limit the amount of memory consumed by my ZeroMQ message queues in a Python application. I know that setting the high-water mark will limit the amount that will be queued on the sender side, but is there a way to control how much will be queued on the receiver side? The Python ZeroMQ binding seems to have it set at unlimited.

My test scenario: I have two python terminals that I am using for testing. One is the receiver:

Python 2.5.1 (r251:54863, Aug 25 2008, 20:50:04) 
[GCC 4.1.2 20071124 (Red Hat 4.1.2-42)] on linux2
Type "help", "copyright", "credits" or "license" for more information.

>>> import zmq
>>> context = zmq.Context()
>>> socket = context.socket(zmq.PULL)
>>> socket.setsockopt(zmq.RCVBUF, 256)
>>> socket.bind("tcp://127.0.0.1:12345")

The other is the sender:

Python 2.5.1 (r251:54863, Aug 25 2008, 20:50:04) 
[GCC 4.1.2 20071124 (Red Hat 4.1.2-42)] on linux2
Type "help", "copyright", "credits" or "license" for more information.

>>> import zmq
>>> context=zmq.Context()
>>> socket = context.socket(zmq.PUSH)
>>> socket.setsockopt(zmq.SNDBUF, 2048)
>>> socket.setsockopt(zmq.HWM, 1)
>>> socket.connect("tcp://127.0.0.1:12345")
>>> num = 0
>>> while True:
...  print num
...  socket.send(str(num))
...  num = num + 1
... 

I run socket.recv() on the receiver side a couple times to make sure that the queue works, but other than that, let the two terminals just sit there. The send loop seems to never block and the receive prompt seems to have a growing memory footprint.

Answer 1

In contradiction to the documentation of ZeroMQ, the high water mark needs to be set on both the PUSH side and the PULL side. Once I changed the PULL, it worked better. The new PULL code is:

Python 2.5.1 (r251:54863, Aug 25 2008, 20:50:04) 
[GCC 4.1.2 20071124 (Red Hat 4.1.2-42)] on linux2
Type "help", "copyright", "credits" or "license" for more information.

>>> import zmq
>>> context=zmq.Context()
>>> socket = context.socket(zmq.PULL)
>>> socket.setsockopt(zmq.RCVBUF, 256)
>>> socket.setsockopt(zmq.HWM, 1)
>>> socket.bind("tcp://127.0.0.1:12345")