implemented configurable high water mark for zmq

Nathan Marz · Nathan Marz · commit 83e2eecc662e · 2012-01-04T23:33:26.000-08:00
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -54,6 +54,7 @@ task.refresh.poll.secs: 10
 
 zmq.threads: 1
 zmq.linger.millis: 5000
+zmq.high.water.mark: null
 
 ### topology.* configs are for specific executing storms
 topology.debug: false
@@ -68,4 +69,4 @@ topology.max.spout.pending: null
 topology.state.synchronization.timeout.secs: 60
 topology.stats.sample.rate: 0.05
 topology.fall.back.on.java.serialization: true
-topology.worker.childopts: nil
+topology.worker.childopts: null
diff --git a/src/clj/backtype/storm/daemon/worker.clj b/src/clj/backtype/storm/daemon/worker.clj
@@ -112,6 +112,7 @@
                      mq-context
                      (msg-loader/mk-zmq-context (storm-conf ZMQ-THREADS)
                                                 (storm-conf ZMQ-LINGER-MILLIS)
+                                                (storm-conf ZMQ-HIGH-WATER-MARK)
                                                 (= (conf STORM-CLUSTER-MODE) "local")))
         outbound-tasks (worker-outbound-tasks task->component mk-topology-context task-ids)
         endpoint-socket-lock (mk-rw-lock)
diff --git a/src/clj/backtype/storm/messaging/zmq.clj b/src/clj/backtype/storm/messaging/zmq.clj
@@ -17,7 +17,7 @@
     (.close socket)
     ))
 
-(deftype ZMQContext [context linger-ms ipc?]
+(deftype ZMQContext [context linger-ms hwm ipc?]
   Context
   (bind [this virtual-port]
     (-> context
@@ -32,6 +32,7 @@
       (-> context
           (mq/socket mq/push)
           (mq/set-linger linger-ms)
+          (mq/set-hwm hwm)
           (mq/connect url)
           (ZMQConnection.))))
   (send-local-task-empty [this virtual-port]
@@ -45,6 +46,6 @@
     context))
 
 
-(defn mk-zmq-context [num-threads linger local?]
-  (ZMQContext. (mq/context num-threads) linger local?))
+(defn mk-zmq-context [num-threads linger hwm local?]
+  (ZMQContext. (mq/context num-threads) linger hwm local?))
 
diff --git a/src/clj/zilch/mq.clj b/src/clj/zilch/mq.clj
@@ -56,6 +56,14 @@
   (doto socket
     (.setLinger (long linger-ms))))
 
+(defn set-hwm
+  [^ZMQ$Socket socket hwm]
+  (if hwm
+    (doto socket
+      (.setHwm (long hwm)))
+    socket
+    ))
+
 (defn bind
   [^ZMQ$Socket socket url]
   (doto socket
diff --git a/src/jvm/backtype/storm/Config.java b/src/jvm/backtype/storm/Config.java
@@ -342,7 +342,14 @@ public class Config extends HashMap<String, Object> {
      * The number of threads that should be used by the zeromq context in each worker process.
      */
     public static String ZMQ_THREADS = "zmq.threads";
-
+    
+    /**
+     * The high water mark for the underlying push sockets used to send messages. If downstream 
+     * tasks to a producer are overloaded, the producer will block until the downstream tasks
+     * have consumed messages. This configuration can be used to prevent overloading in topologies
+     * that use unreliable spouts and are thus not controllable with TOPOLOGY_MAX_SPOUT_PENDING.
+     */
+    public static String ZMQ_HIGH_WATER_MARK = "zmq.high.water.mark";
 
     /**
      * How long a connection should retry sending messages to a target host when
