Add parallel reducers

Wildly incomplete.

Author: jackfirth

streaming/reducer2/base.rkt

@@ -0,0 +1,55 @@
+#lang racket
+
+
+(provide
+ (contract-out
+  #:unprotected-submodule no-contract
+  [reducer? (-> any/c boolean?)]
+  [make-reducer
+   (->* (#:starter (-> any/c)
+         #:accumulator (-> any/c any/c any/c))
+        (#:finisher (-> any/c any/c)
+         #:termination-checker (or/c (-> any/c boolean?) #false)
+         #:cloner (or/c (-> any/c any/c) #false)
+         #:merger (or/c (-> any/c any/c any/c) #false)
+         #:ordered? boolean?
+         #:concurrent? boolean?)
+        reducer?)]
+  [reducer-starter (-> reducer? (-> any/c))]
+  [reducer-accumulator (-> reducer? (-> any/c any/c any/c))]
+  [reducer-finisher (-> reducer? (-> any/c any/c))]
+  [reducer-termination-checker (-> reducer? (or/c (-> any/c boolean?) #false))]
+  [reducer-cloner (-> reducer? (or/c (-> any/c any/c) #false))]
+  [reducer-merger (-> reducer? (or/c (-> any/c any/c any/c) #false))]
+  [reducer-ordered? (-> reducer? boolean?)]
+  [reducer-concurrent? (-> reducer? boolean?)]))
+
+
+
+(struct reducer
+  (starter ; (-> S)
+   accumulator ; (-> S E S)
+   finisher ; (-> S R)
+   termination-checker ; (or/c (-> S boolean?) #false)
+   cloner ; (or/c (-> S S) #false)
+   merger ; (or/c (-> S S S) #false)
+   ordered? ; boolean?
+   concurrent?)) ; boolean?
+
+
+(define (make-reducer #:starter starter
+                      #:accumulator accumulator
+                      #:finisher [finisher values]
+                      #:termination-checker [termination-checker #false]
+                      #:cloner [cloner #false]
+                      #:merger [merger #false]
+                      #:ordered? [ordered? #true]
+                      #:concurrent? [concurrent? #false])
+  (reducer starter
+           accumulator
+           finisher
+           termination-checker
+           cloner
+           merger
+           ordered?
+           concurrent?))

streaming/reducer2/continuation-cloning-experiment.rkt

@@ -0,0 +1,82 @@
+#lang racket
+
+
+(define nondeterminism-prompt-tag (make-continuation-prompt-tag))
+
+
+(define-syntax-rule (all-possible-results body ...)
+  (call-with-nondeterminism (λ () body ...)))
+
+
+(define (call-with-nondeterminism proc)
+  (call-set-producer-with-nondeterminism
+   (λ () (set (proc)))))
+
+
+(define (call-set-producer-with-nondeterminism set-producer)
+  (call-with-continuation-prompt
+   set-producer
+   nondeterminism-prompt-tag
+   (λ (k)
+     (set-union
+      (call-set-producer-with-nondeterminism (λ () (k #true)))
+      (call-set-producer-with-nondeterminism (λ () (k #false)))))))
+
+
+(define (flip)
+  (call-with-composable-continuation
+   (λ (k) (abort-current-continuation nondeterminism-prompt-tag k))
+   nondeterminism-prompt-tag))
+
+
+#;(all-possible-results
+   (list (if (flip) 'a 'b)
+         (if (flip) 'a 'b)))
+
+
+(define (dynamic-log proc)
+  (dynamic-wind
+   (λ () (displayln "going in"))
+   proc
+   (λ () (displayln "coming out"))))
+
+
+#;(all-possible-results
+   (dynamic-log
+    (λ ()
+      (for/sum ([_ (in-range 3)])
+        (if (flip) 1 2)))))
+
+
+(define (flip-sum n)
+  (define total (box 0))
+  (for ([_ (in-range n)])
+    (set-box! total (+ (if (flip) 1 2) (unbox total))))
+  (unbox total))
+
+
+#;(all-possible-results
+   (dynamic-log
+    (λ ()
+      (flip-sum 3))))
+
+
+(module+ main
+  (all-possible-results
+   (define total #f)
+   (define stack (list 0))
+   (define done? #false)
+   (dynamic-wind
+    (λ ()
+      (set! total (first stack))
+      (set! done? #false))
+    (λ ()
+      (for ([_ (in-range 3)])
+        (define next (if (flip) 1 2))
+        (set! total (+ next total)))
+      (set! done? #true)
+      (set! stack (rest stack))
+      total)
+    (λ ()
+      (unless done?
+        (set! stack (cons total stack)))))))

streaming/reducer2/into-any-match.rkt

@@ -0,0 +1,25 @@
+#lang racket/base
+
+
+(require racket/contract/base)
+
+
+(provide
+ (contract-out
+  #:unprotected-submodule unchecked
+  [into-any-match? (-> (-> any/c boolean?) reducer?)]))
+
+
+(require (submod rebellion/streaming/reducer2/base no-contract)
+         rebellion/streaming/reducer2/reducer-map
+         rebellion/streaming/reducer2/into-binary-fold)
+
+
+;@----------------------------------------------------------------------------------------------------
+
+
+(define (into-any-match? predicate)
+  (reducer-map (into-binary-fold (λ (a b) (or a b)) #false
+                                 #:termination-checker values
+                                 #:ordered? #false)
+               #:domain predicate))

streaming/reducer2/into-binary-fold.rkt

@@ -0,0 +1,21 @@
+#lang racket/base
+
+
+(require (submod rebellion/streaming/reducer2/base no-contract))
+
+
+(provide into-binary-fold)
+
+
+(define (into-binary-fold f x
+                          #:finisher [finisher values]
+                          #:ordered? [ordered? #true]
+                          #:termination-checker [termination-checker #false])
+  (make-reducer #:starter (λ () x)
+                #:accumulator (λ (s e) (f e s))
+                #:finisher finisher
+                #:termination-checker termination-checker
+                #:merger f
+                #:cloner values
+                #:ordered? ordered?
+                #:concurrent? #true))

streaming/reducer2/into-list.rkt

@@ -0,0 +1,13 @@
+#lang racket
+
+
+(require rebellion/streaming/reducer2/base)
+
+
+(define into-list
+  (make-reducer #:starter (λ () '())
+                #:accumulator (λ (s e) (cons e s))
+                #:finisher reverse
+                #:cloner values
+                #:merger (λ (left right) (append right left))
+                #:concurrent? #true))

streaming/reducer2/into-product.rkt

@@ -0,0 +1,11 @@
+#lang racket/base
+
+
+(require rebellion/streaming/reducer2/base
+         rebellion/streaming/reducer2/into-binary-fold)
+
+
+(provide into-product)
+
+
+(define into-product (into-binary-fold * 1 #:ordered? #false))

streaming/reducer2/into-sum.rkt

@@ -0,0 +1,11 @@
+#lang racket/base
+
+
+(require racket/unsafe/ops
+         rebellion/streaming/reducer2/into-binary-fold)
+
+
+(provide into-sum)
+
+
+(define into-sum (into-binary-fold unsafe-fx+ 0 #:ordered? #false))

streaming/reducer2/manual-performance-test.rkt

@@ -0,0 +1,8 @@
+#lang racket/base
+
+(require racket/unsafe/ops
+         rebellion/streaming/reducer2/into-sum
+         rebellion/streaming/reducer2/vector-reduce)
+
+(let ([vec (build-vector 100000000 values)])
+  (time (vector-parallel-reduce vec into-sum)))

streaming/reducer2/manual-performance-test2.rkt

@@ -0,0 +1,23 @@
+#lang racket/base
+
+
+(require rebellion/streaming/reducer2/parallel-reduce
+         rebellion/streaming/reducer2/into-sum)
+
+
+(collect-garbage)
+(collect-garbage)
+(collect-garbage)
+
+
+(let ([vec (build-vector 100000000 values)])
+  (time (parallel-reduce vec into-sum)))
+
+
+(collect-garbage)
+(collect-garbage)
+(collect-garbage)
+
+
+(let ([vec (build-vector 100000000 values)])
+  (time (for/sum ([x (in-vector vec)]) x)))

streaming/reducer2/manual-performance-test3.rkt

@@ -0,0 +1,33 @@
+#lang racket
+
+
+
+(require rebellion/streaming/reducer2/parallel-reduce
+         rebellion/streaming/reducer2/sequential-reduce
+         (submod rebellion/streaming/reducer2/into-any-match unchecked))
+
+
+
+(define size 1000000000)
+
+(collect-garbage)
+(collect-garbage)
+(collect-garbage)
+
+(let ([vec (build-vector size values)])
+  (time (parallel-reduce vec (into-any-match? (λ (x) (> x (/ size 2)))))))
+
+(collect-garbage)
+(collect-garbage)
+(collect-garbage)
+
+(let ([vec (build-vector size values)])
+  (time (for/or ([x (in-vector vec)]) (> x (/ size 2)))))
+
+
+(collect-garbage)
+(collect-garbage)
+(collect-garbage)
+
+(let ([vec (build-vector size values)])
+  (time (sequential-reduce vec (into-any-match? (λ (x) (> x (/ size 2)))))))