mcs/class/referencesource/System.Core/System/Linq/Parallel/QueryOperators/Unary/IndexedSelectQueryOperator.cs

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // IndexedSelectQueryOperator.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     /// A variant of the Select operator that supplies element index while performing the
  22     /// projection operation. This requires cooperation with partitioning and merging to
  23     /// guarantee ordering is preserved.
  24     ///
  25     /// </summary>
  26     /// <typeparam name="TInput"></typeparam>
  27     /// <typeparam name="TOutput"></typeparam>
  28     internal sealed class IndexedSelectQueryOperator<TInput, TOutput> : UnaryQueryOperator<TInput, TOutput>
  29     {
  30
  31         // Selector function. Used to project elements to a transformed view during execution.
  32         private readonly Func<TInput, int, TOutput> m_selector;
  33         private bool m_prematureMerge = false; // Whether to prematurely merge the input of this operator.
  34         private bool m_limitsParallelism = false; // Whether this operator limits parallelism
  35
  36         //---------------------------------------------------------------------------------------
  37         // Initializes a new select operator.
  38         //
  39         // Arguments:
  40         //    child         - the child operator or data source from which to pull data
  41         //    selector      - a delegate representing the selector function
  42         //
  43         // Assumptions:
  44         //    selector must be non null.
  45         //
  46
  47         internal IndexedSelectQueryOperator(IEnumerable<TInput> child,
  48                                             Func<TInput, int, TOutput> selector)
  49             :base(child)
  50         {
  51             Contract.Assert(child != null, "child data source cannot be null");
  52             Contract.Assert(selector != null, "need a selector function");
  53
  54             m_selector = selector;
  55
  56             // In an indexed Select, elements must be returned in the order in which
  57             // indices were assigned.
  58             m_outputOrdered = true;
  59
  60             InitOrdinalIndexState();
  61         }
  62
  63         private void InitOrdinalIndexState()
  64         {
  65             OrdinalIndexState childIndexState = Child.OrdinalIndexState;
  66             OrdinalIndexState indexState = childIndexState;
  67
  68             if (ExchangeUtilities.IsWorseThan(childIndexState, OrdinalIndexState.Correct))
  69             {
  70                 m_prematureMerge = true;
  71                 m_limitsParallelism = childIndexState != OrdinalIndexState.Shuffled;
  72                 indexState = OrdinalIndexState.Correct;
  73             }
  74
  75             Contract.Assert(!ExchangeUtilities.IsWorseThan(indexState, OrdinalIndexState.Correct));
  76
  77             SetOrdinalIndexState(indexState);
  78         }
  79
  80         //---------------------------------------------------------------------------------------
  81         // Just opens the current operator, including opening the child and wrapping it with
  82         // partitions as needed.
  83         //
  84
  85         internal override QueryResults<TOutput> Open(
  86             QuerySettings settings, bool preferStriping)
  87         {
  88             QueryResults<TInput> childQueryResults = Child.Open(settings, preferStriping);
  89             return IndexedSelectQueryOperatorResults.NewResults(childQueryResults, this, settings, preferStriping);
  90         }
  91
  92         internal override void  WrapPartitionedStream<TKey>(
  93             PartitionedStream<TInput, TKey> inputStream, IPartitionedStreamRecipient<TOutput> recipient, bool preferStriping, QuerySettings settings)
  94         {
  95             int partitionCount = inputStream.PartitionCount;
  96
  97             // If the index is not correct, we need to reindex.
  98             PartitionedStream<TInput, int> inputStreamInt;
  99             if (m_prematureMerge)
 100             {
 101                 ListQueryResults<TInput> listResults = QueryOperator<TInput>.ExecuteAndCollectResults(
 102                     inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings);
 103                 inputStreamInt = listResults.GetPartitionedStream();
 104             }
 105             else
 106             {
 107                 Contract.Assert(typeof(TKey) == typeof(int));
 108                 inputStreamInt = (PartitionedStream<TInput, int>)(object)inputStream;
 109             }
 110
 111             // Since the index is correct, the type of the index must be int
 112             PartitionedStream<TOutput, int> outputStream =
 113                 new PartitionedStream<TOutput, int>(partitionCount, Util.GetDefaultComparer<int>(), OrdinalIndexState);
 114
 115             for (int i = 0; i < partitionCount; i++)
 116             {
 117                 outputStream[i] = new IndexedSelectQueryOperatorEnumerator(inputStreamInt[i], m_selector);
 118             }
 119
 120             recipient.Receive(outputStream);
 121         }
 122
 123
 124         //---------------------------------------------------------------------------------------
 125         // Whether this operator performs a premature merge that would not be performed in
 126         // a similar sequential operation (i.e., in LINQ to Objects).
 127         //
 128
 129         internal override bool LimitsParallelism
 130         {
 131             get { return m_limitsParallelism; }
 132         }
 133
 134         //---------------------------------------------------------------------------------------
 135         // The enumerator type responsible for projecting elements as it is walked.
 136         //
 137
 138         class IndexedSelectQueryOperatorEnumerator : QueryOperatorEnumerator<TOutput, int>
 139         {
 140
 141             private readonly QueryOperatorEnumerator<TInput, int> m_source; // The data source to enumerate.
 142             private readonly Func<TInput, int, TOutput> m_selector;  // The actual select function.
 143
 144             //---------------------------------------------------------------------------------------
 145             // Instantiates a new select enumerator.
 146             //
 147
 148             internal IndexedSelectQueryOperatorEnumerator(QueryOperatorEnumerator<TInput, int> source, Func<TInput, int, TOutput> selector)
 149             {
 150                 Contract.Assert(source != null);
 151                 Contract.Assert(selector != null);
 152                 m_source = source;
 153                 m_selector = selector;
 154             }
 155
 156             //---------------------------------------------------------------------------------------
 157             // Straightforward IEnumerator<T> methods.
 158             //
 159
 160             internal override bool MoveNext(ref TOutput currentElement, ref int currentKey)
 161             {
 162                 // So long as the source has a next element, we have an element.
 163                 TInput element = default(TInput);
 164                 if (m_source.MoveNext(ref element, ref currentKey))
 165                 {
 166                     Contract.Assert(m_selector != null, "expected a compiled selection function");
 167                     currentElement = m_selector(element, currentKey);
 168                     return true;
 169                 }
 170
 171                 return false;
 172             }
 173
 174             protected override void Dispose(bool disposing)
 175             {
 176                 m_source.Dispose();
 177             }
 178
 179         }
 180
 181
 182         //---------------------------------------------------------------------------------------
 183         // Returns an enumerable that represents the query executing sequentially.
 184         //
 185
 186         internal override IEnumerable<TOutput> AsSequentialQuery(CancellationToken token)
 187         {
 188             return Child.AsSequentialQuery(token).Select(m_selector);
 189         }
 190
 191         //-----------------------------------------------------------------------------------
 192         // Query results for an indexed Select operator. The results are indexible if the child
 193         // results were indexible.
 194         //
 195
 196         class IndexedSelectQueryOperatorResults : UnaryQueryOperatorResults
 197         {
 198             IndexedSelectQueryOperator<TInput, TOutput> m_selectOp;  // Operator that generated the results
 199             int m_childCount; // The number of elements in child results
 200
 201             public static QueryResults<TOutput> NewResults(
 202                 QueryResults<TInput> childQueryResults, IndexedSelectQueryOperator<TInput, TOutput> op,
 203                 QuerySettings settings, bool preferStriping)
 204             {
 205                 if (childQueryResults.IsIndexible)
 206                 {
 207                     return new IndexedSelectQueryOperatorResults(
 208                         childQueryResults, op, settings, preferStriping);
 209                 }
 210                 else
 211                 {
 212                     return new UnaryQueryOperatorResults(
 213                         childQueryResults, op, settings, preferStriping);
 214                 }
 215             }
 216
 217             private IndexedSelectQueryOperatorResults(
 218                 QueryResults<TInput> childQueryResults, IndexedSelectQueryOperator<TInput, TOutput> op,
 219                 QuerySettings settings, bool preferStriping)
 220                 : base(childQueryResults, op, settings, preferStriping)
 221             {
 222                 m_selectOp = op;
 223                 Contract.Assert(m_childQueryResults.IsIndexible);
 224
 225                 m_childCount = m_childQueryResults.ElementsCount;
 226             }
 227
 228             internal override int ElementsCount
 229             {
 230                 get
 231                 {
 232                     Contract.Assert(m_childCount >= 0);
 233                     return m_childQueryResults.ElementsCount;
 234                 }
 235             }
 236
 237             internal override bool IsIndexible
 238             {
 239                 get { return true; }
 240             }
 241
 242             internal override TOutput GetElement(int index)
 243             {
 244                 return m_selectOp.m_selector(m_childQueryResults.GetElement(index), index);
 245             }
 246         }
 247     }
 248 }