             ;;(warn :active-count (count @active))
             (if (seq @active)
               (if (<= (+ colls sigs) max-ops)
-                (let [[act' sigs' colls']
-                      (r/fold
-                       (max 512 (long (/ (count @active) threads)))
-                       combine-stats
-                       processor @active)]
+                (let [grp-size            (max 512 (long (/ (count @active) threads)))
+                      [act' sigs' colls'] (scheduler grp-size processor @active)]
                   ;;(warn :auto (count @enable))
                   (reset! active (into (into #{} v-filter act') @enable))
                   (reset! enable #{})

                  (->result :converged i sigs colls t0)))
              (->result :max-iter-reached i sigs colls t0))))))))

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

  ([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-probabilistic-vertex-processor
   :scheduler      single-pass-scheduler
   :max-ops        1e6
   :threads        #?(:clj (inc (.availableProcessors (Runtime/getRuntime))) :cljs 1)})

(defn execution-context
  [opts]
  (let [ctx       (merge (default-async-context-opts) opts)
        g         (:graph ctx)
