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

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // ReverseQueryOperator.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     /// Reverse imposes ordinal order preservation. There are normally two phases to this
  22     /// operator's execution.  Each partition first builds a buffer containing all of its
  23     /// elements, and then proceeds to yielding the elements in reverse.  There is a
  24     /// 'barrier' (but not a blocking barrier) in between these two steps, at which point the largest index becomes
  25     /// known.  This is necessary so that when elements from the buffer are yielded, the
  26     /// CurrentIndex can be reported as the largest index minus the original index (thereby
  27     /// reversing the indices as well as the elements themselves).  If the largest index is
  28     /// known a priori, because we have an array for example, we can avoid the barrier in
  29     /// between the steps.
  30     /// </summary>
  31     /// <typeparam name="TSource"></typeparam>
  32     internal sealed class ReverseQueryOperator<TSource> : UnaryQueryOperator<TSource, TSource>
  33     {
  34
  35         //---------------------------------------------------------------------------------------
  36         // Initializes a new reverse operator.
  37         //
  38         // Arguments:
  39         //     child                - the child whose data we will reverse
  40         //
  41
  42         internal ReverseQueryOperator(IEnumerable<TSource> child)
  43             :base(child)
  44         {
  45             Contract.Assert(child != null, "child data source cannot be null");
  46
  47             if (Child.OrdinalIndexState == OrdinalIndexState.Indexible)
  48             {
  49                 SetOrdinalIndexState(OrdinalIndexState.Indexible);
  50             }
  51             else
  52             {
  53                 SetOrdinalIndexState(OrdinalIndexState.Shuffled);
  54             }
  55
  56         }
  57
  58         internal override void WrapPartitionedStream<TKey>(
  59             PartitionedStream<TSource, TKey> inputStream, IPartitionedStreamRecipient<TSource> recipient, bool preferStriping, QuerySettings settings)
  60         {
  61             Contract.Assert(Child.OrdinalIndexState != OrdinalIndexState.Indexible, "Don't take this code path if the child is indexible.");
  62
  63             int partitionCount = inputStream.PartitionCount;
  64             PartitionedStream<TSource, TKey> outputStream = new PartitionedStream<TSource, TKey>(
  65                 partitionCount, new ReverseComparer<TKey>(inputStream.KeyComparer), OrdinalIndexState.Shuffled);
  66
  67             for (int i = 0; i < partitionCount; i++)
  68             {
  69                 outputStream[i] = new ReverseQueryOperatorEnumerator<TKey>(inputStream[i], settings.CancellationState.MergedCancellationToken);
  70             }
  71             recipient.Receive(outputStream);
  72         }
  73
  74         //---------------------------------------------------------------------------------------
  75         // Just opens the current operator, including opening the child and wrapping it with
  76         // partitions as needed.
  77         //
  78
  79         internal override QueryResults<TSource> Open(QuerySettings settings, bool preferStriping)
  80         {
  81             QueryResults<TSource> childQueryResults = Child.Open(settings, false);
  82             return ReverseQueryOperatorResults.NewResults(childQueryResults, this, settings, preferStriping);
  83         }
  84
  85         //---------------------------------------------------------------------------------------
  86         // Returns an enumerable that represents the query executing sequentially.
  87         //
  88
  89         internal override IEnumerable<TSource> AsSequentialQuery(CancellationToken token)
  90         {
  91             IEnumerable<TSource> wrappedChild = CancellableEnumerable.Wrap(Child.AsSequentialQuery(token), token);
  92             return wrappedChild.Reverse();
  93         }
  94
  95         //---------------------------------------------------------------------------------------
  96         // Whether this operator performs a premature merge that would not be performed in
  97         // a similar sequential operation (i.e., in LINQ to Objects).
  98         //
  99
 100         internal override bool LimitsParallelism
 101         {
 102             get { return false; }
 103         }
 104
 105         //---------------------------------------------------------------------------------------
 106         // The enumerator type responsible for executing the reverse operation.
 107         //
 108
 109         class ReverseQueryOperatorEnumerator<TKey> : QueryOperatorEnumerator<TSource, TKey>
 110         {
 111
 112             private readonly QueryOperatorEnumerator<TSource, TKey> m_source; // The data source to reverse.
 113             private readonly CancellationToken m_cancellationToken;
 114             private List<Pair<TSource, TKey>> m_buffer; // Our buffer. [allocate in moveNext to avoid false-sharing]
 115             private Shared<int> m_bufferIndex; // Our current index within the buffer. [allocate in moveNext to avoid false-sharing]
 116
 117             //---------------------------------------------------------------------------------------
 118             // Instantiates a new select enumerator.
 119             //
 120
 121             internal ReverseQueryOperatorEnumerator(QueryOperatorEnumerator<TSource, TKey> source,
 122                 CancellationToken cancellationToken)
 123             {
 124                 Contract.Assert(source != null);
 125                 m_source = source;
 126                 m_cancellationToken = cancellationToken;
 127             }
 128
 129             //---------------------------------------------------------------------------------------
 130             // Straightforward IEnumerator<T> methods.
 131             //
 132
 133             internal override bool MoveNext(ref TSource currentElement, ref TKey currentKey)
 134             {
 135                 // If the buffer has not been created, we will generate it lazily on demand.
 136                 if (m_buffer == null)
 137                 {
 138                     m_bufferIndex = new Shared<int>(0);
 139                     // Buffer all of our data.
 140                     m_buffer = new List<Pair<TSource, TKey>>();
 141                     TSource current = default(TSource);
 142                     TKey key = default(TKey);
 143                     int i = 0;
 144                     while (m_source.MoveNext(ref current, ref key))
 145                     {
 146                         if ((i++ & CancellationState.POLL_INTERVAL) == 0)
 147                             CancellationState.ThrowIfCanceled(m_cancellationToken);
 148
 149                         m_buffer.Add(new Pair<TSource, TKey>(current, key));
 150                         m_bufferIndex.Value++;
 151                     }
 152                 }
 153
 154                 // Continue yielding elements from our buffer.
 155                 if (--m_bufferIndex.Value >= 0)
 156                 {
 157                     currentElement = m_buffer[m_bufferIndex.Value].First;
 158                     currentKey = m_buffer[m_bufferIndex.Value].Second;
 159                     return true;
 160                 }
 161
 162                 return false;
 163             }
 164
 165             protected override void Dispose(bool disposing)
 166             {
 167                 m_source.Dispose();
 168             }
 169         }
 170
 171         //-----------------------------------------------------------------------------------
 172         // Query results for a Reverse operator. The results are indexible if the child
 173         // results were indexible.
 174         //
 175
 176         class ReverseQueryOperatorResults : UnaryQueryOperatorResults
 177         {
 178             private int m_count; // The number of elements in child results
 179
 180             public static QueryResults<TSource> NewResults(
 181                 QueryResults<TSource> childQueryResults, ReverseQueryOperator<TSource> op,
 182                 QuerySettings settings, bool preferStriping)
 183             {
 184                 if (childQueryResults.IsIndexible)
 185                 {
 186                     return new ReverseQueryOperatorResults(
 187                         childQueryResults, op, settings, preferStriping);
 188                 }
 189                 else
 190                 {
 191                     return new UnaryQueryOperatorResults(
 192                         childQueryResults, op, settings, preferStriping);
 193                 }
 194             }
 195
 196             private ReverseQueryOperatorResults(
 197                 QueryResults<TSource> childQueryResults, ReverseQueryOperator<TSource> op,
 198                 QuerySettings settings, bool preferStriping)
 199                 : base(childQueryResults, op, settings, preferStriping)
 200             {
 201                 Contract.Assert(m_childQueryResults.IsIndexible);
 202                 m_count = m_childQueryResults.ElementsCount;
 203             }
 204
 205             internal override bool IsIndexible
 206             {
 207                 get { return true; }
 208             }
 209
 210             internal override int ElementsCount
 211             {
 212                 get
 213                 {
 214                     Contract.Assert(m_count >= 0);
 215                     return m_count;
 216                 }
 217             }
 218
 219             internal override TSource GetElement(int index)
 220             {
 221                 Contract.Assert(m_count >= 0);
 222                 Contract.Assert(index >= 0);
 223                 Contract.Assert(index < m_count);
 224
 225                 return m_childQueryResults.GetElement(m_count - index - 1);
 226             }
 227         }
 228
 229     }
 230 }