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

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // LongMinMaxAggregationOperator.cs
   9 //
  10 // <OWNER>[....]</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
  21 namespace System.Linq.Parallel
  22 {
  23     /// <summary>
  24     /// An inlined min/max aggregation and its enumerator, for longs.
  25     /// </summary>
  26     internal sealed class LongMinMaxAggregationOperator : InlinedAggregationOperator<long, long, long>
  27     {
  28         private readonly int m_sign; // The sign (-1 for min, 1 for max).
  29
  30         //---------------------------------------------------------------------------------------
  31         // Constructs a new instance of a min/max associative operator.
  32         //
  33
  34         internal LongMinMaxAggregationOperator(IEnumerable<long> child, int sign) : base(child)
  35         {
  36             Contract.Assert(sign == -1 || sign == 1, "invalid sign");
  37             m_sign = sign;
  38         }
  39
  40         //---------------------------------------------------------------------------------------
  41         // Executes the entire query tree, and aggregates the intermediate results into the
  42         // final result based on the binary operators and final reduction.
  43         //
  44         // Return Value:
  45         //     The single result of aggregation.
  46         //
  47
  48         protected override long InternalAggregate(ref Exception singularExceptionToThrow)
  49         {
  50             // Because the final reduction is typically much cheaper than the intermediate
  51             // reductions over the individual partitions, and because each parallel partition
  52             // will do a lot of work to produce a single output element, we prefer to turn off
  53             // pipelining, and process the final reductions serially.
  54             using (IEnumerator<long> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
  55             {
  56                 // Throw an error for empty results.
  57                 if (!enumerator.MoveNext())
  58                 {
  59                     singularExceptionToThrow = new InvalidOperationException(SR.GetString(SR.NoElements));
  60                     return default(long);
  61                 }
  62
  63                 long best = enumerator.Current;
  64
  65                 // Based on the sign, do either a min or max reduction.
  66                 if (m_sign == -1)
  67                 {
  68                     while (enumerator.MoveNext())
  69                     {
  70                         long current = enumerator.Current;
  71                         if (current < best)
  72                         {
  73                             best = current;
  74                         }
  75                     }
  76                 }
  77                 else
  78                 {
  79                     while (enumerator.MoveNext())
  80                     {
  81                         long current = enumerator.Current;
  82                         if (current > best)
  83                         {
  84                             best = current;
  85                         }
  86                     }
  87                 }
  88
  89                 return best;
  90             }
  91         }
  92
  93         //---------------------------------------------------------------------------------------
  94         // Creates an enumerator that is used internally for the final aggregation step.
  95         //
  96
  97         protected override QueryOperatorEnumerator<long, int> CreateEnumerator<TKey>(
  98             int index, int count, QueryOperatorEnumerator<long, TKey> source, object sharedData,
  99             CancellationToken cancellationToken)
 100         {
 101             return new LongMinMaxAggregationOperatorEnumerator<TKey>(source, index, m_sign, cancellationToken);
 102         }
 103
 104         //---------------------------------------------------------------------------------------
 105         // This enumerator type encapsulates the intermediary aggregation over the underlying
 106         // (possibly partitioned) data source.
 107         //
 108
 109         private class LongMinMaxAggregationOperatorEnumerator<TKey> : InlinedAggregationOperatorEnumerator<long>
 110         {
 111             private QueryOperatorEnumerator<long, TKey> m_source; // The source data.
 112             private int m_sign; // The sign for comparisons (-1 means min, 1 means max).
 113
 114             //---------------------------------------------------------------------------------------
 115             // Instantiates a new aggregation operator.
 116             //
 117
 118             internal LongMinMaxAggregationOperatorEnumerator(QueryOperatorEnumerator<long, TKey> source, int partitionIndex, int sign,
 119                 CancellationToken cancellationToken) :
 120                 base(partitionIndex, cancellationToken)
 121             {
 122                 Contract.Assert(source != null);
 123                 m_source = source;
 124                 m_sign = sign;
 125             }
 126
 127             //---------------------------------------------------------------------------------------
 128             // Tallies up the min/max of the underlying data source, walking the entire thing the first
 129             // time MoveNext is called on this object.
 130             //
 131
 132             protected override bool MoveNextCore(ref long currentElement)
 133             {
 134                 // Based on the sign, do either a min or max reduction.
 135                 QueryOperatorEnumerator<long, TKey> source = m_source;
 136                 TKey keyUnused = default(TKey);
 137
 138                 if (source.MoveNext(ref currentElement, ref keyUnused))
 139                 {
 140                     int i = 0;
 141                     // We just scroll through the enumerator and find the min or max.
 142                     if (m_sign == -1)
 143                     {
 144                         long elem = default(long);
 145                         while (source.MoveNext(ref elem, ref keyUnused))
 146                         {
 147                             if ((i++ & CancellationState.POLL_INTERVAL) == 0)
 148                                 CancellationState.ThrowIfCanceled(m_cancellationToken);
 149
 150                             if (elem < currentElement)
 151                             {
 152                                 currentElement = elem;
 153                             }
 154                         }
 155                     }
 156                     else
 157                     {
 158                         long elem = default(long);
 159                         while (source.MoveNext(ref elem, ref keyUnused))
 160                         {
 161                             if ((i++ & CancellationState.POLL_INTERVAL) == 0)
 162                                 CancellationState.ThrowIfCanceled(m_cancellationToken);
 163
 164                             if (elem > currentElement)
 165                             {
 166                                 currentElement = elem;
 167                             }
 168                         }
 169                     }
 170
 171                     // The sum has been calculated. Now just return.
 172                     return true;
 173                 }
 174
 175                 return false;
 176             }
 177
 178             //---------------------------------------------------------------------------------------
 179             // Dispose of resources associated with the underlying enumerator.
 180             //
 181
 182             protected override void Dispose(bool disposing)
 183             {
 184                 Contract.Assert(m_source != null);
 185                 m_source.Dispose();
 186             }
 187         }
 188     }
 189 }