mcs/class/referencesource/System.Core/System/Linq/Parallel/QueryOperators/PartitionedStreamMerger.cs

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // PartitionedStreamMerger.cs
   9 //
  10 // <OWNER>Microsoft</OWNER>
  11 //
  12 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
  13
  14 using System.Threading.Tasks;
  15 using System.Diagnostics.Contracts;
  16
  17 namespace System.Linq.Parallel
  18 {
  19     /// <summary>
  20     /// Partitioned stream recipient that will merge the results.
  21     /// </summary>
  22     internal class PartitionedStreamMerger<TOutput> : IPartitionedStreamRecipient<TOutput>
  23     {
  24         private bool m_forEffectMerge;
  25         private ParallelMergeOptions m_mergeOptions;
  26         private bool m_isOrdered;
  27         private MergeExecutor<TOutput> m_mergeExecutor = null;
  28         private TaskScheduler m_taskScheduler;
  29         private int m_queryId; // ID of the current query execution
  30
  31         private CancellationState m_cancellationState;
  32
  33 #if DEBUG
  34             private bool m_received = false;
  35 #endif
  36         // Returns the merge executor which merges the received partitioned stream.
  37         internal MergeExecutor<TOutput> MergeExecutor
  38         {
  39             get
  40             {
  41 #if DEBUG
  42                 Contract.Assert(m_received, "Cannot return the merge executor because Receive() has not been called yet.");
  43 #endif
  44                 return m_mergeExecutor;
  45             }
  46         }
  47
  48         internal PartitionedStreamMerger(bool forEffectMerge, ParallelMergeOptions mergeOptions, TaskScheduler taskScheduler, bool outputOrdered,
  49             CancellationState cancellationState, int queryId)
  50         {
  51             m_forEffectMerge = forEffectMerge;
  52             m_mergeOptions = mergeOptions;
  53             m_isOrdered = outputOrdered;
  54             m_taskScheduler = taskScheduler;
  55             m_cancellationState = cancellationState;
  56             m_queryId = queryId;
  57         }
  58
  59         public void Receive<TKey>(PartitionedStream<TOutput, TKey> partitionedStream)
  60         {
  61 #if DEBUG
  62                 m_received = true;
  63 #endif
  64             m_mergeExecutor = MergeExecutor<TOutput>.Execute<TKey>(
  65                 partitionedStream, m_forEffectMerge, m_mergeOptions, m_taskScheduler, m_isOrdered, m_cancellationState, m_queryId);
  66
  67             TraceHelpers.TraceInfo("[timing]: {0}: finished opening - QueryOperator<>::GetEnumerator", DateTime.Now.Ticks);
  68         }
  69     }
  70 }