         [_]
         (stop-execution bus (vertices (:graph ctx)))
         result)
-      (notify [_ evt]
+      (notify! [_ evt]
         (go (>! (rand-nth bus) evt)))
+      (include-vertex!
+        [_ v]
+        (processor v _)
+        (when (should-collect? v c-thresh)
+          (collect! v))
+        (when (should-signal? v s-thresh)
+          (signal! v))
+        _)
       (execute! [_]
         (if-not (realized? result)
           (let [t0 (System/nanoTime)]
             (go
-              (loop [verts (vertices (:graph ctx))]
-                (when-let [v (first verts)]
-                  (processor v _)
-                  (when (should-collect? v c-thresh)
-                    (collect! v))
-                  (when (should-signal? v s-thresh)
-                    (signal! v))
-                  (recur (next verts))))
+              (run! #(include-vertex! _ %) (vertices (:graph ctx)))
               (loop [colls 0 sigs 0]
                 (if (<= (+ colls sigs) max-ops)
                   (let [t (if max-t (timeout max-t))

      _))
  (receive-signal
    [_ src sig]
    (swap! state
           #(-> %
                (update ::uncollected conj sig)
                (assoc-in [::signal-map src] sig)))
    _))

#?(:clj

%s}")
       (spit path)))

#_(def g (compute-graph))
#_(def ctx
  (async-context
   {:graph     g
    :bus-size  64
