anchor.diff:          let degreeOfParallelism = ref None
           { new CloudStream<'T> with
               member self.DegreeOfParallelism = degreeOfParallelism
  -            member self.Apply<'S, 'R> (collectorf : unit -> Collector<'T, 'S>) (projection : 'S -> Cloud<'R>) (combiner : 'R -> 'R -> 'R) =
  +            member self.Apply<'S, 'R> (collectorf : Cloud<Collector<'T, 'S>>) (projection : 'S -> Cloud<'R>) (combiner : 'R -> 'R -> 'R) =
                   cloud { 
                       if Seq.isEmpty sources then 
  -                        return! projection (collectorf ()).Result
  +                        let! collector = collectorf
  +                        return! projection collector.Result
                       else
                           let! workerCount = 
                               match !self.DegreeOfParallelism with
                               | Some n -> cloud { return n }
                               | None -> Cloud.GetWorkerCount()
   
  -                        let createTask (files : CloudFile []) (collectorf : unit -> Collector<'T, 'S>) : Cloud<'R> = 
  +                        let createTask (files : CloudFile []) (collectorf : Cloud<Collector<'T, 'S>>) : Cloud<'R> = 
                               cloud {
                                   let rec partitionByLength (files : CloudFile []) index (currLength : int64) (currAcc : CloudFile list) (acc : CloudFile list list)=
                                       cloud {
  
positive.code:             }
          { new CloudStream<'Key * 'State> with
              member self.DegreeOfParallelism = stream.DegreeOfParallelism
              member self.Apply<'S, 'R> (collectorf : unit -> Collector<'Key * 'State, 'S>) (projection : 'S -> Cloud<'R>) combiner =
                  cloud {
                      let! result = shuffling
                      let! result' = reducer (ofArray result)
  
negative.code:                                      |> Seq.groupBy (fun keyValue -> Math.Abs(keyValue.Key.GetHashCode()) % totalWorkers)
                                       |> Seq.map (fun (key, keyValues) -> (key, keyValues |> Seq.map (fun keyValue -> (keyValue.Key, !keyValue.Value))))
                      partitions }
          }
          // Phase 1
          let shuffling = 
              cloud {
  