2025-07-30 missionary | Clojure Slack Archive

missionary 2025-07-30

braai engineer 2025-07-30T19:22:19.077139Z

Would it be possible to implement my lazy-merge-dedupe-sort function in Missionary if passing in streams instead of seqs, and would it yield any performance improvement over Clojure seqs? https://github.com/CloudAfrica/eacl/blob/main/src/eacl/lazy_merge_sort.clj#L3-L24 (example usage and expected output at bottom of file) This function is used heavily in https://github.com/theronic/eacl to return a merged, sorted & deduplicated stream of potentially duplicate resources emitted by parallel paths through the permission graph which resolves to reading over sorted index ranges from a Relationship tuple index, which is sorted by either subject or resource (depending on whether subject or resource is known).

👀 1

leonoel 2025-08-01T08:48:51.160909Z

First attempt, untested with unsorted inputs

(defn windowed-dedupe-sort [sentinel size]
  (fn [rf]
    (let [state (doto (object-array 2)
                  (aset 0 (sorted-map))
                  (aset 1 size))]
      (fn
        ([] (rf))
        ([r] (rf (reduce rf r (keys (aget state 0)))))
        ([r x]
         (let [n (dec (aget state 1))
               m (aget state 0)
               m (if (= x sentinel)
                   m (assoc m x (inc (get m x 0))))]
           (if (zero? n)
             (if-some [[x n] (first m)]
               (do (aset state 0 (dissoc m x))
                   (aset state 1 n)
                   (rf r x)) r)
             (do (aset state 0 m)
                 (aset state 1 n) r))))))))

(defn cycle [init f & args]
  (m/ap (let [!x (atom init)]
          (reset! !x (m/?> (apply f (m/watch !x) args))))))

(defn merge-dedupe-sort [feedback flows]
  (m/eduction (windowed-dedupe-sort ::done (count flows))
    (m/ap (let [flow (m/?> (count flows) (m/seed flows))]
            (m/amb (let [x (m/?> flow)]
                     (m/amb x
                       (do (m/?> (m/eduction (take-while (partial > x)) feedback))
                           (m/amb)))) ::done)))))

(comment
  ((m/reduce conj []
     (cycle 0 merge-dedupe-sort
       [(m/seed [1 3 5 7 9])
        (m/seed [2 3 6 8 9 10])
        (m/seed [1 4 5 6 11])]))
   prn prn)
  ;; [1 2 3 4 5 6 7 8 9 10 11]
  )

braai engineer 2025-08-01T08:57:15.225029Z

Thanks! Will test 🙂. Any idea if it will offer any performance benefit? There might be an opportunity for volatile! in either impl.

leonoel 2025-08-01T09:06:06.573429Z

it depends how the current input seqs are implemented, if they rely on OS thread blocking then switching to async should scale better

leonoel 2025-08-01T12:11:40.406239Z

this version is also more concurrent, which can only improve perf, although unclear to what extent

Clojurians Log v2

missionary 2025-07-30