Foldable multireducible

gistfile1.clj

;; excerpt of foldable multireducible impl
;;
;; - introducing concept of mutable iterator, which is 
;;   much faster than seq
;; - this is for a (yet unreleased) custom patched clojure, 
;;  where clojure.lang interfaces can behave like protocols
;; - map is reducer variant, thats why I have to call seq on it
;; - seq is patched to support any reducible, not just Seqable

(defn fold-sectionable
  "Perform fold on sectionable collection."
  ([coll pool n combinef reducef reduce-mode]
     (fold-sectionable coll pool n combinef reducef
                       reduce-mode section))
  ([coll pool n combinef reducef reduce-mode section]
     (fold-sectionable coll pool n combinef reducef
                       reduce-mode section count))
  ([coll pool n combinef reducef reduce-mode section count]
     (let [cnt (count coll)
           reduce (reduce-from-mode reduce-mode)
           n (max 1 n)]
       (cond
        (empty? coll) (combinef)
        (<= cnt n) (reduce reducef (combinef) coll)
        :else
        (let [split (quot cnt 2)
              v1 (section coll 0 split)
              v2 (section coll split cnt)
              fc (fn [child]
                   #(-fold child pool n combinef reducef
                           reduce-mode))]
          (invoke pool
                  #(let [f1 (fc v1)
                         t2 (fork (fc v2))]
                     (combinef (f1) (join t2)))))))))

(deftype ZipIterator [^:unsynchronized-mutable iters]
  IOpenAware
  (-open? [this] (every? open? iters))
  IReference
  (-deref [this] (apply tuple (map deref iters)))
  IIterator
  (-next! [this] (set! iters (seq (map next! iters))) this))

(deftype FoldableZip [colls]
  IRed
  (-reduce [this f init]
    (loop [iters (seq (map iterator colls))
           val init]
      (if (every? open? iters)
        (reduced-let [ret (f val (apply tuple (map deref iters)))]
          (recur (seq (map next! iters)) ret))
        val)))
  IIterable
  (-iterator [this]
    (->ZipIterator (seq (map iterator colls))))
  IUnpackedRed
  (-reduce-unpacked [this f init]
    (loop [iters (seq (map iterator colls))
           val init]
      (if (every? open? iters)
        (reduced-let [ret (apply f val (map deref iters))]
          (recur (seq (map next! iters)) ret))
        val)))
  IFoldable
  (-fold [coll pool n combinef reducef reduce-mode]
    (fold-sectionable coll pool n combinef reducef reduce-mode))
  ISectionable
  (-section [this new-begin new-end]
    (let [l (count this)
          new-end (prepare-ordered-section new-begin new-end l)]
      (if (and (zero? new-begin) (== new-end l))
        this
        (->FoldableZip (map #(section % new-begin new-end) colls)))))
  ICounted
  (-count [this] (apply min (map count colls))))