mcs/class/referencesource/System.Core/System/Linq/Parallel/QueryOperators/Inlined/DoubleSumAggregationOperator.cs

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // DoubleSumAggregationOperator.cs
   9 //
  10 // <OWNER>Microsoft</OWNER>
  11 //
  12 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
  13
  14 using System.Collections.Generic;
  15 using System.Diagnostics.Contracts;
  16 using System.Threading;
  17 #if SILVERLIGHT
  18 using System.Core; // for System.Core.SR
  19 #endif
  20 namespace System.Linq.Parallel
  21 {
  22     /// <summary>
  23     /// An inlined sum aggregation and its enumerator, for doubles.
  24     /// </summary>
  25     internal sealed class DoubleSumAggregationOperator : InlinedAggregationOperator<double, double, double>
  26     {
  27
  28         //---------------------------------------------------------------------------------------
  29         // Constructs a new instance of a sum associative operator.
  30         //
  31
  32         internal DoubleSumAggregationOperator(IEnumerable<double> child) : base(child)
  33         {
  34         }
  35
  36         //---------------------------------------------------------------------------------------
  37         // Executes the entire query tree, and aggregates the intermediate results into the
  38         // final result based on the binary operators and final reduction.
  39         //
  40         // Return Value:
  41         //     The single result of aggregation.
  42         //
  43
  44         protected override double InternalAggregate(ref Exception singularExceptionToThrow)
  45         {
  46             // Because the final reduction is typically much cheaper than the intermediate
  47             // reductions over the individual partitions, and because each parallel partition
  48             // will do a lot of work to produce a single output element, we prefer to turn off
  49             // pipelining, and process the final reductions serially.
  50             using (IEnumerator<double> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
  51             {
  52                 // We just reduce the elements in each output partition.
  53                 double sum = 0.0;
  54                 while (enumerator.MoveNext())
  55                 {
  56                     checked
  57                     {
  58                         sum += enumerator.Current;
  59                     }
  60                 }
  61
  62                 return sum;
  63             }
  64         }
  65
  66         //---------------------------------------------------------------------------------------
  67         // Creates an enumerator that is used internally for the final aggregation step.
  68         //
  69
  70         protected override QueryOperatorEnumerator<double,int> CreateEnumerator<TKey>(
  71             int index, int count, QueryOperatorEnumerator<double, TKey> source, object sharedData, CancellationToken cancellationToken)
  72         {
  73             return new DoubleSumAggregationOperatorEnumerator<TKey>(source, index, cancellationToken);
  74         }
  75
  76         //---------------------------------------------------------------------------------------
  77         // This enumerator type encapsulates the intermediary aggregation over the underlying
  78         // (possibly partitioned) data source.
  79         //
  80
  81         private class DoubleSumAggregationOperatorEnumerator<TKey> : InlinedAggregationOperatorEnumerator<double>
  82         {
  83             private readonly QueryOperatorEnumerator<double, TKey> m_source; // The source data.
  84
  85             //---------------------------------------------------------------------------------------
  86             // Instantiates a new aggregation operator.
  87             //
  88
  89             internal DoubleSumAggregationOperatorEnumerator(QueryOperatorEnumerator<double, TKey> source, int partitionIndex,
  90                 CancellationToken cancellationToken) :
  91                 base(partitionIndex, cancellationToken)
  92             {
  93                 Contract.Assert(source != null);
  94                 m_source = source;
  95             }
  96
  97             //---------------------------------------------------------------------------------------
  98             // Tallies up the sum of the underlying data source, walking the entire thing the first
  99             // time MoveNext is called on this object.
 100             //
 101
 102             protected override bool MoveNextCore(ref double currentElement)
 103             {
 104                 double element = default(double);
 105                 TKey keyUnused = default(TKey);
 106
 107                 QueryOperatorEnumerator<double, TKey> source = m_source;
 108                 if (source.MoveNext(ref element, ref keyUnused))
 109                 {
 110                     // We just scroll through the enumerator and accumulate the sum.
 111                     double tempSum = 0.0;
 112                     int i = 0;
 113                     do
 114                     {
 115                         if ((i++ & CancellationState.POLL_INTERVAL) == 0)
 116                             CancellationState.ThrowIfCanceled(m_cancellationToken);
 117                         tempSum += element;
 118                     }
 119                     while (source.MoveNext(ref element, ref keyUnused));
 120
 121                     // The sum has been calculated. Now just return.
 122                     currentElement = tempSum;
 123                     return true;
 124                 }
 125
 126                 return false;
 127             }
 128
 129             //---------------------------------------------------------------------------------------
 130             // Dispose of resources associated with the underlying enumerator.
 131             //
 132
 133             protected override void Dispose(bool disposing)
 134             {
 135                 Contract.Assert(m_source != null);
 136                 m_source.Dispose();
 137             }
 138         }
 139     }
 140 }