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

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