I was wondering if there is any reason why r/take-while, r/take and r/drop in reducers don't use folder (instead of reducer)? With the following practical effects:

(require '[clojure.core.reducers :as r])
(time (->> (range 50000)
       (into [])
       (r/map range)
       (r/mapcat conj)
       (r/drop 0)
       (r/filter odd?)
       (r/fold +)))
;; "Elapsed time: 45516.963356 msecs"
;; 10416041675000
(time (->> (range 50000)
       (into [])
       (r/map range)
       (r/mapcat conj)
       (r/filter odd?)
       (r/fold +)))
;; "Elapsed time: 9190.562896 msecs"
;; 10416041675000

they are stateful, and the state results from a linear scan, which doesn't parallelize as a tree

not sure I get that. The reduction happens on a core on a single chunk. So are the results going to be different if the drop is foldable?

think about how take-while would behave if processing 2 chunks in parallel

as hiredman says, they are inherently linear operations

(def v (range 100))

(= (->> v
        (partition-all 32)
        (mapcat f))
   (drop 1 v))

is there some f that can safely be called in parallel that would result in true?

Still trying to understand. Created:

(defn drop2
  [n coll]
  (r/folder
    coll (fn [rf]
      (let [nv (volatile! n)]
        (fn
          ([result input]
           (let [n @nv]
             (vswap! nv dec)
             (if (pos? n)
               result
               (rf result input)))))))))

that has a clear race

volatiles are not atomic

it has many races

^

and it assumes things are happening in order, which by definition when running in parallel it is not

actually, the volatile doesn't appear to escape each partition, so it's not a race, but it's not tracking a global count

anywhere the docstring says "stateful transducer" you have to be cognizant of threads.

right, you end up dropping from each partition vs. dropping from the whole thing, which is an entirely different behavior from the non-folding drop

drop2 returns the same results as r/drop, it's performing in parallel (as in fork-join parallel) but timings are much longer.

it doesn't

if you think it does then your test isn't correct

for example you are pushing in something that doesn't fold in parallel (like a seq) or you are dropping 0 which does nothing, etc

or you are witnessing one possible interleaving

If I (println (Thread/currentThread)) from inside (fn [result input]) I see fork-join threads. I thought that indicated going fork-join correctly.

you'll be in a FJPool, but the code is not correct. What is your correctness test?

Perhaps we can tweak that to expose the problem

I'm executing the fold at the beginning alternating r/drop and drop2 with different initial ranges. I was expecting different results (even repeating the same) in case of race conditions. Not sure that's good definition of correctness

user=> (require '[clojure.core.reducers :as r])
nil
(defn drop2
  [n coll]
  (r/folder
    coll (fn [rf]
      (let [nv (volatile! n)]
        (fn
          ([result input]
           (let [n @nv]
             (vswap! nv dec)
             (if (pos? n)
               result
               (rf result input)))))))))
#'user/drop2
user=> 
user=> (def x (vec (range 10000)))
#'user/x
user=> (def y (->> x (drop2 10) (r/fold +)))
#'user/y
user=>  (def z (->> x (drop 10) (reduce +)))
#'user/z
user=> (= z y)
false
user=> z
49994955
user=> y
49969955
user=> 

nice

there's a devequivtest macro in the test suite that checks for seq/reducer equivalence https://github.com/clojure/clojure/blob/master/test/clojure/test_clojure/reducers.clj

it or something like it may be helpful

also interesting that (->> x (r/drop 10) (r/fold +)) is 49994955

instead of dropping the first 10 numbers, it is dropping 10 from every partition

which is what I would expect

no, it's just the first 10 in those two examples

it's the 49969955 y result that is erroneous

yeah, no, that is not the contract for r/drop (and the others)

with pure functions, the behavior for reducers is expected to be exactly the same

so you think it is possible because you want to solve an easier problem 🙂

with most of c.c.reducers effectively obsoleted by transducers, I think it would be a useful exercise to translate some fold examples to use transducers instead

perhaps that will illustrate some flaws in fold that need adaptation.

(Like completion)

(r/fold ((drop 10) +) (vec (range 10000))) looks like it's returning different results each run?

the transducer is applied at the wrong time

that will work for something like map (stateless)

fold needs to be in charge of transforming reducing functions

and do it as it partitions

what is happening in your scenario: you create one reducing function ((drop 10) +), it's passed to a bunch of threads that all use it in a non-specified order

they all racily/non-atomically modify the volatile

it's kind of what I was expecting for that drop2 after thinking twice

and yeah, same thing. Also that drop2 suffers the same

starting to connect the dots, thanks for your time

my pleasure. the relationships between Transducers / Reducers / Fold have not been made explicit

I think the "problem" is work stealing. A chunk can always fly on a thread where the volatile doesn't have the right count, since it already processed a chunk

huh?

there is no way to make this work with volatiles because they lack atomicity

not even when the volatile is correctly closed over and the function allocated every time?

then it wouldn't be a global count, but a local count

it would be a different function