           ^ISupervisor isupervisor (:isupervisor supervisor)
           ^LocalState local-state (:local-state supervisor)
           sync-callback (fn [& ignored] (.add event-manager this))
-          storm-code-map (read-storm-code-locations storm-cluster-state sync-callback)
+          assignments-snapshot (assignments-snapshot storm-cluster-state sync-callback)
+          storm-code-map (read-storm-code-locations assignments-snapshot)
           downloaded-storm-ids (set (read-downloaded-storm-ids conf))
           all-assignment (read-assignments
-                           storm-cluster-state
-                           (:assignment-id supervisor)
-                           sync-callback)
+                           assignments-snapshot
+                           (:assignment-id supervisor))
           new-assignment (->> all-assignment
                               (filter-key #(.confirmAssigned isupervisor %)))
           assigned-storm-ids (assigned-storm-ids-from-port-assignments new-assignment)

               ;; need to cast to int b/c it might be a long (due to how yaml parses things)
               ;; doall is to avoid serialization/deserialization problems with lazy seqs
               [(Integer. port) (LocalAssignment. storm-id (doall executors))]
               ))
    ))

(defn- read-assignments
  "Returns map from port to struct containing :storm-id and :executors"
  [storm-cluster-state assignment-id callback]
  (let [storm-ids (.assignments storm-cluster-state callback)]
    (apply merge-with
           (fn [& ignored]
             (throw (RuntimeException.
                     "Should not have multiple topologies assigned to one port")))
           (dofor [sid storm-ids] (read-my-executors storm-cluster-state sid assignment-id callback))
           )))

(defn- read-storm-code-locations
  [storm-cluster-state callback]
  (let [storm-ids (.assignments storm-cluster-state callback)]
    (into {}
      (dofor [sid storm-ids]
        [sid (:master-code-dir (.assignment-info storm-cluster-state sid callback))]
        ))
    ))


(defn- read-downloaded-storm-ids [conf]
  (map #(java.net.URLDecoder/decode %) (read-dir-contents (supervisor-stormdist-root conf)))

  (shutdown-all-workers [this])
  )

(defn- assignments-snapshot [storm-cluster-state callback]
  (let [storm-ids (.assignments storm-cluster-state callback)]
     (->> (dofor [sid storm-ids] {sid (.assignment-info storm-cluster-state sid callback)})
          (apply merge)
          (filter-val not-nil?)
          )))

(defn- read-my-executors [assignments-snapshot storm-id assignment-id]
  (let [assignment (get assignments-snapshot storm-id)
        my-executors (filter (fn [[_ [node _]]] (= node assignment-id))
                           (:executor->node+port assignment))
        port-executors (apply merge-with
                          concat
                          (for [[executor [_ port]] my-executors]
