Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Handle downstream reduced in a multiplexed complete #33

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Handle downstream reduced in a multiplexed complete #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
04c577a
Avoid completing multiplex xforms multiple times
zmthy Oct 23, 2019
1a3cbda
Handle downstream reduced in multiplexed complete
zmthy Oct 23, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 26 additions & 12 deletions src/net/cgrand/xforms.cljc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,7 +348,10 @@
@acc) ; downstream is done, propagate
(do
(vswap! m assoc! k nop-rf)
(krf @acc))) ; TODO think again
(let [acc (krf @acc)]
(when (reduced? acc)
(vreset! m (transient {})))
acc)))
acc))))
(let [kfn (or kfn key')
vfn (or vfn val')]
Expand All @@ -366,7 +369,10 @@
@acc) ; downstream is done, propagate
(do
(vswap! m assoc! k nop-rf)
(krf @acc)))
(let [acc (krf @acc)]
(when (reduced? acc)
(vreset! m (transient {})))
acc)))
acc)))))))))))

(defn into-by-key
Expand Down Expand Up @@ -657,7 +663,7 @@
(transduce (comp xform count) rf/last coll)))

(defn multiplex
"Returns a transducer that runs several transducers (sepcified by xforms) in parallel.
"Returns a transducer that runs several transducers (specified by xforms) in parallel.
If xforms is a map, values of the map are transducers and keys are used to tag each
transducer output:
=> (into [] (x/multiplex [(map inc) (map dec)]) (range 3))
Expand All @@ -674,40 +680,48 @@
xforms)
(into #{} (map #(% mrf)) xforms)))
invoke-rfs (if (map? xforms)
(fn [acc invoke]
(fn [acc step? invoke]
(reduce-kv
(fn [acc tag rf]
(let [acc (invoke rf acc)]
(if (reduced? acc)
(if (and step? (reduced? acc))
(if (reduced? @acc)
(do
(vreset! rfs nil)
acc) ; downstream is done, propagate
(do (vswap! rfs dissoc tag) (rf @acc)))
(do (vswap! rfs dissoc tag)
(let [acc (rf @acc)]
(when (reduced? acc)
(vreset! rfs nil))
acc)))
acc)))
acc @rfs))
(fn [acc invoke]
(fn [acc step? invoke]
(core/reduce
(fn [acc rf]
(let [acc (invoke rf acc)]
(if (reduced? acc)
(if (and step? (reduced? acc))
(if (reduced? @acc)
(do
(vreset! rfs nil)
acc) ; downstream is done, propagate
(do (vswap! rfs disj rf) (rf @acc)))
(do (vswap! rfs disj rf)
(let [acc (rf @acc)]
(when (reduced? acc)
(vreset! rfs nil))
acc)))
acc)))
acc @rfs)))]
(kvrf
([] (rf))
([acc] (rf (invoke-rfs acc #(%1 %2))))
([acc] (rf (invoke-rfs acc false #(%1 %2))))
([acc x]
(let [acc (invoke-rfs acc #(%1 %2 x))]
(let [acc (invoke-rfs acc true #(%1 %2 x))]
(if (zero? (core/count @rfs))
(ensure-reduced acc)
acc)))
([acc k v]
(let [acc (invoke-rfs acc #(%1 %2 k v))]
(let [acc (invoke-rfs acc true #(%1 %2 k v))]
(if (zero? (core/count @rfs))
(ensure-reduced acc)
acc)))))))
Expand Down
80 changes: 78 additions & 2 deletions test/net/cgrand/xforms_test.cljc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
(ns net.cgrand.xforms-test
(:require [clojure.test :refer [is deftest testing]]
[net.cgrand.xforms :as x]))
[net.cgrand.xforms :as x]
[net.cgrand.xforms.rfs :as rf]))

(defn trial
"A transducing context for testing that transducers are well-behaved towards
Expand Down Expand Up @@ -139,4 +140,79 @@
(is (= (range 100) (x/into [] (x/sort) (shuffle (range 100)))))
(is (= (reverse (range 100)) (x/into [] (x/sort >) (shuffle (range 100)))))
(is (= (sort-by str (range 100)) (x/into [] (x/sort-by str) (shuffle (range 100)))))
(is (= (sort-by str (comp - compare) (range 100)) (x/into [] (x/sort-by str (comp - compare)) (shuffle (range 100))))))
(is (= (sort-by str (comp - compare) (range 100)) (x/into [] (x/sort-by str (comp - compare)) (shuffle (range 100))))))

(deftest by-key
(testing "Respects reduced from multiplexed reductions and from the downstream reduction."
(is (= {:x 3 :y 5}
(into {}
(x/by-key (x/reduce rf/some))
[[:x 3] [:y 5] [:x 2]]))
"Respects multiplexed reduced.")

(is (= [:x 4]
(transduce
(x/by-key (map inc))
rf/some
[[:x 3] [:y 5] [:x 2]]))
"Respects downstream reduced.")

(is (= [:y 4]
(transduce
(x/by-key (x/reduce (fn
([] 0)
([sum] sum)
([sum x]
(let [sum (+ sum x)]
(if (> sum 4) (reduced 4) sum))))))
rf/some
[[:x 3] [:y 5] [:x 2]]))
"Respects reduced in downstream complete.")))

(deftest multiplex
(testing "Respects reduced from multiplexed reductions and from the downstream reduction."
(testing "Respects multiplexed reduced."
(is (= [3]
(into []
(x/multiplex [(x/reduce rf/some)])
[3 5 2])))
(is (= {:x 3}
(into {}
(x/multiplex {:x (x/reduce rf/some)})
[3 5 2]))))

(testing "Respects downstream reduced."
(is (= 4
(transduce
(x/multiplex [(map inc)])
rf/some
[3 5 2])))
(is (= [:x 4]
(transduce
(x/multiplex {:x (map inc)})
rf/some
[3 5 2]))))

(testing "Doesn't repeat multiplexed completion."
(is (= 2
(transduce
(x/multiplex [(x/reduce rf/last)])
rf/some
[3 5 2])))
(is (= [:x 2]
(transduce
(x/multiplex {:x (x/reduce rf/last)})
rf/some
[3 5 2]))))

(testing "Respects reduced in downstream complete."
(is (= 3
(transduce
(x/multiplex [(x/reduce rf/some) (x/reduce rf/some)])
rf/some
[3 5 2])))
(is (= 3
(transduce
(x/multiplex {:x (x/reduce rf/some) :y (x/reduce rf/some)})
(completing rf/some second)
[3 5 2]))))))