     (go-loop []
       (let [[src-id sig] (<! in)]
         (if sig
-          (do (info id "receive from " src-id sig)
+          (do (info id "receive from" src-id sig)
               (receive-signal vertex src-id sig)
               (when (should-collect? vertex collect-thresh)
                 (when (>! ctrl [:collect id])
-                  ((:collect-fn @state) vertex)
-                  (swap! state assoc :uncollected [])
+                  (collect! vertex)
+                  (swap! state assoc ::uncollected [])
                   (when (should-signal? vertex signal-thresh)
                     (>! ctrl [:signal id])
                     (signal! vertex))))
               (recur))
-          (info id "disconnected"))))
+          (info id "stopped"))))
     vertex))
+
+(defn execution-result
+  [p t0 colls sigs]
+  (deliver
+   p {:collections colls
+      :signals     sigs
+      :time        (* (- (System/nanoTime) t0) 1e-6)}))
 
 (defn async-context
   [opts]

        ctx      (merge default-context-opts {:ctrl ctrl} opts)
        c-thresh (:collect-thresh ctx)
        s-thresh (:signal-thresh ctx)
        ctx      (atom ctx)]
    (reify
      clojure.lang.IDeref
      (deref [_] @ctx)
      IGraphExecutor
      (execute! [_ vertices]
        (go
          (loop [verts vertices]
            (when-let [v (first verts)]
              (when (should-signal? v s-thresh)
                (signal! v))
              (recur (next verts))))
          (loop []
            (if (<= (+ (:collections @ctx) (:signals @ctx)) (:max-ops @ctx))
              (let [t (timeout (:timeout @ctx))
                    [[evt v ex] port] (alts! [ctrl t])]
                (if (= port t)
                  (info "graph converged, done...")
                  (case evt
                    :collect (do (swap! ctx update :collections inc) (recur))
                    :signal  (do (swap! ctx update :signals inc) (recur))
                    :error   (warn ex "@ vertex" v)
                    (info "done" (:collections @ctx) (:signals @ctx)))))
              (info "reached max ops: " (:max-ops @ctx)))))))))

(def g (compute-graph))
(def ctx (async-context {}))

(def a (eager-async-vertex (add-vertex! g {:val 'a :collect-fn default-collect}) ctx))
(def b (eager-async-vertex (add-vertex! g {:val '[b] :collect-fn default-collect}) ctx))
(def c (eager-async-vertex (add-vertex! g {:val '[c] :collect-fn default-collect}) ctx))

(connect-to! a b signal-forward)
(connect-to! a c signal-forward)
(connect-to! b c signal-forward)

(execute! ctx (vertices g))

(comment
  (disconnect! a)
  (disconnect! b)
  (disconnect! c)
  )

                 (update :vertices assoc id v)
                 (assoc  :curr-vertex v)))))
        :curr-vertex))
  (vertex-for-id
    [_ id] (-> @state :vertices id))
  (vertices
    [_] (-> @state :vertices vals)))

