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

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // NullableFloatMinMaxAggregationOperator.cs
   9 //
  10 // <OWNER>Microsoft</OWNER>
  11 //
  12 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
  13
  14 using System.Collections.Generic;
  15 using System.Diagnostics.Contracts;
  16 using System.Threading;
  17
  18 namespace System.Linq.Parallel
  19 {
  20     /// <summary>
  21     /// An inlined min/max aggregation and its enumerator, for Nullable floats.
  22     ///
  23     /// Notes:
  24     ///     Note that normally float.NaN &lt; anything is false, as is anything &lt; NaN.  This would
  25     ///     lead to some strangeness in Min and Max, e.g. Min({ NaN, 5.0 } == NaN, yet
  26     ///     Min({ 5.0, NaN }) == 5.0!  We impose a total ordering so that NaN is smaller than
  27     ///     everything, including -infinity, which is consistent with Comparer_T.
  28     /// </summary>
  29     internal sealed class NullableFloatMinMaxAggregationOperator : InlinedAggregationOperator<float?, float?, float?>
  30     {
  31         private readonly int m_sign; // The sign (-1 for min, 1 for max).
  32
  33         //---------------------------------------------------------------------------------------
  34         // Constructs a new instance of a min/max associative operator.
  35         //
  36
  37         internal NullableFloatMinMaxAggregationOperator(IEnumerable<float?> child, int sign) : base(child)
  38         {
  39             Contract.Assert(sign == -1 || sign == 1, "invalid sign");
  40             m_sign = sign;
  41         }
  42
  43         //---------------------------------------------------------------------------------------
  44         // Executes the entire query tree, and aggregates the intermediate results into the
  45         // final result based on the binary operators and final reduction.
  46         //
  47         // Return Value:
  48         //     The single result of aggregation.
  49         //
  50
  51         protected override float? InternalAggregate(ref Exception singularExceptionToThrow)
  52         {
  53             // Because the final reduction is typically much cheaper than the intermediate
  54             // reductions over the individual partitions, and because each parallel partition
  55             // will do a lot of work to produce a single output element, we prefer to turn off
  56             // pipelining, and process the final reductions serially.
  57             using (IEnumerator<float?> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
  58             {
  59                 // Just return null right away for empty results.
  60                 if (!enumerator.MoveNext())
  61                 {
  62                     return null;
  63                 }
  64
  65                 float? best = enumerator.Current;
  66
  67                 // Based on the sign, do either a min or max reduction.
  68                 if (m_sign == -1)
  69                 {
  70                     while (enumerator.MoveNext())
  71                     {
  72                         float? current = enumerator.Current;
  73                         if (current == null) continue;
  74                         if (best == null || current < best || float.IsNaN(current.GetValueOrDefault()))
  75                         {
  76                             best = current;
  77                         }
  78                     }
  79                 }
  80                 else
  81                 {
  82                     while (enumerator.MoveNext())
  83                     {
  84                         float? current = enumerator.Current;
  85                         if (current == null) continue;
  86                         if (best == null || current > best || float.IsNaN(best.GetValueOrDefault()))
  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 NullableFloatMinMaxAggregationOperatorEnumerator<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 NullableFloatMinMaxAggregationOperatorEnumerator<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 NullableFloatMinMaxAggregationOperatorEnumerator(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
 153                             if (elem == null) continue;
 154                             if (currentElement == null || elem < currentElement || float.IsNaN(elem.GetValueOrDefault()))
 155                             {
 156                                 currentElement = elem;
 157                             }
 158                         }
 159                     }
 160                     else
 161                     {
 162                         float? elem = default(float?);
 163                         while (source.MoveNext(ref elem, ref keyUnused))
 164                         {
 165                             if ((i++ & CancellationState.POLL_INTERVAL) == 0)
 166                                 CancellationState.ThrowIfCanceled(m_cancellationToken);
 167
 168                             if (elem == null) continue;
 169                             if (currentElement == null || elem > currentElement || float.IsNaN(currentElement.GetValueOrDefault()))
 170                             {
 171                                 currentElement = elem;
 172                             }
 173                         }
 174                     }
 175
 176                     // The sum has been calculated. Now just return.
 177                     return true;
 178                 }
 179
 180                 return false;
 181             }
 182
 183             //---------------------------------------------------------------------------------------
 184             // Dispose of resources associated with the underlying enumerator.
 185             //
 186
 187             protected override void Dispose(bool disposing)
 188             {
 189                 Contract.Assert(m_source != null);
 190                 m_source.Dispose();
 191             }
 192         }
 193     }
 194 }