Merge branch 'master' into 0.9.0

CHANGELOG.md

@@ -32,6 +32,7 @@
  * Execute latency now tracked and shown in Storm UI
  * Adding testTuple methods for easily creating Tuple instances to Testing API (thanks xumingming)
  * Trident now throws an error during construction of a topology when try to select fields that don't exist in a stream (thanks xumingming)
+ * Bug fix: Fix deadlock bug due to variant of dining philosophers problem. Spouts now use an overflow buffer to prevent blocking and guarantee that it can consume the incoming queue of acks/fails.
  * Bug fix: Fix race condition in supervisor that would lead to supervisor continuously crashing due to not finding "stormconf.ser" file for an already killed topology
  * Bug fix: bin/storm script now displays a helpful error message when an invalid command is specified
 

src/clj/backtype/storm/bootstrap.clj

@@ -30,6 +30,7 @@
      (require (quote [backtype.storm [stats :as stats] [disruptor :as disruptor]]))
      (import (quote [org.slf4j Logger]))
 
+     (import (quote [com.lmax.disruptor InsufficientCapacityException]))
      (import (quote [backtype.storm.generated Nimbus Nimbus$Processor
                      Nimbus$Iface StormTopology ShellComponent
                      NotAliveException AlreadyAliveException GlobalStreamId
@@ -42,6 +43,6 @@
                      SupervisorInfo WorkerHeartbeat]))
      (import (quote [backtype.storm.grouping CustomStreamGrouping]))
      (import (quote [java.io File FileOutputStream FileInputStream]))
-     (import (quote [java.util Collection List Random Map HashMap Collections ArrayList]))
+     (import (quote [java.util Collection List Random Map HashMap Collections ArrayList LinkedList]))
      (import (quote [org.apache.commons.io FileUtils]))
      ))

src/clj/backtype/storm/daemon/executor.clj

