     :in       (atom nil)
     :outs     (atom {})}))
 
-(defrecord InMemoryGraph [state]
+(defn notify-watches
+  [watches evt]
+  (loop [watches (vals (watches (first evt)))]
+    (when watches
+      ((first watches) evt)
+      (recur (next watches)))))
+
+(defrecord InMemoryGraph [state watches]
   IComputeGraph
   (add-vertex!
     [_ val vspec]
-    (-> state
-        (swap!
-         (fn [state]
-           (let [id (:next-id state)
-                 v  (vertex id val vspec)]
-             (-> state
-                 (update :next-id inc)
-                 (update :vertices assoc id v)
-                 (assoc  :curr-vertex v)))))
-        :curr-vertex))
+    (let [v (-> state
+                (swap!
+                 (fn [state]
+                   (let [id (:next-id state)
+                         v  (vertex id val vspec)]
+                     (-> state
+                         (update :next-id inc)
+                         (update :vertices assoc id v)
+                         (assoc  :curr-vertex v)))))
+                :curr-vertex)]
+      (notify-watches @watches [:add-vertex v])
+      v))
   (remove-vertex!
     [_ v]
     (when (get-in @state [:vertices (:id v)])
       (swap! state update :vertices dissoc (:id v))
       (disconnect-all! v)
+      (notify-watches @watches [:remove-vertex v])
       true))
   (vertex-for-id
     [_ id] (get-in @state [:vertices id]))

        [clojure.core.reducers :as r]
        [cljs.core.async :as a :refer [chan close! <! >! alts! timeout]])]))

;; #?(:clj (taoensso.timbre/set-level! :warn))

;; (warn :free-ram (.freeMemory (Runtime/getRuntime)))


  [type out colls sigs t0 & [opts]]
  (go (>! out (execution-result type colls sigs t0 opts))))

(defn sync-signal-vertices
  [vertices thresh]
  (loop [sigs 0, verts vertices]
    (if verts
      (let [v (first verts)]
        (if (should-signal? v thresh)
          (do (debug (:id v) "signaling")
              (signal! v sync-vertex-signal)
              (recur (inc sigs) (next verts)))
          (recur sigs (next verts))))
      sigs)))

(defn sync-collect-vertices
  [vertices thresh]
  (loop [colls 0, verts vertices]
    (if verts
      (let [v (first verts)]
        (if (should-collect? v thresh)
          (do (debug (:id v) "collecting")
              (collect! v)
              (recur (inc colls) (next verts)))
          (recur colls (next verts))))
      colls)))

(defn parallel-sync-signal-vertices
  [vertices thresh]
  (r/fold
   +
   (fn
     ([] 0)
     ([acc v] (if (should-signal? v thresh) (do (signal! v sync-vertex-signal) (inc acc)) acc)))
   vertices))

(defn parallel-sync-collect-vertices
  [vertices thresh]
  (r/fold
   +
   (fn
     ([] 0)
     ([acc v] (if (should-collect? v thresh) (do (collect! v) (inc acc)) acc)))
   vertices))

(def default-sync-context-opts
  {:collect-thresh 0
   :signal-thresh  0
   :max-iter       1e6
   :signal-fn      sync-signal-vertices
   :collect-fn     sync-collect-vertices})

(defn sync-execution-context
  [opts]
  (let [ctx       (merge default-sync-context-opts opts)
        g         (:graph ctx)
        c-thresh  (:collect-thresh ctx)
        s-thresh  (:signal-thresh ctx)
        coll-fn   (:collect-fn ctx)
        sig-fn    (:signal-fn ctx)
        max-iter  (:max-iter ctx)
        watch-id  (keyword (gensym))]
    (reify
      #?@(:clj
           [clojure.lang.IDeref
            (deref [_] ctx)]
           :cljs
           [IDeref
            (-deref [_] ctx)])
      IGraphExecutor
      (stop! [_] (err/unsupported!))
      (notify! [_ evt] (err/unsupported!))
      (include-vertex! [_ v] (err/unsupported!))
      (execute!
        [_]
        (add-watch! g :remove-vertex watch-id (fn [evt] (signal! (peek evt) sync-vertex-signal)))
        (let [t0 (now)]
          (loop [i 0, colls 0, sigs 0]
            (if (<= i max-iter)
              (let [verts (vertices g)
                    sigs' (sig-fn verts s-thresh)
                    colls' (coll-fn verts c-thresh)]
                (if (and (pos? sigs') (pos? colls'))
                  (recur (inc i) (long (+ colls colls')) (long (+ sigs sigs')))
                  (do (remove-watch! g :remove-vertex watch-id)
                      (execution-result :converged colls sigs t0 {:iterations i}))))
              (do (remove-watch! g :remove-vertex watch-id)
                  (execution-result :max-iter-reached colls sigs t0 {:iterations i})))))))))

(defn stop-async-execution
  [bus vertices]
  (info "stopping execution context...")
