#lang racket
(require racket/future)

;; reduce : (Y X -> Y) Y (vectorof X) num num -> Y
;;  a.k.a "vector-foldl"
(define (reduce accum init-y xvec start end)
  (for/fold ([y init-y]) ([x (in-vector xvec start end)])
    (accum y x)))

;; parallel-reduce : num (Y Y -> Y)
;;                   (Y X -> Y) Y (vectorof X) num num -> Y
;;  Same as reduce when the array segment is smaller
;;  than `M', otherwise runs reductions in paralle
;;  and combines reduction results with `combine'
(define (parallel-reduce M combine accum init-y xvec start end)
  (cond
    [(< (- end start) M)
     (reduce accum init-y xvec start end)]
    [else
     (define midpoint (quotient (+ start end) 2))
     (define y-left-future
       (future
        (lambda ()
          (parallel-reduce M combine accum init-y xvec start midpoint))))
     (define y-right
       (parallel-reduce M combine accum init-y xvec midpoint end))
     (combine (touch y-left-future) y-right)]))

;; ----------------------------------------

(define xvec (make-vector 6000 1))
(define M 500)

;(time (reduce + 0 xvec 0 (vector-length xvec)))
;(time (parallel-reduce M + + 0 xvec 0 (vector-length xvec)))

;; - - - - - - - - - - - - - - - - - - - -

(define vec2 (make-vector 100 "apple"))
(define M2 10)

(define (slow-strlen a b)
  (if (= (for/fold ([v 0]) ([i (in-range 10000000)]) i) 0)
      a
      (+ a (string-length b))))
(define (plus a b) (+ a b))

;(time (reduce slow-strlen  0 vec2 0 (vector-length vec2)))
;(time (parallel-reduce M2 plus slow-strlen 0 vec2 0 (vector-length vec2)))

;; ----------------------------------------

;; scan : (Y X -> Y) (Y -> Z) Y (vectorof X) (vectorof Z) num num -> Y
;;  Like reduces, except that the result accumulated so far (roughly) for each
;;  input index is saved in an output vector. More precisely, the accumulated
;;  result is transformed by `gen' to save in the result vector.
(define (scan accum gen init-y xvec zvec start end)
  (for/fold ([y init-y]) ([i (in-range start end)])
    (let* ([x (vector-ref xvec i)]
           [y (accum y x)])
      (vector-set! zvec i (gen y))
      y)))

;; parallel-scan* : num (Y Y -> Y) (Y X -> Y) Y (vectorof X) num num 
;;                  -> (values Y (Y (Y-> Z) (vectorof Z) -> void))
;;  Used to build a parallel version of `scan'. The first phase is
;;  a parallel reduce, and a function is returned for the second phase,
;;  which computes the result-vector content given the accumulated value
;;  from the preceding part of the array.
(define (parallel-scan* M combine accum init-y xvec start end)
  (cond
    [(< (- end start) M)
     (values 
      (reduce accum init-y xvec start end)
      (lambda (pre-y gen zvec)
        (scan accum gen pre-y xvec zvec start end)
        (void)))]
    [else
     (define midpoint (quotient (+ start end) 2))
     (define left-future
       (future
        (lambda ()
          (parallel-scan* M combine accum init-y xvec start midpoint))))
     (define-values (y-right finish-right)
       (parallel-scan* M combine accum init-y xvec midpoint end))
     (define-values (y-left finish-left) (touch left-future))
     (let ([y (combine y-left y-right)])
       (values y
               (lambda (pre-y gen zvec)
                 (define left-future
                   (future (lambda () (finish-left pre-y gen zvec))))
                 (finish-right (combine pre-y y-left) gen zvec)
                 (touch left-future)
                 (void))))]))

;; parallel-scan : num (Y Y -> Y) (Y X -> Y) (Y -> Z) Y (vectorof X) 
;;                 (vectorof Z) num num -> Y
;;  Like `scan' but parallelizes both the reduce and result phases
;;  --- usefully if `gen' is relatively expensive
(define (parallel-scan M combine accum gen init-y xvec zvec start end)
  (define-values (y finish)
    (parallel-scan* M combine accum init-y xvec start end))
  (finish init-y gen zvec)
  y)

;; ----------------------------------------

(define zvec (make-vector (vector-length xvec)))

;(time (scan plus number->string 0 xvec zvec 0 (vector-length xvec)))
;(time (parallel-scan M plus plus number->string 0 xvec zvec 0 (vector-length xvec)))

;; ----------------------------------------

;; map-reduce : num (X -> Z) (Y Y -> Y) (Y Z -> Y) Y (vectorof X) num num -> Y
(define (map-reduce M xform combine accum init-y xvec start end)
  (define len (- end start))
  (define futures
    (for/vector #:length len ([x (in-vector xvec start end)])
      (future (lambda () (xform x)))))
  (define zvec
    (for/vector ([z-future (in-vector futures)])
      (touch z-future)))
  (parallel-reduce M combine accum init-y zvec 0 len))

;(time (map-reduce M - + + 0 xvec 0 (vector-length xvec)))