mcs/class/referencesource/mscorlib/system/threading/Tasks/ParallelRangeManager.cs

   1 // ==++==
   2 //
   3 //   Copyright (c) Microsoft Corporation.  All rights reserved.
   4 //
   5 // ==--==
   6 // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
   7 //
   8 // ParallelRangeManager.cs
   9 //
  10 // <OWNER>Microsoft</OWNER>
  11 //
  12 // Implements the algorithm for distributing loop indices to parallel loop workers
  13 //
  14 // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
  15
  16 using System;
  17 using System.Threading;
  18 using System.Diagnostics.Contracts;
  19
  20 #pragma warning disable 0420
  21
  22 namespace System.Threading.Tasks
  23 {
  24     /// <summary>
  25     /// Represents an index range
  26     /// </summary>
  27     internal struct IndexRange
  28     {
  29         // the From and To values for this range. These do not change.
  30         internal long m_nFromInclusive;
  31         internal long m_nToExclusive;
  32
  33         // The shared index, stored as the offset from nFromInclusive. Using an offset rather than the actual
  34         // value saves us from overflows that can happen due to multiple workers racing to increment this.
  35         // All updates to this field need to be interlocked.
  36         internal volatile Shared<long> m_nSharedCurrentIndexOffset;
  37
  38         // to be set to 1 by the worker that finishes this range. It's OK to do a non-interlocked write here.
  39         internal int m_bRangeFinished;
  40     }
  41
  42
  43     /// <summary>
  44     /// The RangeWorker struct wraps the state needed by a task that services the parallel loop
  45     /// </summary>
  46     internal struct RangeWorker
  47     {
  48         // reference to the IndexRange array allocated by the range manager
  49         internal readonly IndexRange[] m_indexRanges;
  50
  51         // index of the current index range that this worker is grabbing chunks from
  52         internal int m_nCurrentIndexRange;
  53
  54         // the step for this loop. Duplicated here for quick access (rather than jumping to rangemanager)
  55         internal long m_nStep;
  56
  57         // increment value is the current amount that this worker will use
  58         // to increment the shared index of the range it's working on
  59         internal long m_nIncrementValue;
  60
  61         // the increment value is doubled each time this worker finds work, and is capped at this value
  62         internal readonly long m_nMaxIncrementValue;
  63
  64         /// <summary>
  65         /// Initializes a RangeWorker struct
  66         /// </summary>
  67         internal RangeWorker(IndexRange[] ranges, int nInitialRange, long nStep)
  68         {
  69             m_indexRanges = ranges;
  70             m_nCurrentIndexRange = nInitialRange;
  71             m_nStep = nStep;
  72
  73             m_nIncrementValue = nStep;
  74
  75             m_nMaxIncrementValue = Parallel.DEFAULT_LOOP_STRIDE * nStep;
  76         }
  77
  78         /// <summary>
  79         /// Implements the core work search algorithm that will be used for this range worker.
  80         /// </summary>
  81         ///
  82         /// Usage pattern is:
  83         ///    1) the thread associated with this rangeworker calls FindNewWork
  84         ///    2) if we return true, the worker uses the nFromInclusiveLocal and nToExclusiveLocal values
  85         ///       to execute the sequential loop
  86         ///    3) if we return false it means there is no more work left. It's time to quit.
  87         ///
  88         internal bool FindNewWork(out long nFromInclusiveLocal, out long nToExclusiveLocal)
  89         {
  90             // since we iterate over index ranges circularly, we will use the
  91             // count of visited ranges as our exit condition
  92             int numIndexRangesToVisit = m_indexRanges.Length;
  93
  94             do
  95             {
  96                 // local snap to save array access bounds checks in places where we only read fields
  97                 IndexRange currentRange = m_indexRanges[m_nCurrentIndexRange];
  98
  99                 if (currentRange.m_bRangeFinished == 0)
 100                 {
 101                     if (m_indexRanges[m_nCurrentIndexRange].m_nSharedCurrentIndexOffset == null)
 102                     {
 103                         Interlocked.CompareExchange(ref m_indexRanges[m_nCurrentIndexRange].m_nSharedCurrentIndexOffset, new Shared<long>(0), null);
 104                     }
 105
 106                     // this access needs to be on the array slot
 107                     long nMyOffset = Interlocked.Add(ref m_indexRanges[m_nCurrentIndexRange].m_nSharedCurrentIndexOffset.Value,
 108                                                     m_nIncrementValue) - m_nIncrementValue;
 109
 110                     if (currentRange.m_nToExclusive - currentRange.m_nFromInclusive > nMyOffset)
 111                     {
 112                         // we found work
 113
 114                         nFromInclusiveLocal = currentRange.m_nFromInclusive + nMyOffset;
 115                         nToExclusiveLocal = nFromInclusiveLocal + m_nIncrementValue;
 116
 117                         // Check for going past end of range, or wrapping
 118                         if ( (nToExclusiveLocal > currentRange.m_nToExclusive) || (nToExclusiveLocal < currentRange.m_nFromInclusive) )
 119                         {
 120                             nToExclusiveLocal = currentRange.m_nToExclusive;
 121                         }
 122
 123                         // We will double our unit of increment until it reaches the maximum.
 124                         if (m_nIncrementValue < m_nMaxIncrementValue)
 125                         {
 126                             m_nIncrementValue *= 2;
 127                             if (m_nIncrementValue > m_nMaxIncrementValue)
 128                             {
 129                                 m_nIncrementValue = m_nMaxIncrementValue;
 130                             }
 131                         }
 132
 133                         return true;
 134                     }
 135                     else
 136                     {
 137                         // this index range is completed, mark it so that others can skip it quickly
 138                         Interlocked.Exchange(ref m_indexRanges[m_nCurrentIndexRange].m_bRangeFinished, 1);
 139                     }
 140                 }
 141
 142                 // move on to the next index range, in circular order.
 143                 m_nCurrentIndexRange = (m_nCurrentIndexRange + 1) % m_indexRanges.Length;
 144                 numIndexRangesToVisit--;
 145
 146             } while (numIndexRangesToVisit > 0);
 147             // we've visited all index ranges possible => there's no work remaining
 148
 149             nFromInclusiveLocal = 0;
 150             nToExclusiveLocal = 0;
 151
 152             return false;
 153         }
 154
 155
 156         /// <summary>
 157         /// 32 bit integer version of FindNewWork. Assumes the ranges were initialized with 32 bit values.
 158         /// </summary>
 159         internal bool FindNewWork32(out int nFromInclusiveLocal32, out int nToExclusiveLocal32)
 160         {
 161             long nFromInclusiveLocal;
 162             long nToExclusiveLocal;
 163
 164             bool bRetVal = FindNewWork(out nFromInclusiveLocal, out nToExclusiveLocal);
 165
 166             Contract.Assert((nFromInclusiveLocal <= Int32.MaxValue) && (nFromInclusiveLocal >= Int32.MinValue) &&
 167                             (nToExclusiveLocal <= Int32.MaxValue) && (nToExclusiveLocal >= Int32.MinValue));
 168
 169             // convert to 32 bit before returning
 170             nFromInclusiveLocal32 = (int)nFromInclusiveLocal;
 171             nToExclusiveLocal32 = (int)nToExclusiveLocal;
 172
 173             return bRetVal;
 174         }
 175     }
 176
 177
 178     /// <summary>
 179     /// Represents the entire loop operation, keeping track of workers and ranges.
 180     /// </summary>
 181     ///
 182     /// The usage pattern is:
 183     ///    1) The Parallel loop entry function (ForWorker) creates an instance of this class
 184     ///    2) Every thread joining to service the parallel loop calls RegisterWorker to grab a
 185     ///       RangeWorker struct to wrap the state it will need to find and execute work,
 186     ///       and they keep interacting with that struct until the end of the loop
 187     internal class RangeManager
 188     {
 189         internal readonly IndexRange[] m_indexRanges;
 190
 191         internal int m_nCurrentIndexRangeToAssign;
 192         internal long m_nStep;
 193
 194         /// <summary>
 195         /// Initializes a RangeManager with the given loop parameters, and the desired number of outer ranges
 196         /// </summary>
 197         internal RangeManager(long nFromInclusive, long nToExclusive, long nStep, int nNumExpectedWorkers)
 198         {
 199             m_nCurrentIndexRangeToAssign = 0;
 200             m_nStep = nStep;
 201
 202             // Our signed math breaks down w/ nNumExpectedWorkers == 1.  So change it to 2.
 203             if (nNumExpectedWorkers == 1)
 204                 nNumExpectedWorkers = 2;
 205
 206             //
 207             // calculate the size of each index range
 208             //
 209
 210             ulong uSpan = (ulong)(nToExclusive - nFromInclusive);
 211             ulong uRangeSize = uSpan / (ulong) nNumExpectedWorkers; // rough estimate first
 212
 213             uRangeSize -= uRangeSize % (ulong) nStep; // snap to multiples of nStep
 214                                                       // otherwise index range transitions will derail us from nStep
 215
 216             if (uRangeSize == 0)
 217             {
 218                 uRangeSize = (ulong) nStep;
 219             }
 220
 221             //
 222             // find the actual number of index ranges we will need
 223             //
 224             Contract.Assert((uSpan / uRangeSize) < Int32.MaxValue);
 225
 226             int nNumRanges = (int)(uSpan / uRangeSize);
 227
 228             if (uSpan % uRangeSize != 0)
 229             {
 230                 nNumRanges++;
 231             }
 232
 233
 234             // Convert to signed so the rest of the logic works.
 235             // Should be fine so long as uRangeSize < Int64.MaxValue, which we guaranteed by setting #workers >= 2.
 236             long nRangeSize = (long)uRangeSize;
 237
 238             // allocate the array of index ranges
 239             m_indexRanges = new IndexRange[nNumRanges];
 240
 241             long nCurrentIndex = nFromInclusive;
 242             for (int i = 0; i < nNumRanges; i++)
 243             {
 244                 // the fromInclusive of the new index range is always on nCurrentIndex
 245                 m_indexRanges[i].m_nFromInclusive = nCurrentIndex;
 246                 m_indexRanges[i].m_nSharedCurrentIndexOffset = null;
 247                 m_indexRanges[i].m_bRangeFinished = 0;
 248
 249                 // now increment it to find the toExclusive value for our range
 250                 nCurrentIndex += nRangeSize;
 251
 252                 // detect integer overflow or range overage and snap to nToExclusive
 253                 if (nCurrentIndex < nCurrentIndex - nRangeSize ||
 254                     nCurrentIndex > nToExclusive)
 255                 {
 256                     // this should only happen at the last index
 257                     Contract.Assert(i == nNumRanges - 1);
 258
 259                     nCurrentIndex = nToExclusive;
 260                 }
 261
 262                 // now that the end point of the new range is calculated, assign it.
 263                 m_indexRanges[i].m_nToExclusive = nCurrentIndex;
 264             }
 265         }
 266
 267         /// <summary>
 268         /// The function that needs to be called by each new worker thread servicing the parallel loop
 269         /// in order to get a RangeWorker struct that wraps the state for finding and executing indices
 270         /// </summary>
 271         internal RangeWorker RegisterNewWorker()
 272         {
 273             Contract.Assert(m_indexRanges != null && m_indexRanges.Length != 0);
 274
 275             int nInitialRange = (Interlocked.Increment(ref m_nCurrentIndexRangeToAssign) - 1) % m_indexRanges.Length;
 276
 277             return new RangeWorker(m_indexRanges, nInitialRange, m_nStep);
 278         }
 279     }
 280 }
 281 #pragma warning restore 0420