        (deliver p)))
 
 (defn stop-execution
-  [ctrl vertices]
-  (close! ctrl)
+  [bus vertices]
+  (run! close! bus)
   (run! close-input! vertices))
 
 (defn async-context
   [opts]
   (let [ctx       (merge default-context-opts opts)
-        ctx       (update ctx :channel #(or % (chan)))
-        ctrl      (:channel ctx)
+        bus       (vec (repeatedly (:bus-size ctx) chan))
+        ctx       (assoc ctx :bus bus)
         processor (:processor ctx)
         c-thresh  (:collect-thresh ctx)
         s-thresh  (:signal-thresh ctx)

(defn eager-vertex-processor
  [{:keys [id state] :as vertex} ctx]
  (let [{:keys [collect-thresh signal-thresh]} @ctx
        ctrl (control-channel ctx)
        in   (input-channel vertex)]
    (go-loop []
      (let [[src-id sig] (<! in)]

  (simple-collect
   (fn [val uncollected]
     (if val (reduce min val uncollected) (reduce min uncollected)))))

(defn make-strand
  [verts e-length]
