ZMQ_HWM split into ZMQ_SNDHWM and ZMQ_RCVHWM

These new options allow to control the maximum size of the
inbound and outbound message pipe separately.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

4 files changed, 44 insertions, 18 deletions

diff --git a/src/options.cpp b/src/options.cpp
index 39f8984..556ffd8 100644
--- a/src/options.cpp
+++ b/src/options.cpp
@@ -26,7 +26,8 @@
 #include "err.hpp"
 
 zmq::options_t::options_t () :
-    hwm (0),
+    sndhwm (0),
+    rcvhwm (0),
     affinity (0),
     rate (100),
     recovery_ivl (10000),
@@ -49,12 +50,20 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
 {
     switch (option_) {
 
-    case ZMQ_HWM:
+    case ZMQ_SNDHWM:
         if (optvallen_ != sizeof (int) || *((int*) optval_) < 0) {
             errno = EINVAL;
             return -1;
         }
-        hwm = *((int*) optval_);
+        sndhwm = *((int*) optval_);
+        return 0;
+
+    case ZMQ_RCVHWM:
+        if (optvallen_ != sizeof (int) || *((int*) optval_) < 0) {
+            errno = EINVAL;
+            return -1;
+        }
+        rcvhwm = *((int*) optval_);
         return 0;
 
     case ZMQ_AFFINITY:
@@ -168,13 +177,22 @@ int zmq::options_t::getsockopt (int option_, void *optval_, size_t *optvallen_)
 {
     switch (option_) {
 
-    case ZMQ_HWM:
+    case ZMQ_SNDHWM:
         if (*optvallen_ < sizeof (int)) {
             errno = EINVAL;
             return -1;
         }
-        *((int*) optval_) = hwm;
-        *optvallen_ = sizeof (uint64_t);
+        *((int*) optval_) = sndhwm;
+        *optvallen_ = sizeof (int);
+        return 0;
+
+    case ZMQ_RCVHWM:
+        if (*optvallen_ < sizeof (int)) {
+            errno = EINVAL;
+            return -1;
+        }
+        *((int*) optval_) = rcvhwm;
+        *optvallen_ = sizeof (int);
         return 0;
 
     case ZMQ_AFFINITY:
diff --git a/src/options.hpp b/src/options.hpp
index 971e643..9ba06e3 100644
--- a/src/options.hpp
+++ b/src/options.hpp
@@ -35,8 +35,9 @@ namespace zmq
         int setsockopt (int option_, const void *optval_, size_t optvallen_);
         int getsockopt (int option_, void *optval_, size_t *optvallen_);
 
-        //  High-water mark for messages in pipe.
-        int hwm;
+        //  High-water marks for message pipes.
+        int sndhwm;
+        int rcvhwm;
 
         uint64_t affinity;
         blob_t identity;
diff --git a/src/session.cpp b/src/session.cpp
index 176f0ef..5f970cc 100644
--- a/src/session.cpp
+++ b/src/session.cpp
@@ -247,11 +247,13 @@ void zmq::session_t::process_attach (i_engine *engine_,
 
         //  Create the pipes, as required.
         if (options.requires_in) {
-            create_pipe (socket, this, options.hwm, &socket_reader, &out_pipe);
+            create_pipe (socket, this, options.rcvhwm, &socket_reader,
+                &out_pipe);
             out_pipe->set_event_sink (this);
         }
         if (options.requires_out) {
-            create_pipe (this, socket, options.hwm, &in_pipe, &socket_writer);
+            create_pipe (this, socket, options.sndhwm, &in_pipe,
+                &socket_writer);
             in_pipe->set_event_sink (this);
         }
 
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 374e342..bf873b6 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -377,20 +377,25 @@ int zmq::socket_base_t::connect (const char *addr_)
 
         // The total HWM for an inproc connection should be the sum of
         // the binder's HWM and the connector's HWM.
-        int  hwm;
-        if (options.hwm == 0 || peer.options.hwm == 0)
-            hwm = 0;
+        int  sndhwm;
+        int  rcvhwm;
+        if (options.sndhwm == 0 || peer.options.rcvhwm == 0)
+            sndhwm = 0;
         else
-            hwm = options.hwm + peer.options.hwm;
+            sndhwm = options.sndhwm + peer.options.rcvhwm;
+        if (options.rcvhwm == 0 || peer.options.sndhwm == 0)
+            rcvhwm = 0;
+        else
+            rcvhwm = options.rcvhwm + peer.options.sndhwm;
 
         //  Create inbound pipe, if required.
         if (options.requires_in)
-            create_pipe (this, peer.socket, hwm, &inpipe_reader,
+            create_pipe (this, peer.socket, rcvhwm, &inpipe_reader,
                 &inpipe_writer);
 
         //  Create outbound pipe, if required.
         if (options.requires_out)
-            create_pipe (peer.socket, this, hwm, &outpipe_reader,
+            create_pipe (peer.socket, this, sndhwm, &outpipe_reader,
                 &outpipe_writer);
 
         //  Attach the pipes to this socket object.
@@ -429,12 +434,12 @@ int zmq::socket_base_t::connect (const char *addr_)
 
         //  Create inbound pipe, if required.
         if (options.requires_in)
-            create_pipe (this, session, options.hwm,
+            create_pipe (this, session, options.rcvhwm,
                 &inpipe_reader, &inpipe_writer);
 
         //  Create outbound pipe, if required.
         if (options.requires_out)
-            create_pipe (session, this, options.hwm,
+            create_pipe (session, this, options.sndhwm,
                 &outpipe_reader, &outpipe_writer);
 
         //  Attach the pipes to the socket object.