 
 
 (comment
-  (disconnect! a)
-  (disconnect! b)
-  (disconnect! c)
-  (disconnect! d)
+  (close-input! a)
+  (close-input! b)
+  (close-input! c)
+  (close-input! d)
   )

       (merge opts)
       (deliver p)))

(defn async-context
  [opts]
  (let [ctrl     (chan)
        ctx      (merge default-context-opts {:ctrl ctrl} opts)
        c-thresh (:collect-thresh ctx)
        s-thresh (:signal-thresh ctx)
        max-ops  (:max-ops ctx)
        max-t    (:timeout ctx)]
    (reify
      clojure.lang.IDeref
      (deref [_] ctx)
      IGraphExecutor
      (execute! [_ g]
        (let [t0  (System/nanoTime)
              res (promise)]
          (go
            (loop [verts (vertices g)]
              (when-let [v (first verts)]
                (when (should-collect? v c-thresh)
                  (collect! v))
                (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)
                    (execution-result :converged res colls sigs t0)
                    (case evt
                      :collect (recur (inc colls) sigs)
                      :signal  (recur colls (inc sigs))
                      :error   (do (warn ex "@ vertex" v)
                                   (execution-result
                                    :error res colls sigs t0
                                    {:reason           :error
                                     :reason-exception ex
                                     :reason-vertex    v}))
                      (do (info "wrong event" evt v)
                          (execution-result
                           :interrupted res colls sigs t0
                           {:reason        :unknown
                            :reason-event  evt
                            :reason-vertex v})))))
                (execution-result :terminated res colls sigs t0))))
          res)))))

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

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

                    (>! ctrl [:signal id])
                    (signal! vertex))))
              (recur))
          (debug (str id " stopped")))))
    vertex))

(defn execution-result
