Problem in High Water Mark option in ZMQ communication

[agn-7]

I have a problem in High Water Mark option in ZMQ communication:

According to the ZeroMQ documentation, a pub socket is supposed to drop messages once the number of queued messages reaches the high-water mark.

This doesn't seem to work in the following example (and yes I do set the hwm before bind/connect):

import time
import pickle
from threading import Thread
import zmq

ctx = zmq.Context()

def pub_thread():
    pub = ctx.socket(zmq.PUB)
    pub.set_hwm(2)
    pub.bind('tcp://*:5555')

    i = 0
    while True:
        # Send message every 100ms
        time.sleep(0.1)
        pub.send_string("test", zmq.SNDMORE)
        pub.send_pyobj(i)
        i += 1

def sub_thread():
    sub = ctx.socket(zmq.SUB)
    sub.subscribe("test")
    sub.connect('tcp://localhost:5555')
    while True:
        # Receive messages only every second
        time.sleep(1)
        msg = sub.recv_multipart()
        print("Sub: %d" % pickle.loads(msg[1]))

t_pub = Thread(target=pub_thread)
t_sub = Thread(target=sub_thread)
t_pub.start()
t_sub.start()

while True:
    pass

I'm sending messages on pub 10 times faster than reading them on the sub socket, hwm is set to 2. I would expect to only receive about every 10th message. Instead, I see the following output:

Sub: 0
Sub: 1
Sub: 2
Sub: 3
Sub: 4
Sub: 5
Sub: 6
Sub: 7
Sub: 8
Sub: 9
Sub: 10
Sub: 11
Sub: 12
Sub: 13
Sub: 14
...

So I see all messages arriving, thus they are held in some queue until I read them. Same holds true when adding a hwm=2 on the sub socket as well before connect.

What am I doing wrong or am I misunderstanding hwm?

I use pyzmq version 17.1.2

---

Also I tried with the following code, but result was the same:

import time
import pickle
import zmq
from threading import Thread

ctx = zmq.Context()

def pub_thread():
    pub = ctx.socket(zmq.PUB)
    pub.setsockopt(zmq.SNDHWM, 2)  # This line was changed.
    pub.bind('tcp://*:5555')

    i = 0
    while True:
        time.sleep(0.1)
        pub.send_string("test", zmq.SNDMORE)
        pub.send_pyobj(i)
        i += 1

def sub_thread():
    sub = ctx.socket(zmq.SUB)
    sub.setsockopt(zmq.SUBSCRIBER, b'')
    sub.setsockopt(zmq.RCVHWM, 2)  # This line added.
    # sub.setsockopt(zmq.CONFLATE, 1)  # Last msg only (another option).
    sub.connect('tcp://localhost:5555')
    while True:
        time.sleep(1)
        msg = sub.recv_multipart()
        print("Sub: %d" % pickle.loads(msg[1]))

t_pub = Thread(target=pub_thread)
t_pub.start()
sub_thread()  # Start with main thread.

---

These posts (POST1, POST2, POST3) are relevant to this issue in the Stack Overflow.

---

Thanks in advance,

[haverins]

Messages are held in operating system's network buffers. I have found HWMs to be not that useful because of that. Here is modified code where subscriber misses messages:

import time
import pickle
import zmq
from threading import Thread
import os

ctx = zmq.Context()

def pub_thread():
    pub = ctx.socket(zmq.PUB)
    pub.setsockopt(zmq.SNDHWM, 2)
    pub.setsockopt(zmq.SNDBUF, 2*1024) # see: http://api.zeromq.org/4-2:zmq-setsockopt
    pub.bind('tcp://*:5555')
    i = 0
    while True:
        time.sleep(0.001)
        pub.send_string(str(i), zmq.SNDMORE)
        pub.send(os.urandom(1024))
        i += 1

def sub_thread():
    sub = ctx.socket(zmq.SUB)
    sub.setsockopt(zmq.SUBSCRIBE, b'')
    sub.setsockopt(zmq.RCVHWM, 2)
    sub.setsockopt(zmq.RCVBUF, 2*1024)
    sub.connect('tcp://localhost:5555')
    while True:
        time.sleep(0.1)
        msg, _ = sub.recv_multipart()
        print("Received:", msg.decode())

t_pub = Thread(target=pub_thread)
t_pub.start()
sub_thread()

Output looks something like this:

Received: 0
Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Received: 6
Received: 47
Received: 48
Received: 64
Received: 65
Received: 84
Received: 85
Received: 159
Received: 160
Received: 270

Messages are missed because all queues/buffers are full and publisher starts to drop messages (see documentation for ZMQ_PUB: http://api.zeromq.org/4-2:zmq-socket).

[agn-7]

@haverins Thanks it works.

Is there any other way to set a limit on the number of buffer instead of buffer size?

[ilejn]

At first glance set_hwm method is broken for modern 0mq.
It is effectively

        if major >= 3:
            raised = None
            try:
                self.sndhwm = value
            except Exception as e:
                raised = e
            try:
                self.rcvhwm = value
            except Exception as e:
                raised = e
            
            if raised:
                raise raised

sndhwm is not used in any other way.
Am I missing something?

Update: yes, I am missing AttributeSetter, sorry.

[minrk]

Closing some old issues, especially those about libzmq socket behavior not specific to pyzmq. If you still have issues or questions, always feel free to open against libzmq, or other pyzmq-specific questions here.