Reducers

// http://clojure.com/blog/2012/05/08/reducers-a-library-and-model-for-collection-processing.html

#r "FSharp.PowerPack.Parallel.Seq.dll"

module Reducer =
    open System 
    open System.Text
    open System.Collections.Generic
    open System.Linq
    
    type ReduceFunc<'T,'R> =
        abstract Invoke : 'T * 'R -> 'R

    type CombineFunc<'R> =
        abstract Invoke : 'R * 'R -> 'R    

    type Reducer<'T> = 
        abstract Apply<'R> : ReduceFunc<'T, 'R> * CombineFunc<'R> * (unit -> 'R) -> 'R 
        
    // helper functions
    let inline toReduceFunc (f : 'T -> 'R -> 'R) =
        { 
            new ReduceFunc<'T, 'R> with 
                member self.Invoke(value : 'T, acc : 'R) = f value acc 
        }
        
    let inline toCombineFunc (f : 'R -> 'R -> 'R) =
        { 
            new CombineFunc<'R> with 
                member self.Invoke(left : 'R, right : 'R) = f left right 
        }

    // executor functions
    let inline toSeqReducer (values : seq<'T>) : Reducer<'T> =
        { 
            new Reducer<'T> with 
                member self.Apply<'R>(rf : ReduceFunc<'T, 'R>, _, init : unit -> 'R) : 'R =  
                    let mutable r = init ()
                    for value in values do
                        r <- rf.Invoke(value, r) 
                    r
        }
        
    let inline toParallelReducer (seqReduceCount : int) (values : 'T []) : Reducer<'T> =
        { 
            new Reducer<'T> with 
                member self.Apply<'R>(rf : ReduceFunc<'T, 'R>, cf : CombineFunc<'R>, init : unit -> 'R) : 'R =  
                    let rec reduceCombine s e =
                        async { 
                            if e - s <= seqReduceCount then
                                let s' = if s > 0 then s + 1 else s
                                let result = 
                                    let mutable r = init()
                                    for i = s' to e do
                                        r <- rf.Invoke(values.[i], r) 
                                    r
                                return result
                            else 
                                let m = (s + e) / 2
                                let! result =  Async.Parallel [| reduceCombine s m; reduceCombine m e |]
                                return cf.Invoke(result.[0], result.[1])
                        }
                    reduceCombine 0 (values.Length - 1) |> Async.RunSynchronously
        }

    let inline toParallelLinqReducer (values : seq<'T>) : Reducer<'T> =
        { 
            new Reducer<'T> with 
                member self.Apply<'R>(rf : ReduceFunc<'T, 'R>, cf : CombineFunc<'R>, init : unit -> 'R) : 'R =  
                    ParallelEnumerable.Aggregate(values.AsParallel(), Func<'R>(init), 
                                                    Func<'R, 'T, 'R>(fun acc v -> rf.Invoke(v, acc)),
                                                    Func<'R, 'R, 'R>(fun left right -> cf.Invoke(left, right)), Func<'R, 'R>(id))
        }

    // transform functions
    let inline collect (f : 'A -> Reducer<'B>) (input : Reducer<'A>) : Reducer<'B> = 
            { 
            new Reducer<'B> with 
            member self.Apply<'R> (rf, cf, init) = 
                    input.Apply<'R>(toReduceFunc (fun a r -> (f a).Apply(rf, cf, fun () -> r)), cf, init)
            }        
                    
    let inline map (f : 'A -> 'B) (input : Reducer<'A>) : Reducer<'B> =
        { 
            new Reducer<'B> with 
            member self.Apply<'R> (rf, cf, init) = 
                    input.Apply<'R>(toReduceFunc (fun a r -> rf.Invoke(f a, r)), cf, init)
        }

    let inline filter (p : 'A -> bool) (input : Reducer<'A>) : Reducer<'A> =
        { 
            new Reducer<'A> with 
                member self.Apply<'R> (rf, cf, init) = 
                    input.Apply<'R>(toReduceFunc (fun a r -> if p a then rf.Invoke(a, r) else r), cf, init)
        }

    // reduce functions
    let inline reduce (reducef : 'T -> 'R -> 'R) (combineF : 'R -> 'R -> 'R) (init : (unit -> 'R)) (reducer : Reducer<'T>) : 'R = 
        reducer.Apply(toReduceFunc reducef, toCombineFunc combineF, init)

    let inline sum (reducer : Reducer<int>) : int = 
        reduce (+) (+) (fun () -> 0) reducer

    let inline length (reducer : Reducer<'T>) : int =
        reduce (fun _ r -> r + 1) (+) (fun () -> 0) reducer

    let inline concat (reducer : Reducer<string>) : string = 
        let result = 
            reduce (fun (v : string) (builder : StringBuilder) -> builder.Append(v))
                    (fun (left : StringBuilder) (right : StringBuilder) -> left.Append(right)) 
                    (fun () -> new StringBuilder())
                    reducer
        result.ToString()

    let inline toArray (reducer : Reducer<'T>) : 'T [] =
        let result = 
            reduce (fun v (list : List<'T>) -> list.Add(v); list)
                    (fun (left : List<'T>) (right : List<'T>) -> left.AddRange(right); left) 
                    (fun () -> new List<'T>())
                    reducer 
        result.ToArray()

    let inline groupBy (selectorF : 'T -> 'Key) 
                        (transformF : 'T -> 'Elem) 
                        (aggregateF : 'Key * seq<'Elem> -> 'Elem) 
                        (reducer : Reducer<'T>) : seq<'Key * 'Elem> =
        let inline reduceF (v : 'T) (r : Dictionary<'Key, List<'Elem>>) =
            let key = selectorF v
            let elem = transformF v
            if r.ContainsKey(key) then
                r.[key].Add(elem)
                let result = (key, r.[key]) |> aggregateF
                r.[key].Clear()
                r.[key].Add(result)
            else 
                r.Add(key, new List<_>([| elem |]))
            r
        let inline combineF (left : Dictionary<'Key, List<'Elem>>) (right : Dictionary<'Key, List<'Elem>>) =
            for keyValue in right do
                if left.ContainsKey(keyValue.Key) then
                    left.[keyValue.Key].AddRange(right.[keyValue.Key])
                    let result = (keyValue.Key, left.[keyValue.Key]) |> aggregateF
                    left.[keyValue.Key].Clear()
                    left.[keyValue.Key].Add(result)
                else
                    left.[keyValue.Key] <- new List<_>([| (keyValue.Key, keyValue.Value) |> aggregateF |])
            left
                    
        let result = 
            reduce  reduceF combineF
                    (fun () -> new Dictionary<'Key, List<'Elem>>())
                    reducer
        result |> Seq.map (fun keyValue -> (keyValue.Key, (keyValue.Key, keyValue.Value) |> aggregateF))

    let inline countBy (selectorF : 'T -> 'Key) (reducer : Reducer<'T>) : seq<'Key * int> =
        let inline reduceF (v : 'T) (r : Dictionary<'Key, int>) =
            let key = selectorF v
            if r.ContainsKey(key) then
                r.[key] <- r.[key] + 1
            else 
                r.[key] <- 1
            r
        let inline combineF (left : Dictionary<'Key, int>) (right : Dictionary<'Key, int>) =
            for keyValue in right do
                if left.ContainsKey(keyValue.Key) then
                    left.[keyValue.Key] <- left.[keyValue.Key] + right.[keyValue.Key]
                else
                    left.[keyValue.Key] <- keyValue.Value
            left
                    
        let result = 
            reduce  reduceF combineF
                    (fun () -> new Dictionary<'Key, int>())
                    reducer
        result |> Seq.map (fun keyValue -> (keyValue.Key, keyValue.Value))

// Example - wordcount

let lines = System.IO.File.ReadAllLines("largefile.txt")

lines
|> Reducer.toParallelReducer 10
|> Reducer.collect (fun line -> Reducer.toSeqReducer <| line.Split(' '))
|> Reducer.groupBy id (fun _ -> 1) (fun (_, items) -> Seq.sum items)