   ([acc] acc)
   ([[act as ac] [act' s c]] [(into act act') (+ as s) (+ ac c)]))
 
+(defn single-pass-scheduler
+  [workgroup-size processor vertices]
+  (r/fold workgroup-size combine-stats processor vertices))
+
+(defn two-pass-scheduler
+  [workgroup-size processor vertices]
+  )
+
 (defn default-async-context-opts
   []
   {:collect-thresh 0
    :signal-thresh  0
-   :processor      eager-async-vertex-processor
+   :processor      eager-probabilistic-vertex-processor
+   :scheduler      single-pass-scheduler
    :max-ops        1e6
    :threads        #?(:clj (inc (.availableProcessors (Runtime/getRuntime))) :cljs 1)})
 
-(defn async-execution-context
+(defn execution-context
   [opts]
   (let [ctx       (merge (default-async-context-opts) opts)
         g         (:graph ctx)

        s-thresh  (:signal-thresh ctx)
        max-ops   (:max-ops ctx)
        processor ((:processor ctx) s-thresh c-thresh)
        threads   (:threads ctx)
        watch-id  (keyword (gensym))
        v-filter  (filter #(or (should-signal? % s-thresh) (should-collect? % c-thresh)))

              [active sigs (inc colls)])
          [active sigs colls])))))

(defn eager-probabilistic-vertex-processor
  [s-thresh c-thresh]
  (fn [[active sigs colls] v]
    (let [active (conj active v)]
