cleanup code

Nathan Marz · Nathan Marz · commit a52724062c5f · 2012-04-28T17:47:17.000-07:00
diff --git a/src/clj/backtype/storm/daemon/task.clj b/src/clj/backtype/storm/daemon/task.clj
@@ -294,13 +294,10 @@
 (defn mk-task-receiver [^LinkedBlockingQueue receive-queue ^KryoTupleDeserializer deserializer tuple-action-fn]
   (fn []
     (let [msg (.take receive-queue)
-          is-ser-msg? (not (instance? Tuple msg))
-          is-empty-msg? (or (nil? msg) (and is-ser-msg? (empty? msg)))]
-      (when-not is-empty-msg? ; skip empty messages (used during shutdown)
-        (log-debug "Processing message")
-        (let [^Tuple tuple (if is-ser-msg?
-                        (.deserialize deserializer msg)
-                        msg)]
+          is-tuple? (instance? Tuple msg)]
+      (when (or is-tuple? (not (empty? msg))) ; skip empty messages (used during shutdown)
+        (log-debug "Processing message " msg)
+        (let [^Tuple tuple (if is-tuple? msg (.deserialize deserializer msg))]
           (tuple-action-fn tuple)
           ))
       )))
@@ -381,6 +378,7 @@
     (log-message "Opening spout " component-id ":" task-id)
     (.open spout storm-conf user-context (SpoutOutputCollector. output-collector))
     (log-message "Opened spout " component-id ":" task-id)
+    ;; TODO: should redesign this to only use one thread
     [(fn []
        ;; This design requires that spouts be non-blocking
        (loop []
diff --git a/src/clj/backtype/storm/daemon/worker.clj b/src/clj/backtype/storm/daemon/worker.clj
@@ -70,8 +70,7 @@
         storm-active-atom (atom false)
         cluster-state (cluster/mk-distributed-cluster-state conf)
         storm-cluster-state (cluster/mk-storm-cluster-state cluster-state)
-        task-ids (read-worker-task-ids storm-cluster-state storm-id supervisor-id port)
-        task-ids-set (set task-ids)
+        task-ids (set (read-worker-task-ids storm-cluster-state storm-id supervisor-id port))
         ;; because in local mode, its not a separate
         ;; process. supervisor will register it in this case
         _ (when (= :distributed (cluster-mode conf))
@@ -94,7 +93,7 @@
                                                (worker-pids-root conf worker-id)
                                                %
                                                port
-                                               task-ids)
+                                               (vec task-ids))
         mk-user-context #(TopologyContext. topology
                                            storm-conf
                                            task->component
@@ -104,7 +103,7 @@
                                            (worker-pids-root conf worker-id)
                                            %
                                            port
-                                           task-ids)
+                                           (vec task-ids))
         mq-context (if mq-context
                      mq-context
                      (msg-loader/mk-zmq-context (storm-conf ZMQ-THREADS)
@@ -129,7 +128,7 @@
                                       my-assignment (select-keys (:task->node+port assignment) outbound-tasks)
                                       ;; we dont need a connection for the local tasks anymore
                                       needed-connections (->> my-assignment
-                                                              (filter #(->> % key (contains? task-ids-set) not))
+                                                              (filter-key (complement task-ids))
                                                               vals
                                                               set)
                                       current-connections (set (keys @node+port->socket))
@@ -195,9 +194,9 @@
                     (read-locked endpoint-socket-lock
                       (let [node+port->socket @node+port->socket
                             task->node+port @task->node+port]
-                        (doseq [[task tuple] drainer]
+                        (doseq [[task ser-tuple] drainer]
                           (let [socket (node+port->socket (task->node+port task))]
-                            (msg/send socket task tuple)
+                            (msg/send socket task ser-tuple)
                             )
                         )))
                     (.clear drainer)
diff --git a/src/clj/backtype/storm/messaging/loader.clj b/src/clj/backtype/storm/messaging/loader.clj
@@ -1,9 +1,7 @@
 (ns backtype.storm.messaging.loader
-  (:use [backtype.storm.bootstrap])
+  (:use [backtype.storm util log])
   (:import [java.util.concurrent LinkedBlockingQueue])
-  (:require [backtype.storm.messaging.local :as local]))
-
-(bootstrap)
+  (:require [backtype.storm.messaging [local :as local] [protocol :as msg]]))
 
 (defn mk-local-context []
   (local/mk-local-context))
diff --git a/src/clj/backtype/storm/messaging/local.clj b/src/clj/backtype/storm/messaging/local.clj
@@ -16,8 +16,7 @@
   (recv [this]
     (when-not queue
       (throw (IllegalArgumentException. "Cannot receive on this socket")))
-    (let [[port tuple] (.take queue)]
-      [(short port) tuple]))
+    (.take queue))
   (send [this task message]
     (let [send-queue (add-queue! queues-map lock storm-id port)]
       (.put send-queue [task message])
