   #?@(:clj
       [(:require
         [taoensso.timbre :refer [debug info warn]]
+        [clojure.core.reducers :as r]
         [clojure.core.async :refer [go go-loop chan close! <! >! alts! timeout]])]
       :cljs
       [(:require-macros
         [cljs.core.async.macros :refer [go go-loop]]
         [cljs-log.core :refer [debug info warn]])
        (:require
+        [clojure.core.reducers :as r]
         [cljs.core.async :as a :refer [chan close! <! >! alts! timeout]])]))
 
 ;; #?(:clj (taoensso.timbre/set-level! :warn))

          (recur colls (next verts))))
      colls)))

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

(defn sync-execution-context
  [opts]
  (let [ctx       (merge default-sync-context-opts opts)
        c-thresh  (:collect-thresh ctx)
        s-thresh  (:signal-thresh ctx)
        max-ops   (:max-ops ctx)]
    (reify
      #?@(:clj

              verts (vertices (:graph ctx))]
          (loop [colls 0, sigs 0]
            (if (<= (+ colls sigs) max-ops)
              (let [sigs' (sig-fn verts s-thresh)
                    colls' (coll-fn verts c-thresh)]
                (if (and (pos? sigs') (pos? colls'))
                  (recur (long (+ colls colls')) (long (+ sigs sigs')))
                  (execution-result :converged colls sigs t0)))
