     (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]


(defn default-collect
  [vertex]
  (swap! (:state vertex) #(update % :val into (:uncollected %))))

(defn signal-forward [state] (:val state))

(defn default-score-signal
  [{:keys [state last-sig-state] :as vertex}]
  (if (= (:val @state) @last-sig-state) 0 1))

(defn default-score-collect
  [vertex] (-> vertex :state deref :uncollected count))

(defn should-signal?
  [vertex thresh]
  (> ((-> vertex :state deref :score-signal) vertex) thresh))

(defn should-collect?
  [vertex thresh]
  (> ((-> vertex :state deref :score-collect) vertex) thresh))

(defrecord Vertex [id state last-sig-state in outs]
  IVertex
  (connect-to!
    [_ v sig-fn]
    (swap! outs assoc v sig-fn) _)
  (disconnect-from!
    [_ v] (swap! outs dissoc v) _)
  (disconnect! [_]
    (close! in) _)
  (input-channel
    [_] in)
  (signal!
    [_]
    (let [state @state]
      (reset! last-sig-state (:val state))
      (go-loop [outs @outs]
        (let [[v f] (first outs)]
          (when v
            (>! (input-channel v) [id (f state)])
            (recur (next outs)))))
      _))
  (receive-signal

   [taoensso.timbre :refer [info warn error]]))

(defprotocol IVertex
  (connect-to! [_ v sig-fn edge-opts])
  (disconnect-from! [_ v])
  (disconnect! [_])
  (input-channel [_])
  (collect! [_])
  (score-collect [_])
  (signal! [_])
  (score-signal [_])
  (receive-signal [_ src sig]))

(defprotocol IComputeGraph
  (add-vertex! [_ vstate])
  (vertex-for-id [_ id])
  (vertices [_]))

(defprotocol IGraphExecutor
