mcs/class/referencesource/System.Core/System/Linq/Parallel/Partitioning/OrderedHashRepartitionStream.cs

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // OrderedHashRepartitionStream.cs
   9 //
  10 // <OWNER>Microsoft</OWNER>
  11 //
  12 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
  13
  14 using System.Collections.Generic;
  15 using System.Threading;
  16
  17 namespace System.Linq.Parallel
  18 {
  19     internal class OrderedHashRepartitionStream<TInputOutput, THashKey, TOrderKey> : HashRepartitionStream<TInputOutput, THashKey, TOrderKey>
  20     {
  21         internal OrderedHashRepartitionStream(
  22             PartitionedStream<TInputOutput, TOrderKey> inputStream, Func<TInputOutput, THashKey> hashKeySelector,
  23             IEqualityComparer<THashKey> hashKeyComparer, IEqualityComparer<TInputOutput> elementComparer, CancellationToken cancellationToken)
  24             : base(inputStream.PartitionCount, inputStream.KeyComparer, hashKeyComparer, elementComparer)
  25         {
  26             m_partitions =
  27                 new OrderedHashRepartitionEnumerator<TInputOutput, THashKey, TOrderKey>[inputStream.PartitionCount];
  28
  29             // Initialize state shared among the partitions. A latch and a matrix of buffers. Note that
  30             // the actual elements in the buffer array are lazily allocated if needed.
  31             CountdownEvent barrier = new CountdownEvent(inputStream.PartitionCount);
  32             ListChunk<Pair<TInputOutput, THashKey>>[,] valueExchangeMatrix =
  33                 new ListChunk<Pair<TInputOutput, THashKey>>[inputStream.PartitionCount, inputStream.PartitionCount];
  34             ListChunk<TOrderKey>[,] keyExchangeMatrix = new ListChunk<TOrderKey>[inputStream.PartitionCount, inputStream.PartitionCount];
  35
  36             // Now construct each partition object.
  37             for (int i = 0; i < inputStream.PartitionCount; i++)
  38             {
  39                 m_partitions[i] = new OrderedHashRepartitionEnumerator<TInputOutput, THashKey, TOrderKey>(
  40                     inputStream[i], inputStream.PartitionCount, i, hashKeySelector, this, barrier,
  41                     valueExchangeMatrix, keyExchangeMatrix, cancellationToken);
  42             }
  43         }
  44     }
  45 }