 
 (defn parallel-signal-pass
   [thresh]
-  (fn [vertices]
+  (fn [workgroup-size vertices]
     (r/fold
+     workgroup-size
      sub-pass-combine
      (fn [acc v]
        (if (should-signal? v thresh)


(defn parallel-collect-pass
  [thresh]
  (fn [vertices]
    (r/fold
     sub-pass-combine
     (fn [acc v]
       (if (should-collect? v thresh)


(defn two-pass-scheduler
  [ctx]
  ((:processor ctx) (:signal-thresh ctx) (:collect-thresh ctx)))

(defn add-context-watches
  [g watch-id queue]
