Agfdhyk

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

With borrowing an answer to the issue which I opened in Github, I've updated my answer as follows:

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).

[NOTE]:

• You should use the high-water mark option in listener/subscriber and advertiser/publisher.


• These posts are also relevant (Post1 - Post2)


• 
  sock.setsockopt(zmq.CONFLATE, 1) is another option to get the last message only which defined in subscriber side.