         ctx      (merge default-context-opts {:ctrl ctrl} opts)
         c-thresh (:collect-thresh ctx)
         s-thresh (:signal-thresh ctx)
-        ctx      (atom ctx)]
+        max-ops  (:max-ops ctx)
+        max-t    (:timeout ctx)]
     (reify
       clojure.lang.IDeref
-      (deref [_] @ctx)
+      (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)))))))))
+      (execute! [_ g]
+        (let [t0  (System/nanoTime)
+              res (promise)]
+          (go
+            (loop [verts (vertices g)]
+              (when-let [v (first verts)]
+                (when (should-collect? v c-thresh)
+                  ((-> v :state deref ::collect-fn) vertex))
+                (when (should-signal? v s-thresh)
+                  (signal! v))
+                (recur (next verts))))
+            (loop [colls 0 sigs 0]
+              (if (<= (+ colls sigs) max-ops)
+                (let [t (timeout max-t)
+                      [[evt v ex] port] (alts! [ctrl t])]
+                  (if (= port t)
+                    (do (info "graph converged, done...")
+                        (execution-result res t0 colls sigs))
+                    (case evt
+                      :collect (recur (inc colls) sigs)
+                      :signal  (recur colls (inc sigs))
+                      :error   (warn ex "@ vertex" v)
+                      (do (info "wrong event" evt v)))))
+                (do (info "reached max ops: " max-ops)
+                    (execution-result res t0 colls sigs)))))
+          res)))))
 
 (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))
+(defn signal-sssp
+  [state dist] (if-let [v (::val state)] (+ dist v)))
+
+(defn collect-sssp
+  [vertex]
+  (swap! (:state vertex)
+         (fn [{:keys [val] :as state}]
+           (update state ::val
+                   (if val
+                     (fn [_] (reduce min _ (::uncollected state)))
+                     (fn [_] (reduce min (::uncollected state))))))))
+
+(def a (eager-async-vertex (add-vertex! g {::val 0   ::collect-fn collect-sssp}) ctx))
+(def b (eager-async-vertex (add-vertex! g {::val nil ::collect-fn collect-sssp}) ctx))
+(def c (eager-async-vertex (add-vertex! g {::val nil ::collect-fn collect-sssp}) ctx))
+(def d (eager-async-vertex (add-vertex! g {::val nil ::collect-fn collect-sssp}) ctx))
+
+(connect-to! a b signal-sssp 1)
+(connect-to! a c signal-sssp 3)
+(connect-to! b c signal-sssp 1)
+(connect-to! c d signal-sssp 1)
+
+(prn @(execute! ctx g))
+
+(prn @a)
+(prn @b)
+(prn @c)
+(prn @d)
+
 
 (comment
   (disconnect! a)
   (disconnect! b)
   (disconnect! c)
+  (disconnect! d)
   )

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

(def default-vertex-state
  {:uncollected       []
   :signal-map        {}
   :score-collect     default-score-collect
   :score-signal      default-score-signal
   :collect-fn        default-collect})

(def default-context-opts
  {:collect-thresh 0
   :signal-thresh  0
   :collections    0
   :signals        0
   :timeout        1000
   :max-ops        1e6})

(defn vertex


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

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

(defn default-score-signal
  "Computes vertex signal score. Returns 0 if state's ::val
  equals :last-sig-state, else returns 1."
  [{:keys [state last-sig-state]}]
  (if (= (::val @state) @last-sig-state) 0 1))

(defn default-score-collect
  "Computes vertex collect score, here simply the number
  of ::uncollected signal values."
  [vertex] (-> vertex :state deref ::uncollected count))

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

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

(defrecord Vertex [id state last-sig-state in outs]
  clojure.lang.IDeref
  (deref [_] (::val @state))
  IVertex
  (collect! [_] ((::collect-fn @state) _))
  (score-collect [_] ((::score-collect-fn @state) _))
  (connect-to!
    [_ v sig-fn opts]
    (info (format "%d connecting to %d (%s)" id (:id v) (pr-str opts)))
    (swap! outs assoc v [sig-fn opts]) _)
  (disconnect-from!
    [_ v]
    (info (format "%d disconnecting from %d" id (:id v)))
    (swap! outs dissoc v)
    _)
  (disconnect! [_]
    (info (format "%d disconnecting..." id))
    (close! in)
    _)
  (input-channel
    [_] in)
  (score-signal [_] ((::score-signal-fn @state) _))
  (signal!
    [_]
    (let [state @state]
      (reset! last-sig-state (::val state))
      (go-loop [outs @outs]
        (let [[v [f opts]] (first outs)]
          (when v
            (let [signal (f state opts)]
              (if-not (nil? signal)
                (>! (input-channel v) [id signal])
                (warn (format "signal fn for %d return nil, skipping..." (:id v)))))
            (recur (next outs)))))
      _))
  (receive-signal
