ZMQ crashes from Out of Memory with PUB/SUB

[jhawk28]

The following code causes ZMQ to run out of memory on Windows 32bit. I suspect it dies on OSX too, but the memory caps are a little higher (I see the virtual space continuing to increase)

I kick off the publisher and then the subscriber. I then wait and the publisher dies with an error for line 108 of yqueue.hpp

It does not die if the publisher uses a bind and the subscriber uses a connect. Another observation is that if I move the HWM to before the connect, it doesn't crash. So it looks like the HWM can be set before and after a BIND, but only before a CONNECT.

import org.zeromq.ZMQ;
class publisherApp
{
  public static void main(String[] args)
  {
    ZMQ.Context ctx = ZMQ.context(1);
    ZMQ.Socket s = ctx.socket(ZMQ.PUB);
    s.connect("tcp://127.0.0.1:5555");

    s.setHWM(1); // High water mark so only one message is left in queue (expect no memory increase
    s.setSwap(25000000); // Swap space on disk in bytes

    for ( ;; )
    {
      s.send("21600".getBytes(), 0);
    }
  }
}

import org.zeromq.ZMQ;
public class subscriberApp
{
  public static void main(String[] args)
  {
    // Initialize 0MQ with a single I/O thread,
    ZMQ.Context ctx = ZMQ.context(1);

    // Create a PUB socket for port 5555 on the loopback interface
    ZMQ.Socket s = ctx.socket(ZMQ.SUB);
    s.subscribe("21600".getBytes());
    s.bind("tcp://127.0.0.1:5555");

    for (int i = 0; i < 15; i++)
    {
      System.out.println(new String(s.recv(0)));
    }
  }
}

[sustrik]

You have to use ZMQ_HWM socket option to limit number of messages held in the memory.

[jhawk28]

See line 14 of publisherApp, it sets the HWM to 1.

[sustrik]

Ok. Can you provide a minimal program to reproduce the behaviour?

[jhawk28]

Will http://gist.github.com/572604 do?

[sustrik]

I mean, try removing as many parts of the program as possible while still keeping the problem around. If you post a large batch of code nobody would have enough time to check it.

[jhawk28]

ok, updated the issue.

[sustrik]

Can you check whether following patch helps:

diff --git a/src/pub.cpp b/src/pub.cpp
index 4e73b19..d4a1824 100644
--- a/src/pub.cpp
+++ b/src/pub.cpp
@@ -133,13 +133,25 @@ int zmq::pub_t::xsend (zmq_msg_t *msg_, int flags_)
     }
 
     //  Push the message to all destinations.
+    int passivised = 0;
     for (pipes_t::size_type i = 0; i != active;) {
         if (!write (pipes [i], msg_))
-            content->refcnt.sub (1);
+            passivised++;
         else
             i++;
     }
 
+    //   If some of the pipes we've tried to send the message to have become
+    //   passive we have to subtract corresponding number of references from
+    //   the message. If there are no more references on the message we can
+    //   destroy it.
+    if (passivised) {
+        if (!content->refcnt.sub (passivised)) {
+            msg_->flags &= ~ZMQ_MSG_SHARED;
+            zmq_msg_close (msg_);
+        }
+    }
+
     //  Detach the original message from the data buffer.
     int rc = zmq_msg_init (msg_);
     zmq_assert (rc == 0);

[jhawk28]

No, the patch did not fix the problem.

[jhawk28]

Based on IRC discussions, the software is working as designed. The HWM should be set before the connect is called. If it isn't, it sets it to the default (infinite) and will eventually die of out of memory.