PartitionerForEachWorker< TSource, TLocal >()

static ParallelLoopResult System.Threading.Tasks.Parallel.PartitionerForEachWorker< TSource, TLocal > ( Partitioner< TSource > source, ParallelOptions parallelOptions, Action< TSource > simpleBody, Action< TSource, ParallelLoopState > bodyWithState, Action< TSource, ParallelLoopState, long > bodyWithStateAndIndex, Func< TSource, ParallelLoopState, TLocal, TLocal > bodyWithStateAndLocal, Func< TSource, ParallelLoopState, long, TLocal, TLocal > bodyWithEverything, Func< TLocal > localInit, Action< TLocal > localFinally )

inlinestaticprivate

Definition at line 1206 of file Parallel.cs.

    {
        OrderablePartitioner<TSource> orderedSource = source as OrderablePartitioner<TSource>;
        if (!source.SupportsDynamicPartitions)
        {
            throw new InvalidOperationException(System.SR.Parallel_ForEach_PartitionerNotDynamic);
        }
        parallelOptions.CancellationToken.ThrowIfCancellationRequested();
        int forkJoinContextID = 0;
        if (ParallelEtwProvider.Log.IsEnabled())
        {
            forkJoinContextID = Interlocked.Increment(ref s_forkJoinContextID);
            ParallelEtwProvider.Log.ParallelLoopBegin(TaskScheduler.Current.Id, Task.CurrentId.GetValueOrDefault(), forkJoinContextID, ParallelEtwProvider.ForkJoinOperationType.ParallelForEach, 0L, 0L);
        }
        ParallelLoopStateFlags64 sharedPStateFlags = new ParallelLoopStateFlags64();
        ParallelLoopResult result = default(ParallelLoopResult);
        OperationCanceledException oce = null;
        CancellationTokenRegistration cancellationTokenRegistration = ((!parallelOptions.CancellationToken.CanBeCanceled) ? default(CancellationTokenRegistration) : parallelOptions.CancellationToken.UnsafeRegister((Action<object?>)delegate
        {
            oce = new OperationCanceledException(parallelOptions.CancellationToken);
            sharedPStateFlags.Cancel();
        }, (object?)null));
        IEnumerable<TSource> partitionerSource = null;
        IEnumerable<KeyValuePair<long, TSource>> orderablePartitionerSource = null;
        if (orderedSource != null)
        {
            orderablePartitionerSource = orderedSource.GetOrderableDynamicPartitions();
            if (orderablePartitionerSource == null)
            {
                throw new InvalidOperationException(System.SR.Parallel_ForEach_PartitionerReturnedNull);
            }
        }
        else
        {
            partitionerSource = source.GetDynamicPartitions();
            if (partitionerSource == null)
            {
                throw new InvalidOperationException(System.SR.Parallel_ForEach_PartitionerReturnedNull);
            }
        }
        try
        {
            try
            {
                TaskReplicator.Run(delegate(ref IEnumerator partitionState, int timeout, out bool replicationDelegateYieldedBeforeCompletion)
                {
                    replicationDelegateYieldedBeforeCompletion = false;
                    if (ParallelEtwProvider.Log.IsEnabled())
                    {
                        ParallelEtwProvider.Log.ParallelFork(TaskScheduler.Current.Id, Task.CurrentId.GetValueOrDefault(), forkJoinContextID);
                    }
                    TLocal val = default(TLocal);
                    bool flag = false;
                    try
                    {
                        ParallelLoopState64 parallelLoopState = null;
                        if (bodyWithState != null || bodyWithStateAndIndex != null)
                        {
                            parallelLoopState = new ParallelLoopState64(sharedPStateFlags);
                        }
                        else if (bodyWithStateAndLocal != null || bodyWithEverything != null)
                        {
                            parallelLoopState = new ParallelLoopState64(sharedPStateFlags);
                            if (localInit != null)
                            {
                                val = localInit();
                                flag = true;
                            }
                        }
                        int timeoutOccursAt = ComputeTimeoutPoint(timeout);
                        if (orderedSource != null)
                        {
                            IEnumerator<KeyValuePair<long, TSource>> enumerator = partitionState as IEnumerator<KeyValuePair<long, TSource>>;
                            if (enumerator == null)
                            {
                                enumerator = (IEnumerator<KeyValuePair<long, TSource>>)(partitionState = orderablePartitionerSource.GetEnumerator());
                            }
                            if (enumerator == null)
                            {
                                throw new InvalidOperationException(System.SR.Parallel_ForEach_NullEnumerator);
                            }
                            while (enumerator.MoveNext())
                            {
                                KeyValuePair<long, TSource> current = enumerator.Current;
                                long key = current.Key;
                                TSource value = current.Value;
                                if (parallelLoopState != null)
                                {
                                    parallelLoopState.CurrentIteration = key;
                                }
                                if (simpleBody != null)
                                {
                                    simpleBody(value);
                                }
                                else if (bodyWithState != null)
                                {
                                    bodyWithState(value, parallelLoopState);
                                }
                                else if (bodyWithStateAndIndex == null)
                                {
                                    val = ((bodyWithStateAndLocal == null) ? bodyWithEverything(value, parallelLoopState, key, val) : bodyWithStateAndLocal(value, parallelLoopState, val));
                                }
                                else
                                {
                                    bodyWithStateAndIndex(value, parallelLoopState, key);
                                }
                                if (sharedPStateFlags.ShouldExitLoop(key))
                                {
                                    break;
                                }
                                if (CheckTimeoutReached(timeoutOccursAt))
                                {
                                    replicationDelegateYieldedBeforeCompletion = true;
                                    break;
                                }
                            }
                        }
                        else
                        {
                            IEnumerator<TSource> enumerator2 = partitionState as IEnumerator<TSource>;
                            if (enumerator2 == null)
                            {
                                enumerator2 = (IEnumerator<TSource>)(partitionState = partitionerSource.GetEnumerator());
                            }
                            if (enumerator2 == null)
                            {
                                throw new InvalidOperationException(System.SR.Parallel_ForEach_NullEnumerator);
                            }
                            if (parallelLoopState != null)
                            {
                                parallelLoopState.CurrentIteration = 0L;
                            }
                            while (enumerator2.MoveNext())
                            {
                                TSource current2 = enumerator2.Current;
                                if (simpleBody != null)
                                {
                                    simpleBody(current2);
                                }
                                else if (bodyWithState != null)
                                {
                                    bodyWithState(current2, parallelLoopState);
                                }
                                else if (bodyWithStateAndLocal != null)
                                {
                                    val = bodyWithStateAndLocal(current2, parallelLoopState, val);
                                }
                                if (sharedPStateFlags.LoopStateFlags != 0)
                                {
                                    break;
                                }
                                if (CheckTimeoutReached(timeoutOccursAt))
                                {
                                    replicationDelegateYieldedBeforeCompletion = true;
                                    break;
                                }
                            }
                        }
                    }
                    catch (Exception source2)
                    {
                        sharedPStateFlags.SetExceptional();
                        ExceptionDispatchInfo.Throw(source2);
                    }
                    finally
                    {
                        if (localFinally != null && flag)
                        {
                            localFinally(val);
                        }
                        if (!replicationDelegateYieldedBeforeCompletion && partitionState is IDisposable disposable2)
                        {
                            disposable2.Dispose();
                        }
                        if (ParallelEtwProvider.Log.IsEnabled())
                        {
                            ParallelEtwProvider.Log.ParallelJoin(TaskScheduler.Current.Id, Task.CurrentId.GetValueOrDefault(), forkJoinContextID);
                        }
                    }
                }, parallelOptions, stopOnFirstFailure: true);
            }
            finally
            {
                if (parallelOptions.CancellationToken.CanBeCanceled)
                {
                    cancellationTokenRegistration.Dispose();
                }
            }
            if (oce != null)
            {
                throw oce;
            }
        }
        catch (AggregateException ex)
        {
            ThrowSingleCancellationExceptionOrOtherException(ex.InnerExceptions, parallelOptions.CancellationToken, ex);
        }
        finally
        {
            int loopStateFlags = sharedPStateFlags.LoopStateFlags;
            result._completed = loopStateFlags == 0;
            if (((uint)loopStateFlags & 2u) != 0)
            {
                result._lowestBreakIteration = sharedPStateFlags.LowestBreakIteration;
            }
            IDisposable disposable = null;
            ((orderablePartitionerSource == null) ? (partitionerSource as IDisposable) : (orderablePartitionerSource as IDisposable))?.Dispose();
            if (ParallelEtwProvider.Log.IsEnabled())
            {
                ParallelEtwProvider.Log.ParallelLoopEnd(TaskScheduler.Current.Id, Task.CurrentId.GetValueOrDefault(), forkJoinContextID, 0L);
            }
        }
        return result;
    }

References System.Threading.Tasks.Parallel.CheckTimeoutReached(), System.Threading.Tasks.Parallel.ComputeTimeoutPoint(), System.Threading.Tasks.TaskScheduler.Current, System.Threading.Tasks.Task< TResult >.CurrentId, System.Collections.Generic.Dictionary< TKey, TValue >.GetEnumerator(), System.Threading.Interlocked.Increment(), System.key, System.L, System.Threading.Tasks.ParallelEtwProvider.Log, System.SR.Parallel_ForEach_NullEnumerator, System.SR.Parallel_ForEach_PartitionerNotDynamic, System.SR.Parallel_ForEach_PartitionerReturnedNull, System.Threading.Tasks.Parallel.s_forkJoinContextID, System.source, System.Runtime.ExceptionServices.ExceptionDispatchInfo.Throw(), System.Threading.Tasks.Parallel.ThrowSingleCancellationExceptionOrOtherException(), System.timeout, and System.value.