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

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // SortQueryOperator.cs
   9 //
  10 // <OWNER>[....]</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     /// The query operator for OrderBy and ThenBy.
  22     /// </summary>
  23     /// <typeparam name="TInputOutput"></typeparam>
  24     /// <typeparam name="TSortKey"></typeparam>
  25     internal sealed class SortQueryOperator<TInputOutput, TSortKey> :
  26         UnaryQueryOperator<TInputOutput, TInputOutput>, IOrderedEnumerable<TInputOutput>
  27     {
  28         private readonly Func<TInputOutput, TSortKey> m_keySelector; // Key selector used when sorting.
  29         private readonly IComparer<TSortKey> m_comparer; // Key comparison logic to use during sorting.
  30
  31         //---------------------------------------------------------------------------------------
  32         // Instantiates a new sort operator.
  33         //
  34
  35         internal SortQueryOperator(IEnumerable<TInputOutput> source, Func<TInputOutput, TSortKey> keySelector,
  36                                    IComparer<TSortKey> comparer, bool descending)
  37             : base(source, true)
  38         {
  39             Contract.Assert(keySelector != null, "key selector must not be null");
  40
  41             m_keySelector = keySelector;
  42
  43             // If a comparer wasn't supplied, we use the default one for the key type.
  44             if (comparer == null)
  45             {
  46                 m_comparer = Util.GetDefaultComparer<TSortKey>();
  47             }
  48             else
  49             {
  50                 m_comparer = comparer;
  51             }
  52
  53             if (descending)
  54             {
  55                 m_comparer = new ReverseComparer<TSortKey>(m_comparer);
  56             }
  57
  58             SetOrdinalIndexState(OrdinalIndexState.Shuffled);
  59         }
  60
  61         //---------------------------------------------------------------------------------------
  62         // IOrderedEnumerable method for nesting an order by operator inside another.
  63         //
  64
  65         IOrderedEnumerable<TInputOutput> IOrderedEnumerable<TInputOutput>.CreateOrderedEnumerable<TKey2>(
  66             Func<TInputOutput, TKey2> key2Selector, IComparer<TKey2> key2Comparer, bool descending)
  67         {
  68             key2Comparer = key2Comparer ?? Util.GetDefaultComparer<TKey2>();
  69
  70             if (descending)
  71             {
  72                 key2Comparer = new ReverseComparer<TKey2>(key2Comparer);
  73             }
  74
  75             IComparer<Pair<TSortKey, TKey2>> pairComparer = new PairComparer<TSortKey, TKey2>(m_comparer, key2Comparer);
  76             Func<TInputOutput, Pair<TSortKey, TKey2>> pairKeySelector =
  77                 (TInputOutput elem) => new Pair<TSortKey, TKey2>(m_keySelector(elem), key2Selector(elem));
  78
  79             return new SortQueryOperator<TInputOutput, Pair<TSortKey, TKey2>>(Child, pairKeySelector, pairComparer, false);
  80         }
  81
  82         //---------------------------------------------------------------------------------------
  83         // Accessor the the key selector.
  84         //
  85
  86         internal Func<TInputOutput, TSortKey> KeySelector
  87         {
  88             get { return m_keySelector; }
  89         }
  90
  91         //---------------------------------------------------------------------------------------
  92         // Accessor the the key comparer.
  93         //
  94
  95         internal IComparer<TSortKey> KeyComparer
  96         {
  97             get { return m_comparer; }
  98         }
  99
 100         //---------------------------------------------------------------------------------------
 101         // Opens the current operator. This involves opening the child operator tree, enumerating
 102         // the results, sorting them, and then returning an enumerator that walks the result.
 103         //
 104
 105         internal override QueryResults<TInputOutput> Open(QuerySettings settings, bool preferStriping)
 106         {
 107             QueryResults<TInputOutput> childQueryResults = Child.Open(settings, false);
 108             return new SortQueryOperatorResults<TInputOutput, TSortKey>(childQueryResults, this, settings, preferStriping);
 109         }
 110
 111
 112         internal override void WrapPartitionedStream<TKey>(
 113             PartitionedStream<TInputOutput, TKey> inputStream, IPartitionedStreamRecipient<TInputOutput> recipient, bool preferStriping, QuerySettings settings)
 114         {
 115             PartitionedStream<TInputOutput, TSortKey> outputStream =
 116                 new PartitionedStream<TInputOutput, TSortKey>(inputStream.PartitionCount, this.m_comparer, OrdinalIndexState);
 117
 118             for (int i = 0; i < outputStream.PartitionCount; i++)
 119             {
 120                 outputStream[i] = new SortQueryOperatorEnumerator<TInputOutput, TKey, TSortKey>(
 121                     inputStream[i], m_keySelector, m_comparer);
 122             }
 123
 124             recipient.Receive<TSortKey>(outputStream);
 125         }
 126
 127         //---------------------------------------------------------------------------------------
 128         // Returns an enumerable that represents the query executing sequentially.
 129         //
 130
 131         internal override IEnumerable<TInputOutput> AsSequentialQuery(CancellationToken token)
 132         {
 133             IEnumerable<TInputOutput> wrappedChild = CancellableEnumerable.Wrap(Child.AsSequentialQuery(token), token);
 134             return wrappedChild.OrderBy(m_keySelector, m_comparer);
 135         }
 136
 137         //---------------------------------------------------------------------------------------
 138         // Whether this operator performs a premature merge that would not be performed in
 139         // a similar sequential operation (i.e., in LINQ to Objects).
 140         //
 141
 142         internal override bool LimitsParallelism
 143         {
 144             get { return false; }
 145         }
 146     }
 147
 148     internal class SortQueryOperatorResults<TInputOutput, TSortKey> : QueryResults<TInputOutput>
 149     {
 150         protected QueryResults<TInputOutput> m_childQueryResults; // Results of the child query
 151         private SortQueryOperator<TInputOutput, TSortKey> m_op; // Operator that generated these results
 152         private QuerySettings m_settings; // Settings collected from the query
 153 #if !MONO
 154         private bool m_preferStriping; // If the results are indexible, should we use striping when partitioning them
 155 #endif
 156
 157         internal SortQueryOperatorResults(
 158             QueryResults<TInputOutput> childQueryResults, SortQueryOperator<TInputOutput, TSortKey> op,
 159             QuerySettings settings, bool preferStriping)
 160         {
 161             m_childQueryResults = childQueryResults;
 162             m_op = op;
 163             m_settings = settings;
 164 #if !MONO
 165             m_preferStriping = preferStriping;
 166 #endif
 167         }
 168
 169         internal override bool IsIndexible
 170         {
 171             get { return false; }
 172         }
 173
 174         internal override void GivePartitionedStream(IPartitionedStreamRecipient<TInputOutput> recipient)
 175         {
 176             m_childQueryResults.GivePartitionedStream(new ChildResultsRecipient(recipient, m_op, m_settings));
 177         }
 178
 179         private class ChildResultsRecipient : IPartitionedStreamRecipient<TInputOutput>
 180         {
 181             IPartitionedStreamRecipient<TInputOutput> m_outputRecipient;
 182             SortQueryOperator<TInputOutput, TSortKey> m_op;
 183             QuerySettings m_settings;
 184
 185             internal ChildResultsRecipient(IPartitionedStreamRecipient<TInputOutput> outputRecipient, SortQueryOperator<TInputOutput, TSortKey> op, QuerySettings settings)
 186             {
 187                 m_outputRecipient = outputRecipient;
 188                 m_op = op;
 189                 m_settings = settings;
 190             }
 191
 192             public void Receive<TKey>(PartitionedStream<TInputOutput, TKey> childPartitionedStream)
 193             {
 194                 m_op.WrapPartitionedStream(childPartitionedStream, m_outputRecipient, false, m_settings);
 195             }
 196         }
 197     }
 198
 199     //---------------------------------------------------------------------------------------
 200     // This enumerator performs sorting based on a key selection and comparison routine.
 201     //
 202
 203     internal class SortQueryOperatorEnumerator<TInputOutput, TKey, TSortKey> : QueryOperatorEnumerator<TInputOutput, TSortKey>
 204     {
 205         private readonly QueryOperatorEnumerator<TInputOutput, TKey> m_source; // Data source to sort.
 206         private readonly Func<TInputOutput, TSortKey> m_keySelector; // Key selector used when sorting.
 207         private readonly IComparer<TSortKey> m_keyComparer; // Key comparison logic to use during sorting.
 208
 209         //---------------------------------------------------------------------------------------
 210         // Instantiates a new sort operator enumerator.
 211         //
 212
 213         internal SortQueryOperatorEnumerator(QueryOperatorEnumerator<TInputOutput, TKey> source,
 214             Func<TInputOutput, TSortKey> keySelector, IComparer<TSortKey> keyComparer)
 215         {
 216             Contract.Assert(source != null);
 217             Contract.Assert(keySelector != null, "need a key comparer");
 218             Contract.Assert(keyComparer != null, "expected a compiled operator");
 219
 220             m_source = source;
 221             m_keySelector = keySelector;
 222             m_keyComparer = keyComparer;
 223         }
 224         //---------------------------------------------------------------------------------------
 225         // Accessor for the key comparison routine.
 226         //
 227
 228         public IComparer<TSortKey> KeyComparer
 229         {
 230             get { return m_keyComparer; }
 231         }
 232
 233         //---------------------------------------------------------------------------------------
 234         // Moves to the next element in the sorted output. When called for the first time, the
 235         // descendents in the sort's child tree are executed entirely, the results accumulated
 236         // in memory, and the data sorted.
 237         //
 238
 239         internal override bool MoveNext(ref TInputOutput currentElement, ref TSortKey currentKey)
 240         {
 241             Contract.Assert(m_source != null);
 242
 243             TKey keyUnused = default(TKey);
 244             if (!m_source.MoveNext(ref currentElement, ref keyUnused))
 245             {
 246                 return false;
 247             }
 248
 249             currentKey = m_keySelector(currentElement);
 250             return true;
 251         }
 252
 253         protected override void Dispose(bool disposing)
 254         {
 255             Contract.Assert(m_source != null);
 256             m_source.Dispose();
 257         }
 258     }
 259 }