@@ -164,8 +164,22 @@
 
 ;; in its own function so that it can be mocked out by tracked topologies
 (defn mk-executor-transfer-fn [batch-transfer->worker]
-  (fn [task tuple]
-    (disruptor/publish batch-transfer->worker [task tuple])))
+  (fn this
+    ([task tuple block? ^List overflow-buffer]
+      (if (and overflow-buffer (not (.isEmpty overflow-buffer)))
+        (.add overflow-buffer [task tuple])
+        (try-cause
+          (disruptor/publish batch-transfer->worker [task tuple] block?)
+        (catch InsufficientCapacityException e
+          (if overflow-buffer
+            (.add overflow-buffer [task tuple])
+            (throw e))
+          ))))
+    ([task tuple overflow-buffer]
+      (this task tuple (nil? overflow-buffer) overflow-buffer))
+    ([task tuple]
+      (this task tuple nil)
+      )))
 
 (defn executor-data [worker executor-id]
   (let [worker-context (worker-context worker)
@@ -383,7 +397,16 @@
         event-handler (mk-task-receiver executor-data tuple-action-fn)
         has-ackers? (has-ackers? storm-conf)
         emitted-count (MutableLong. 0)
-        empty-emit-streak (MutableLong. 0)]
+        empty-emit-streak (MutableLong. 0)
+
+        ;; the overflow buffer is used to ensure that spouts never block when emitting
+        ;; this ensures that the spout can always clear the incoming buffer (acks and fails), which
+        ;; prevents deadlock from occuring across the topology (e.g. Spout -> Bolt -> Acker -> Spout, and all
+        ;; buffers filled up)
+        ;; when the overflow buffer is full, spouts stop calling nextTuple until it's able to clear the overflow buffer
+        ;; this limits the size of the overflow buffer to however many tuples a spout emits in one call of nextTuple, 
+        ;; preventing memory issues
+        overflow-buffer (LinkedList.)]
 
     [(async-loop
       (fn []
@@ -406,13 +429,16 @@
                                          (fast-list-iter [out-task out-tasks id out-ids]
                                                          (let [tuple-id (if rooted?
                                                                           (MessageId/makeRootId root-id id)
-                                                                          (MessageId/makeUnanchored))]                                                        
+                                                                          (MessageId/makeUnanchored))
+                                                               out-tuple (TupleImpl. worker-context
+                                                                                     values
+                                                                                     task-id
+                                                                                     out-stream-id
+                                                                                     tuple-id)]
                                                            (transfer-fn out-task
-                                                                        (TupleImpl. worker-context
-                                                                                    values
-                                                                                    task-id
-                                                                                    out-stream-id
-                                                                                    tuple-id))))
+                                                                        out-tuple
+                                                                        overflow-buffer)
+                                                           ))
                                          (if rooted?
                                            (do
                                              (.put pending root-id [task-id
@@ -421,7 +447,8 @@
                                                                     (if (sampler) (System/currentTimeMillis))])
                                              (task/send-unanchored task-data
                                                                    ACKER-INIT-STREAM-ID
-                                                                   [root-id (bit-xor-vals out-ids) task-id]))
+                                                                   [root-id (bit-xor-vals out-ids) task-id]
+                                                                   overflow-buffer))
                                            (when message-id
                                              (ack-spout-msg executor-data task-data message-id
                                                             {:stream out-stream-id :values values}
@@ -450,10 +477,21 @@
         (fn []
           ;; This design requires that spouts be non-blocking
           (disruptor/consume-batch receive-queue event-handler)
+
+          ;; try to clear the overflow-buffer
+          (try-cause
+            (while (not (.isEmpty overflow-buffer))
+              (let [[out-task out-tuple] (.peek overflow-buffer)]
+                (transfer-fn out-task out-tuple false nil)
+                (.removeFirst overflow-buffer)))
+          (catch InsufficientCapacityException e
+            ))
+
           (let [active? @(:storm-active-atom executor-data)
                 curr-count (.get emitted-count)]
-            (if (or (not max-spout-pending)
-                    (< (.size pending) max-spout-pending))
+            (if (and (.isEmpty overflow-buffer)
+                     (or (not max-spout-pending)
+                         (< (.size pending) max-spout-pending)))
               (if active?
                 (do
                   (when-not @last-active

src/clj/backtype/storm/daemon/task.clj

@@ -82,16 +82,23 @@
 
 
 ;; TODO: this is all expensive... should be precomputed
-(defn send-unanchored [task-data stream values]
-  (let [^TopologyContext topology-context (:system-context task-data)
-        tasks-fn (:tasks-fn task-data)
-        transfer-fn (-> task-data :executor-data :transfer-fn)]
-    (fast-list-iter [t (tasks-fn stream values)]
-      (transfer-fn t
-                   (TupleImpl. topology-context
-                               values
-                               (.getThisTaskId topology-context)
-                               stream)))))
+(defn send-unanchored
+  ([task-data stream values overflow-buffer]
+    (let [^TopologyContext topology-context (:system-context task-data)
+          tasks-fn (:tasks-fn task-data)
+          transfer-fn (-> task-data :executor-data :transfer-fn)
+          out-tuple (TupleImpl. topology-context
+                                 values
+                                 (.getThisTaskId topology-context)
+                                 stream)]
+      (fast-list-iter [t (tasks-fn stream values)]
+        (transfer-fn t
+                     out-tuple
+                     overflow-buffer)
+        )))
+    ([task-data stream values]
+      (send-unanchored task-data stream values nil)
+      ))
 
 (defn mk-tasks-fn [task-data]
   (let [task-id (:task-id task-data)

src/clj/backtype/storm/disruptor.clj

@@ -46,8 +46,14 @@
 (defmacro handler [& args]
   `(clojure-handler (fn ~@args)))
 
-(defn publish [^DisruptorQueue q o]
-  (.publish q o))
+(defn publish
+  ([^DisruptorQueue q o block?]
+    (.publish q o block?))
+  ([q o]
+    (publish q o true)))
+
+(defn try-publish [^DisruptorQueue q o]
+  (.tryPublish q o))
 
 (defn consume-batch [^DisruptorQueue queue handler]
   (.consumeBatch queue handler))

src/jvm/backtype/storm/utils/DisruptorQueue.java

@@ -4,13 +4,16 @@
 import com.lmax.disruptor.ClaimStrategy;
 import com.lmax.disruptor.EventFactory;
 import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.InsufficientCapacityException;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.Sequence;
 import com.lmax.disruptor.SequenceBarrier;
 import com.lmax.disruptor.SingleThreadedClaimStrategy;
 import com.lmax.disruptor.WaitStrategy;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  *
@@ -92,8 +95,25 @@ private void consumeBatchToCursor(long cursor, EventHandler<Object> handler) {
      * Caches until consumerStarted is called, upon which the cache is flushed to the consumer
      */
     public void publish(Object obj) {
+        try {
+            publish(obj, true);
+        } catch (InsufficientCapacityException ex) {
+            throw new RuntimeException("This code should be unreachable!");
+        }
+    }
+
+    public void tryPublish(Object obj) throws InsufficientCapacityException {
+        publish(obj, false);
+    }
+
+    public void publish(Object obj, boolean block) throws InsufficientCapacityException {
         if(consumerStartedFlag) {
-            final long id = _buffer.next();
+            final long id;
+            if(block) {
+                id = _buffer.next();
+            } else {
+                id = _buffer.tryNext(1);
+            }
             final MutableObject m = _buffer.get(id);
             m.setObject(obj);
             _buffer.publish(id);