code.grnet.gr Git - pithos-ms-client/blob - trunk/Libraries/ParallelExtensionsExtras/ParallelAlgorithms/ParallelAlgorithms_Reduce.cs

   1 //--------------------------------------------------------------------------
   2 //
   3 //  Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 //  File: ParallelAlgorithms_Reduce.cs
   6 //
   7 //--------------------------------------------------------------------------
   8
   9 using System.Collections.Generic;
  10 using System.Threading.Tasks;
  11
  12 namespace System.Threading.Algorithms
  13 {
  14     public static partial class ParallelAlgorithms
  15     {
  16         /// <summary>Reduces the input data using the specified aggregation operation.</summary>
  17         /// <typeparam name="T">Specifies the type of data being aggregated.</typeparam>
  18         /// <param name="input">The input data to be reduced.</param>
  19         /// <param name="seed">The seed to use to initialize the operation; this seed may be used multiple times.</param>
  20         /// <param name="associativeCommutativeOperation">The reduction operation.</param>
  21         /// <returns>The reduced value.</returns>
  22         public static T Reduce<T>(
  23             IList<T> input, T seed,
  24             Func<T, T, T> associativeCommutativeOperation)
  25         {
  26             return Reduce(input, s_defaultParallelOptions, seed, associativeCommutativeOperation);
  27         }
  28
  29         /// <summary>Reduces the input data using the specified aggregation operation.</summary>
  30         /// <typeparam name="T">Specifies the type of data being aggregated.</typeparam>
  31         /// <param name="input">The input data to be reduced.</param>
  32         /// <param name="parallelOptions">A ParallelOptions instance that configures the behavior of this operation.</param>
  33         /// <param name="seed">The seed to use to initialize the operation; this seed may be used multiple times.</param>
  34         /// <param name="associativeCommutativeOperation">The reduction operation.</param>
  35         /// <returns>The reduced value.</returns>
  36         public static T Reduce<T>(
  37             IList<T> input, ParallelOptions parallelOptions,
  38             T seed, Func<T, T, T> associativeCommutativeOperation)
  39         {
  40             if (input == null) throw new ArgumentNullException("input");
  41             return Reduce(0, input.Count, parallelOptions, i => input[i], seed, associativeCommutativeOperation);
  42         }
  43
  44         /// <summary>Reduces the input range using the specified aggregation operation.</summary>
  45         /// <typeparam name="T">Specifies the type of data being aggregated.</typeparam>
  46         /// <param name="fromInclusive">The start index, inclusive.</param>
  47         /// <param name="toExclusive">The end index, exclusive.</param>
  48         /// <param name="mapOperation">The function used to retrieve the data to be reduced for a given index.</param>
  49         /// <param name="seed">The seed to use to initialize the operation; this seed may be used multiple times.</param>
  50         /// <param name="associativeCommutativeOperation">The reduction operation.</param>
  51         /// <returns>The reduced value.</returns>
  52         public static T Reduce<T>(
  53             int fromInclusive, int toExclusive,
  54             Func<int, T> mapOperation, T seed, Func<T, T, T> associativeCommutativeOperation)
  55         {
  56             return Reduce(fromInclusive, toExclusive, s_defaultParallelOptions, mapOperation, seed, associativeCommutativeOperation);
  57         }
  58
  59         /// <summary>Reduces the input range using the specified aggregation operation.</summary>
  60         /// <typeparam name="T">Specifies the type of data being aggregated.</typeparam>
  61         /// <param name="fromInclusive">The start index, inclusive.</param>
  62         /// <param name="toExclusive">The end index, exclusive.</param>
  63         /// <param name="parallelOptions">A ParallelOptions instance that configures the behavior of this operation.</param>
  64         /// <param name="mapOperation">The function used to retrieve the data to be reduced for a given index.</param>
  65         /// <param name="seed">The seed to use to initialize the operation; this seed may be used multiple times.</param>
  66         /// <param name="associativeCommutativeOperation">The reduction operation.</param>
  67         /// <returns>The reduced value.</returns>
  68         public static T Reduce<T>(
  69             int fromInclusive, int toExclusive, ParallelOptions parallelOptions,
  70             Func<int, T> mapOperation, T seed, Func<T, T, T> associativeCommutativeOperation)
  71         {
  72             if (parallelOptions == null) throw new ArgumentNullException("parallelOptions");
  73             if (mapOperation == null) throw new ArgumentNullException("mapOperation");
  74             if (associativeCommutativeOperation == null) throw new ArgumentNullException("associativeCommutativeOperation");
  75             if (toExclusive < fromInclusive) throw new ArgumentOutOfRangeException("toExclusive");
  76
  77             object obj = new object(); // used as a monitor for the final reduction
  78             T result = seed; // accumulator for final reduction
  79
  80             // Reduce in parallel
  81             Parallel.For(fromInclusive, toExclusive, parallelOptions,
  82                 // Initialize each thread with the user-specified seed
  83                 () => seed,
  84                 // Map the current index to a value and aggregate that value into the local reduction
  85                 (i, loop, localResult) => associativeCommutativeOperation(mapOperation(i), localResult),
  86                 // Combine all of the local reductions
  87                 localResult => { lock (obj) result = associativeCommutativeOperation(localResult, result); });
  88
  89             // Return the final result
  90             return result;
  91         }
  92     }
  93 }