diff --git a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/Parallel.cs b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/Parallel.cs index 8d00d903af3da0..c1ec052557c3e6 100644 --- a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/Parallel.cs +++ b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/Parallel.cs @@ -13,6 +13,7 @@ using System.Collections.Concurrent; using System.Runtime.ExceptionServices; using System.Diagnostics; +using System.Numerics; namespace System.Threading.Tasks { @@ -384,7 +385,7 @@ public static ParallelLoopResult For(int fromInclusive, int toExclusive, Action< { ArgumentNullException.ThrowIfNull(body); - return ForWorker( + return ForWorker( fromInclusive, toExclusive, s_defaultParallelOptions, body, null, null, null, null); @@ -410,7 +411,7 @@ public static ParallelLoopResult For(long fromInclusive, long toExclusive, Actio { ArgumentNullException.ThrowIfNull(body); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, s_defaultParallelOptions, body, null, null, null, null); } @@ -447,7 +448,7 @@ public static ParallelLoopResult For(int fromInclusive, int toExclusive, Paralle ArgumentNullException.ThrowIfNull(parallelOptions); ArgumentNullException.ThrowIfNull(body); - return ForWorker( + return ForWorker( fromInclusive, toExclusive, parallelOptions, body, null, null, null, null); } @@ -484,7 +485,7 @@ public static ParallelLoopResult For(long fromInclusive, long toExclusive, Paral ArgumentNullException.ThrowIfNull(parallelOptions); ArgumentNullException.ThrowIfNull(body); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, parallelOptions, body, null, null, null, null); } @@ -533,7 +534,7 @@ public static ParallelLoopResult For(int fromInclusive, int toExclusive, Action< { ArgumentNullException.ThrowIfNull(body); - return ForWorker( + return ForWorker( fromInclusive, toExclusive, s_defaultParallelOptions, null, body, null, null, null); } @@ -560,7 +561,7 @@ public static ParallelLoopResult For(long fromInclusive, long toExclusive, Actio { ArgumentNullException.ThrowIfNull(body); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, s_defaultParallelOptions, null, body, null, null, null); } @@ -599,7 +600,7 @@ public static ParallelLoopResult For(int fromInclusive, int toExclusive, Paralle ArgumentNullException.ThrowIfNull(parallelOptions); ArgumentNullException.ThrowIfNull(body); - return ForWorker( + return ForWorker( fromInclusive, toExclusive, parallelOptions, null, body, null, null, null); } @@ -639,7 +640,7 @@ public static ParallelLoopResult For(long fromInclusive, long toExclusive, Paral ArgumentNullException.ThrowIfNull(parallelOptions); ArgumentNullException.ThrowIfNull(body); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, parallelOptions, null, body, null, null, null); } @@ -745,7 +746,7 @@ public static ParallelLoopResult For( ArgumentNullException.ThrowIfNull(body); ArgumentNullException.ThrowIfNull(localFinally); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, s_defaultParallelOptions, null, null, body, localInit, localFinally); } @@ -875,7 +876,7 @@ public static ParallelLoopResult For( ArgumentNullException.ThrowIfNull(body); ArgumentNullException.ThrowIfNull(localFinally); - return ForWorker64( + return ForWorker( fromInclusive, toExclusive, parallelOptions, null, null, body, localInit, localFinally); } @@ -907,267 +908,6 @@ private static int ComputeTimeoutPoint(int timeoutLength) } } - /// - /// Performs the major work of the parallel for loop. It assumes that argument validation has already - /// been performed by the caller. This function's whole purpose in life is to enable as much reuse of - /// common implementation details for the various For overloads we offer. Without it, we'd end up - /// with lots of duplicate code. It handles: (1) simple for loops, (2) for loops that depend on - /// ParallelState, and (3) for loops with thread local data. - /// - /// - /// The type of the local data. - /// The loop's start index, inclusive. - /// The loop's end index, exclusive. - /// A ParallelOptions instance. - /// The simple loop body. - /// The loop body for ParallelState overloads. - /// The loop body for thread local state overloads. - /// A selector function that returns new thread local state. - /// A cleanup function to destroy thread local state. - /// Only one of the body arguments may be supplied (i.e. they are exclusive). - /// A structure. - private static ParallelLoopResult ForWorker( - int fromInclusive, int toExclusive, - ParallelOptions parallelOptions, - Action? body, - Action? bodyWithState, - Func? bodyWithLocal, - Func? localInit, Action? localFinally) - { - Debug.Assert(((body == null ? 0 : 1) + (bodyWithState == null ? 0 : 1) + (bodyWithLocal == null ? 0 : 1)) == 1, - "expected exactly one body function to be supplied"); - Debug.Assert(bodyWithLocal != null || (localInit == null && localFinally == null), - "thread local functions should only be supplied for loops w/ thread local bodies"); - - // Instantiate our result. Specifics will be filled in later. - ParallelLoopResult result = default; - - // We just return immediately if 'to' is smaller (or equal to) 'from'. - if (toExclusive <= fromInclusive) - { - result._completed = true; - return result; - } - - // For all loops we need a shared flag even though we don't have a body with state, - // because the shared flag contains the exceptional bool, which triggers other workers - // to exit their loops if one worker catches an exception - ParallelLoopStateFlags32 sharedPStateFlags = new ParallelLoopStateFlags32(); - - // Before getting started, do a quick peek to see if we have been canceled already - parallelOptions.CancellationToken.ThrowIfCancellationRequested(); - - // initialize ranges with passed in loop arguments and expected number of workers - int numExpectedWorkers = (parallelOptions.EffectiveMaxConcurrencyLevel == -1) ? - Environment.ProcessorCount : - parallelOptions.EffectiveMaxConcurrencyLevel; - RangeManager rangeManager = new RangeManager(fromInclusive, toExclusive, 1, numExpectedWorkers); - - // Keep track of any cancellations - OperationCanceledException? oce = null; - - // if cancellation is enabled, we need to register a callback to stop the loop when it gets signaled - CancellationTokenRegistration ctr = (!parallelOptions.CancellationToken.CanBeCanceled) - ? default(CancellationTokenRegistration) - : parallelOptions.CancellationToken.UnsafeRegister((o) => - { - // Record our cancellation before stopping processing - oce = new OperationCanceledException(parallelOptions.CancellationToken); - // Cause processing to stop - sharedPStateFlags.Cancel(); - }, state: null); - - // ETW event for Parallel For begin - int forkJoinContextID = 0; - if (ParallelEtwProvider.Log.IsEnabled()) - { - forkJoinContextID = Interlocked.Increment(ref s_forkJoinContextID); - ParallelEtwProvider.Log.ParallelLoopBegin(TaskScheduler.Current.Id, Task.CurrentId ?? 0, - forkJoinContextID, ParallelEtwProvider.ForkJoinOperationType.ParallelFor, - fromInclusive, toExclusive); - } - - try - { - try - { - TaskReplicator.Run( - (ref RangeWorker currentWorker, int timeout, out bool replicationDelegateYieldedBeforeCompletion) => - { - // First thing we do upon entering the task is to register as a new "RangeWorker" with the - // shared RangeManager instance. - - if (!currentWorker.IsInitialized) - currentWorker = rangeManager.RegisterNewWorker(); - - // We will need to reset this to true if we exit due to a timeout: - replicationDelegateYieldedBeforeCompletion = false; - - // We need to call FindNewWork32() on it to see whether there's a chunk available. - // These are the local index values to be used in the sequential loop. - // Their values filled in by FindNewWork32 - int nFromInclusiveLocal; - int nToExclusiveLocal; - - if (currentWorker.FindNewWork32(out nFromInclusiveLocal, out nToExclusiveLocal) == false || - sharedPStateFlags.ShouldExitLoop(nFromInclusiveLocal)) - { - return; // no need to run - } - - // ETW event for ParallelFor Worker Fork - if (ParallelEtwProvider.Log.IsEnabled()) - { - ParallelEtwProvider.Log.ParallelFork(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID); - } - - TLocal localValue = default!; - bool bLocalValueInitialized = false; // Tracks whether localInit ran without exceptions, so that we can skip localFinally if it wasn't - - try - { - // Create a new state object that references the shared "stopped" and "exceptional" flags - // If needed, it will contain a new instance of thread-local state by invoking the selector. - ParallelLoopState32? state = null; - - if (bodyWithState != null) - { - Debug.Assert(sharedPStateFlags != null); - state = new ParallelLoopState32(sharedPStateFlags); - } - else if (bodyWithLocal != null) - { - Debug.Assert(sharedPStateFlags != null); - state = new ParallelLoopState32(sharedPStateFlags); - if (localInit != null) - { - localValue = localInit(); - bLocalValueInitialized = true; - } - } - - // initialize a loop timer which will help us decide whether we should exit early - int loopTimeout = ComputeTimeoutPoint(timeout); - - // Now perform the loop itself. - do - { - if (body != null) - { - for (int j = nFromInclusiveLocal; - j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline - || !sharedPStateFlags.ShouldExitLoop()); // the no-arg version is used since we have no state - j += 1) - { - body(j); - } - } - else if (bodyWithState != null) - { - for (int j = nFromInclusiveLocal; - j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline - || !sharedPStateFlags.ShouldExitLoop(j)); - j += 1) - { - state!.CurrentIteration = j; - bodyWithState(j, state); - } - } - else - { - for (int j = nFromInclusiveLocal; - j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline - || !sharedPStateFlags.ShouldExitLoop(j)); - j += 1) - { - state!.CurrentIteration = j; - localValue = bodyWithLocal!(j, state, localValue); - } - } - - // Cooperative multitasking: - // Check if allowed loop time is exceeded, if so save current state and return. - // The task replicator will queue up a replacement task. Note that we don't do this on the root task. - if (CheckTimeoutReached(loopTimeout)) - { - replicationDelegateYieldedBeforeCompletion = true; - break; - } - // Exit DO-loop if we can't find new work, or if the loop was stopped: - } while (currentWorker.FindNewWork32(out nFromInclusiveLocal, out nToExclusiveLocal) && - ((sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone) || - !sharedPStateFlags.ShouldExitLoop(nFromInclusiveLocal))); - } - catch (Exception ex) - { - // if we catch an exception in a worker, we signal the other workers to exit the loop, and we rethrow - sharedPStateFlags.SetExceptional(); - ExceptionDispatchInfo.Throw(ex); - } - finally - { - // If a cleanup function was specified, call it. Otherwise, if the type is - // IDisposable, we will invoke Dispose on behalf of the user. - if (localFinally != null && bLocalValueInitialized) - { - localFinally(localValue); - } - - // ETW event for ParallelFor Worker Join - if (ParallelEtwProvider.Log.IsEnabled()) - { - ParallelEtwProvider.Log.ParallelJoin(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID); - } - } - }, - parallelOptions, - stopOnFirstFailure: true); - } - finally - { - // Dispose the cancellation token registration before checking for a cancellation exception - if (parallelOptions.CancellationToken.CanBeCanceled) - ctr.Dispose(); - } - - // If we got through that with no exceptions, and we were canceled, then - // throw our cancellation exception - if (oce != null) throw oce; - } - catch (AggregateException aggExp) - { - // If we have many cancellation exceptions all caused by the specified user cancel control, then throw only one OCE: - ThrowSingleCancellationExceptionOrOtherException(aggExp.InnerExceptions, parallelOptions.CancellationToken, aggExp); - } - finally - { - int sb_status = sharedPStateFlags.LoopStateFlags; - result._completed = (sb_status == ParallelLoopStateFlags.ParallelLoopStateNone); - if ((sb_status & ParallelLoopStateFlags.ParallelLoopStateBroken) != 0) - { - result._lowestBreakIteration = sharedPStateFlags.LowestBreakIteration; - } - - // ETW event for Parallel For End - if (ParallelEtwProvider.Log.IsEnabled()) - { - int nTotalIterations = 0; - - // calculate how many iterations we ran in total - if (sb_status == ParallelLoopStateFlags.ParallelLoopStateNone) - nTotalIterations = toExclusive - fromInclusive; - else if ((sb_status & ParallelLoopStateFlags.ParallelLoopStateBroken) != 0) - nTotalIterations = sharedPStateFlags.LowestBreakIteration - fromInclusive; - else - nTotalIterations = -1; //ParallelLoopStateStopped! We can't determine this if we were stopped.. - - ParallelEtwProvider.Log.ParallelLoopEnd(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID, nTotalIterations); - } - } - - return result; - } - /// /// Performs the major work of the 64-bit parallel for loop. It assumes that argument validation has already /// been performed by the caller. This function's whole purpose in life is to enable as much reuse of @@ -1177,6 +917,7 @@ private static ParallelLoopResult ForWorker( /// /// /// The type of the local data. + /// The type of the range integer. /// The loop's start index, inclusive. /// The loop's end index, exclusive. /// A ParallelOptions instance. @@ -1187,14 +928,16 @@ private static ParallelLoopResult ForWorker( /// A cleanup function to destroy thread local state. /// Only one of the body arguments may be supplied (i.e. they are exclusive). /// A structure. - private static ParallelLoopResult ForWorker64( - long fromInclusive, long toExclusive, + private static ParallelLoopResult ForWorker( + TInt fromInclusive, TInt toExclusive, ParallelOptions parallelOptions, - Action? body, - Action? bodyWithState, - Func? bodyWithLocal, + Action? body, + Action? bodyWithState, + Func? bodyWithLocal, Func? localInit, Action? localFinally) + where TInt : struct, IBinaryInteger, IMinMaxValue { + Debug.Assert(typeof(TInt) == typeof(int) || typeof(TInt) == typeof(long)); Debug.Assert(((body == null ? 0 : 1) + (bodyWithState == null ? 0 : 1) + (bodyWithLocal == null ? 0 : 1)) == 1, "expected exactly one body function to be supplied"); Debug.Assert(bodyWithLocal != null || (localInit == null && localFinally == null), @@ -1213,7 +956,7 @@ private static ParallelLoopResult ForWorker64( // For all loops we need a shared flag even though we don't have a body with state, // because the shared flag contains the exceptional bool, which triggers other workers // to exit their loops if one worker catches an exception - ParallelLoopStateFlags64 sharedPStateFlags = new ParallelLoopStateFlags64(); + ParallelLoopStateFlags sharedPStateFlags = new ParallelLoopStateFlags(); // Before getting started, do a quick peek to see if we have been canceled already parallelOptions.CancellationToken.ThrowIfCancellationRequested(); @@ -1223,7 +966,7 @@ private static ParallelLoopResult ForWorker64( int numExpectedWorkers = (parallelOptions.EffectiveMaxConcurrencyLevel == -1) ? Environment.ProcessorCount : parallelOptions.EffectiveMaxConcurrencyLevel; - RangeManager rangeManager = new RangeManager(fromInclusive, toExclusive, 1, numExpectedWorkers); + RangeManager rangeManager = new RangeManager(long.CreateTruncating(fromInclusive), long.CreateTruncating(toExclusive), 1, numExpectedWorkers); // Keep track of any cancellations OperationCanceledException? oce = null; @@ -1246,7 +989,7 @@ private static ParallelLoopResult ForWorker64( forkJoinContextID = Interlocked.Increment(ref s_forkJoinContextID); ParallelEtwProvider.Log.ParallelLoopBegin(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID, ParallelEtwProvider.ForkJoinOperationType.ParallelFor, - fromInclusive, toExclusive); + long.CreateTruncating(fromInclusive), long.CreateTruncating(toExclusive)); } try @@ -1268,8 +1011,8 @@ private static ParallelLoopResult ForWorker64( // These are the local index values to be used in the sequential loop. // Their values filled in by FindNewWork - long nFromInclusiveLocal; - long nToExclusiveLocal; + TInt nFromInclusiveLocal; + TInt nToExclusiveLocal; if (currentWorker.FindNewWork(out nFromInclusiveLocal, out nToExclusiveLocal) == false || sharedPStateFlags.ShouldExitLoop(nFromInclusiveLocal)) @@ -1291,17 +1034,17 @@ private static ParallelLoopResult ForWorker64( { // Create a new state object that references the shared "stopped" and "exceptional" flags // If needed, it will contain a new instance of thread-local state by invoking the selector. - ParallelLoopState64? state = null; + ParallelLoopState? state = null; if (bodyWithState != null) { Debug.Assert(sharedPStateFlags != null); - state = new ParallelLoopState64(sharedPStateFlags); + state = new ParallelLoopState(sharedPStateFlags); } else if (bodyWithLocal != null) { Debug.Assert(sharedPStateFlags != null); - state = new ParallelLoopState64(sharedPStateFlags); + state = new ParallelLoopState(sharedPStateFlags); // If a thread-local selector was supplied, invoke it. Otherwise, use the default. if (localInit != null) @@ -1319,20 +1062,20 @@ private static ParallelLoopResult ForWorker64( { if (body != null) { - for (long j = nFromInclusiveLocal; + for (TInt j = nFromInclusiveLocal; j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline || !sharedPStateFlags.ShouldExitLoop()); // the no-arg version is used since we have no state - j += 1) + j++) { body(j); } } else if (bodyWithState != null) { - for (long j = nFromInclusiveLocal; + for (TInt j = nFromInclusiveLocal; j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline || !sharedPStateFlags.ShouldExitLoop(j)); - j += 1) + j++) { state!.CurrentIteration = j; bodyWithState(j, state); @@ -1340,10 +1083,10 @@ private static ParallelLoopResult ForWorker64( } else { - for (long j = nFromInclusiveLocal; + for (TInt j = nFromInclusiveLocal; j < nToExclusiveLocal && (sharedPStateFlags.LoopStateFlags == ParallelLoopStateFlags.ParallelLoopStateNone // fast path check as SEL() doesn't inline || !sharedPStateFlags.ShouldExitLoop(j)); - j += 1) + j++) { state!.CurrentIteration = j; localValue = bodyWithLocal!(j, state, localValue); @@ -1410,13 +1153,13 @@ private static ParallelLoopResult ForWorker64( result._completed = (sb_status == ParallelLoopStateFlags.ParallelLoopStateNone); if ((sb_status & ParallelLoopStateFlags.ParallelLoopStateBroken) != 0) { - result._lowestBreakIteration = sharedPStateFlags.LowestBreakIteration; + result._lowestBreakIteration = long.CreateTruncating(sharedPStateFlags.LowestBreakIteration); } // ETW event for Parallel For End if (ParallelEtwProvider.Log.IsEnabled()) { - long nTotalIterations = 0; + TInt nTotalIterations; // calculate how many iterations we ran in total if (sb_status == ParallelLoopStateFlags.ParallelLoopStateNone) @@ -1424,9 +1167,9 @@ private static ParallelLoopResult ForWorker64( else if ((sb_status & ParallelLoopStateFlags.ParallelLoopStateBroken) != 0) nTotalIterations = sharedPStateFlags.LowestBreakIteration - fromInclusive; else - nTotalIterations = -1; //ParallelLoopStateStopped! We can't determine this if we were stopped.. + nTotalIterations = TInt.CreateTruncating(-1); //ParallelLoopStateStopped! We can't determine this if we were stopped.. - ParallelEtwProvider.Log.ParallelLoopEnd(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID, nTotalIterations); + ParallelEtwProvider.Log.ParallelLoopEnd(TaskScheduler.Current.Id, Task.CurrentId ?? 0, forkJoinContextID, long.CreateTruncating(nTotalIterations)); } } @@ -1982,27 +1725,27 @@ private static ParallelLoopResult ForEachWorker( if (body != null) { - return ForWorker( + return ForWorker( from, to, parallelOptions, (i) => body(array[i]), null, null, null, null); } else if (bodyWithState != null) { - return ForWorker( + return ForWorker( from, to, parallelOptions, null, (i, state) => bodyWithState(array[i], state), null, null, null); } else if (bodyWithStateAndIndex != null) { - return ForWorker( + return ForWorker( from, to, parallelOptions, null, (i, state) => bodyWithStateAndIndex(array[i], state, i), null, null, null); } else if (bodyWithStateAndLocal != null) { - return ForWorker( + return ForWorker( from, to, parallelOptions, null, null, (i, state, local) => bodyWithStateAndLocal(array[i], state, local), localInit, localFinally); } else { - return ForWorker( + return ForWorker( from, to, parallelOptions, null, null, (i, state, local) => bodyWithEverything!(array[i], state, i, local), localInit, localFinally); } } @@ -2038,27 +1781,27 @@ private static ParallelLoopResult ForEachWorker( if (body != null) { - return ForWorker( + return ForWorker( 0, list.Count, parallelOptions, (i) => body(list[i]), null, null, null, null); } else if (bodyWithState != null) { - return ForWorker( + return ForWorker( 0, list.Count, parallelOptions, null, (i, state) => bodyWithState(list[i], state), null, null, null); } else if (bodyWithStateAndIndex != null) { - return ForWorker( + return ForWorker( 0, list.Count, parallelOptions, null, (i, state) => bodyWithStateAndIndex(list[i], state, i), null, null, null); } else if (bodyWithStateAndLocal != null) { - return ForWorker( + return ForWorker( 0, list.Count, parallelOptions, null, null, (i, state, local) => bodyWithStateAndLocal(list[i], state, local), localInit, localFinally); } else { - return ForWorker( + return ForWorker( 0, list.Count, parallelOptions, null, null, (i, state, local) => bodyWithEverything!(list[i], state, i, local), localInit, localFinally); } } @@ -2819,7 +2562,7 @@ private static ParallelLoopResult PartitionerForEachWorker( // For all loops we need a shared flag even though we don't have a body with state, // because the shared flag contains the exceptional bool, which triggers other workers // to exit their loops if one worker catches an exception - ParallelLoopStateFlags64 sharedPStateFlags = new ParallelLoopStateFlags64(); + ParallelLoopStateFlags sharedPStateFlags = new ParallelLoopStateFlags(); // Instantiate our result. Specifics will be filled in later. ParallelLoopResult result = default; @@ -2882,15 +2625,15 @@ private static ParallelLoopResult PartitionerForEachWorker( { // Create a new state object that references the shared "stopped" and "exceptional" flags. // If needed, it will contain a new instance of thread-local state by invoking the selector. - ParallelLoopState64? state = null; + ParallelLoopState? state = null; if (bodyWithState != null || bodyWithStateAndIndex != null) { - state = new ParallelLoopState64(sharedPStateFlags); + state = new ParallelLoopState(sharedPStateFlags); } else if (bodyWithStateAndLocal != null || bodyWithEverything != null) { - state = new ParallelLoopState64(sharedPStateFlags); + state = new ParallelLoopState(sharedPStateFlags); // If a thread-local selector was supplied, invoke it. Otherwise, stick with the default. if (localInit != null) { diff --git a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelLoopState.cs b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelLoopState.cs index 71a1c31457f5c9..e00c096abb97e2 100644 --- a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelLoopState.cs +++ b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelLoopState.cs @@ -9,6 +9,8 @@ // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- using System.Diagnostics; +using System.Numerics; +using System.Runtime.CompilerServices; // Prevents compiler warnings/errors regarding the use of ref params in Interlocked methods @@ -21,7 +23,7 @@ namespace System.Threading.Tasks [DebuggerDisplay("ShouldExitCurrentIteration = {ShouldExitCurrentIteration}")] public class ParallelLoopState { - // Derived classes will track a ParallelStateFlags32 or ParallelStateFlags64. + // Derived classes will track a ParallelLoopStateFlags. // So this is slightly redundant, but it enables us to implement some // methods in this base class. private readonly ParallelLoopStateFlags _flagsBase; @@ -179,50 +181,10 @@ public void Break() InternalBreak(); } - // Helper method to avoid repeating Break() logic between ParallelState32 and ParallelState32 - internal static void Break(int iteration, ParallelLoopStateFlags32 pflags) + internal static void Break(TInt iteration, ParallelLoopStateFlags pflags) where TInt : struct, IBinaryInteger, IMinMaxValue { - int oldValue = ParallelLoopStateFlags.ParallelLoopStateNone; + Debug.Assert(typeof(TInt) == typeof(int) || typeof(TInt) == typeof(long)); - // Attempt to change state from "not stopped or broken or canceled or exceptional" to "broken". - if (!pflags.AtomicLoopStateUpdate(ParallelLoopStateFlags.ParallelLoopStateBroken, - ParallelLoopStateFlags.ParallelLoopStateStopped | ParallelLoopStateFlags.ParallelLoopStateExceptional | ParallelLoopStateFlags.ParallelLoopStateCanceled, - ref oldValue)) - { - // If we were already stopped, we have a problem - if ((oldValue & ParallelLoopStateFlags.ParallelLoopStateStopped) != 0) - { - throw new InvalidOperationException( - SR.ParallelState_Break_InvalidOperationException_BreakAfterStop); - } - else - { - // Apparently we previously got cancelled or became exceptional. No action necessary - return; - } - } - - // replace shared LowestBreakIteration with CurrentIteration, but only if CurrentIteration - // is less than LowestBreakIteration. - int oldLBI = pflags._lowestBreakIteration; - if (iteration < oldLBI) - { - SpinWait wait = default; - while (Interlocked.CompareExchange( - ref pflags._lowestBreakIteration, - iteration, - oldLBI) != oldLBI) - { - wait.SpinOnce(); - oldLBI = pflags._lowestBreakIteration; - if (iteration > oldLBI) break; - } - } - } - - // Helper method to avoid repeating Break() logic between ParallelState64 and ParallelState64 - internal static void Break(long iteration, ParallelLoopStateFlags64 pflags) - { int oldValue = ParallelLoopStateFlags.ParallelLoopStateNone; // Attempt to change state from "not stopped or broken or canceled or exceptional" to "broken". @@ -245,14 +207,13 @@ internal static void Break(long iteration, ParallelLoopStateFlags64 pflags) // replace shared LowestBreakIteration with CurrentIteration, but only if CurrentIteration // is less than LowestBreakIteration. - long oldLBI = pflags.LowestBreakIteration; + TInt oldLBI = pflags.LowestBreakIteration; if (iteration < oldLBI) { SpinWait wait = default; - while (Interlocked.CompareExchange( - ref pflags._lowestBreakIteration, - iteration, - oldLBI) != oldLBI) + while (typeof(TInt) == typeof(int) ? + Interlocked.CompareExchange(ref Unsafe.As(ref pflags._lowestBreakIteration), Unsafe.As(ref iteration), Unsafe.As(ref oldLBI)) != Unsafe.As(ref oldLBI) : + Interlocked.CompareExchange(ref Unsafe.As(ref pflags._lowestBreakIteration), Unsafe.As(ref iteration), Unsafe.As(ref oldLBI)) != Unsafe.As(ref oldLBI)) { wait.SpinOnce(); oldLBI = pflags.LowestBreakIteration; @@ -262,19 +223,20 @@ internal static void Break(long iteration, ParallelLoopStateFlags64 pflags) } } - internal sealed class ParallelLoopState32 : ParallelLoopState + internal sealed class ParallelLoopState : ParallelLoopState where TInt : struct, IBinaryInteger, IMinMaxValue { - private readonly ParallelLoopStateFlags32 _sharedParallelStateFlags; - private int _currentIteration; + private readonly ParallelLoopStateFlags _sharedParallelStateFlags; + private TInt _currentIteration; /// /// Internal constructor to ensure an instance isn't created by users. /// /// A flag shared among all threads participating /// in the execution of a certain loop. - internal ParallelLoopState32(ParallelLoopStateFlags32 sharedParallelStateFlags) + internal ParallelLoopState(ParallelLoopStateFlags sharedParallelStateFlags) : base(sharedParallelStateFlags) { + Debug.Assert(typeof(TInt) == typeof(int) || typeof(TInt) == typeof(long)); _sharedParallelStateFlags = sharedParallelStateFlags; } @@ -283,7 +245,7 @@ internal ParallelLoopState32(ParallelLoopStateFlags32 sharedParallelStateFlags) /// This is used to compute whether or not the task should /// terminate early due to a Break() call. /// - internal int CurrentIteration + internal TInt CurrentIteration { get { return _currentIteration; } set { _currentIteration = value; } @@ -319,81 +281,12 @@ internal override long? InternalLowestBreakIteration /// internal override void InternalBreak() { - ParallelLoopState.Break(CurrentIteration, _sharedParallelStateFlags); - } - } - - /// - /// Allows independent iterations of a parallel loop to interact with other iterations. - /// - internal sealed class ParallelLoopState64 : ParallelLoopState - { - private readonly ParallelLoopStateFlags64 _sharedParallelStateFlags; - private long _currentIteration; - - /// - /// Internal constructor to ensure an instance isn't created by users. - /// - /// A flag shared among all threads participating - /// in the execution of a certain loop. - internal ParallelLoopState64(ParallelLoopStateFlags64 sharedParallelStateFlags) - : base(sharedParallelStateFlags) - { - _sharedParallelStateFlags = sharedParallelStateFlags; - } - - /// - /// Tracks the current loop iteration for the owning task. - /// This is used to compute whether or not the task should - /// terminate early due to a Break() call. - /// - internal long CurrentIteration - { - // No interlocks needed, because this value is only accessed in a single thread. - get { return _currentIteration; } - set { _currentIteration = value; } - } - - /// - /// Returns true if we should be exiting from the current iteration - /// due to Stop(), Break() or exception. - /// - internal override bool InternalShouldExitCurrentIteration - { - get { return _sharedParallelStateFlags.ShouldExitLoop(CurrentIteration); } - } - - /// - /// Returns the lowest iteration at which Break() has been called, or - /// null if Break() has not yet been called. - /// - internal override long? InternalLowestBreakIteration - { - // We don't need to worry about torn read/write here because - // ParallelStateFlags64.LowestBreakIteration property is protected - // by an Interlocked.Read(). - get { return _sharedParallelStateFlags.NullableLowestBreakIteration; } - } - - /// - /// Communicates that parallel tasks should stop when they reach a specified iteration element. - /// (which is CurrentIteration of the caller). - /// - /// Break() called after Stop(). - /// - /// Atomically sets shared StoppedBroken flag to BROKEN, then atomically sets shared - /// LowestBreakIteration to CurrentIteration, but only if CurrentIteration is less than - /// LowestBreakIteration. - /// - internal override void InternalBreak() - { - ParallelLoopState.Break(CurrentIteration, _sharedParallelStateFlags); + Break(CurrentIteration, _sharedParallelStateFlags); } } /// - /// State information that is common between ParallelStateFlags class - /// and ParallelStateFlags64 class. + /// State information that is common between ParallelLoopStateFlags{TInt} types. /// internal class ParallelLoopStateFlags { @@ -455,90 +348,24 @@ internal bool Cancel() } } - /// - /// An internal class used to share accounting information in 32-bit versions - /// of For()/ForEach() loops. - /// - internal sealed class ParallelLoopStateFlags32 : ParallelLoopStateFlags + internal sealed unsafe class ParallelLoopStateFlags : ParallelLoopStateFlags + where TInt : struct, IBinaryInteger, IMinMaxValue { // Records the lowest iteration at which a Break() has been called, - // or Int32.MaxValue if no break has been called. Used directly - // by Break(). - internal volatile int _lowestBreakIteration = int.MaxValue; + // or TInt.MaxValue if no break has been called. + internal TInt _lowestBreakIteration = TInt.MaxValue; - // Not strictly necessary, but maintains consistency with ParallelStateFlags64 - internal int LowestBreakIteration - { - get { return _lowestBreakIteration; } - } - - // Does some processing to convert _lowestBreakIteration to a long?. - internal long? NullableLowestBreakIteration + internal TInt LowestBreakIteration { get { - if (_lowestBreakIteration == int.MaxValue) return null; - else + if (typeof(TInt) == typeof(int)) { - // protect against torn read of 64-bit value - long rval = _lowestBreakIteration; - if (IntPtr.Size >= 8) return rval; - else return Interlocked.Read(ref rval); + return Unsafe.BitCast(Volatile.Read(ref Unsafe.As(ref _lowestBreakIteration))); } - } - } - - /// - /// Lets the caller know whether or not to prematurely exit the For/ForEach loop. - /// If this returns true, then exit the loop. Otherwise, keep going. - /// - /// The caller's current iteration point - /// in the loop. - /// - /// The loop should exit on any one of the following conditions: - /// (1) Stop() has been called by one or more tasks. - /// (2) An exception has been raised by one or more tasks. - /// (3) Break() has been called by one or more tasks, and - /// CallerIteration exceeds the (lowest) iteration at which - /// Break() was called. - /// (4) The loop was canceled. - /// - internal bool ShouldExitLoop(int CallerIteration) - { - int flags = LoopStateFlags; - return (flags != ParallelLoopStateNone && ( - ((flags & (ParallelLoopStateExceptional | ParallelLoopStateStopped | ParallelLoopStateCanceled)) != 0) || - (((flags & ParallelLoopStateBroken) != 0) && (CallerIteration > LowestBreakIteration)))); - } - - // This lighter version of ShouldExitLoop will be used when the body type doesn't contain a state. - // Since simpler bodies cannot stop or break, we can safely skip checks for those flags here. - internal bool ShouldExitLoop() - { - int flags = LoopStateFlags; - return ((flags != ParallelLoopStateNone) && ((flags & (ParallelLoopStateExceptional | ParallelLoopStateCanceled)) != 0)); - } - } - - /// - /// An internal class used to share accounting information in 64-bit versions - /// of For()/ForEach() loops. - /// - internal sealed class ParallelLoopStateFlags64 : ParallelLoopStateFlags - { - // Records the lowest iteration at which a Break() has been called, - // or Int64.MaxValue if no break has been called. Used directly - // by Break(). - internal long _lowestBreakIteration = long.MaxValue; - - // Performs a conditionally interlocked read of _lowestBreakIteration. - internal long LowestBreakIteration - { - get - { - if (IntPtr.Size >= 8) return _lowestBreakIteration; - else return Interlocked.Read(ref _lowestBreakIteration); + Debug.Assert(typeof(TInt) == typeof(long)); + return Unsafe.BitCast(Volatile.Read(ref Unsafe.As(ref _lowestBreakIteration))); } } @@ -547,12 +374,8 @@ internal long? NullableLowestBreakIteration { get { - if (_lowestBreakIteration == long.MaxValue) return null; - else - { - if (IntPtr.Size >= 8) return _lowestBreakIteration; - else return Interlocked.Read(ref _lowestBreakIteration); - } + TInt lowestBreakIteration = LowestBreakIteration; + return lowestBreakIteration == TInt.MaxValue ? null : long.CreateTruncating(lowestBreakIteration); } } @@ -571,7 +394,7 @@ internal long? NullableLowestBreakIteration /// Break() was called. /// (4) The loop has been canceled. /// - internal bool ShouldExitLoop(long CallerIteration) + internal bool ShouldExitLoop(TInt CallerIteration) { int flags = LoopStateFlags; return (flags != ParallelLoopStateNone && ( diff --git a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelRangeManager.cs b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelRangeManager.cs index 0112b3b25b0539..c53a3346c7d9b7 100644 --- a/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelRangeManager.cs +++ b/src/libraries/System.Threading.Tasks.Parallel/src/System/Threading/Tasks/ParallelRangeManager.cs @@ -8,6 +8,7 @@ // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- using System.Diagnostics; +using System.Numerics; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; @@ -89,7 +90,7 @@ internal RangeWorker(IndexRange[] ranges, int nInitialRange, long nStep, bool us /// to execute the sequential loop /// 3) if we return false it means there is no more work left. It's time to quit. /// - internal bool FindNewWork(out long nFromInclusiveLocal, out long nToExclusiveLocal) + private bool FindNewWork(out long nFromInclusiveLocal, out long nToExclusiveLocal) { // since we iterate over index ranges circularly, we will use the // count of visited ranges as our exit condition @@ -172,25 +173,20 @@ internal bool FindNewWork(out long nFromInclusiveLocal, out long nToExclusiveLoc return false; } - - /// - /// 32 bit integer version of FindNewWork. Assumes the ranges were initialized with 32 bit values. - /// - internal bool FindNewWork32(out int nFromInclusiveLocal32, out int nToExclusiveLocal32) + internal bool FindNewWork(out TInt fromInclusive, out TInt toExclusive) where TInt : struct, IBinaryInteger, IMinMaxValue { - long nFromInclusiveLocal; - long nToExclusiveLocal; + Debug.Assert(typeof(TInt) == typeof(int) || typeof(TInt) == typeof(long)); - bool bRetVal = FindNewWork(out nFromInclusiveLocal, out nToExclusiveLocal); + bool success = FindNewWork(out long fromInclusiveInt64, out long toExclusiveInt64); - Debug.Assert((nFromInclusiveLocal <= int.MaxValue) && (nFromInclusiveLocal >= int.MinValue) && - (nToExclusiveLocal <= int.MaxValue) && (nToExclusiveLocal >= int.MinValue)); + Debug.Assert( + fromInclusiveInt64 <= long.CreateTruncating(TInt.MaxValue) && fromInclusiveInt64 >= long.CreateTruncating(TInt.MinValue) && + toExclusiveInt64 <= long.CreateTruncating(TInt.MaxValue) && toExclusiveInt64 >= long.CreateTruncating(TInt.MinValue)); - // convert to 32 bit before returning - nFromInclusiveLocal32 = (int)nFromInclusiveLocal; - nToExclusiveLocal32 = (int)nToExclusiveLocal; + fromInclusive = TInt.CreateTruncating(fromInclusiveInt64); + toExclusive = TInt.CreateTruncating(toExclusiveInt64); - return bRetVal; + return success; } }