ForEachAsync< TSource >() [3/8]

static Task System.Threading.Tasks.Parallel.ForEachAsync< TSource > ( IAsyncEnumerable< TSource > source, int dop, TaskScheduler scheduler, CancellationToken cancellationToken, Func< TSource, CancellationToken, ValueTask > body )

inlinestaticprivate

Definition at line 1606 of file Parallel.cs.

    {
        if (cancellationToken.IsCancellationRequested)
        {
            return Task.FromCanceled(cancellationToken);
        }
        if (dop < 0)
        {
            dop = DefaultDegreeOfParallelism;
        }
        Func<object, Task> taskBody = async delegate(object o)
        {
            AsyncForEachAsyncState<TSource> state = (AsyncForEachAsyncState<TSource>)o;
            bool launchedNext = false;
            try
            {
                _ = 2;
                try
                {
                    while (!state.Cancellation.IsCancellationRequested)
                    {
                        await state.Lock.WaitAsync(state.Cancellation.Token);
                        TSource current;
                        try
                        {
                            if (await state.Enumerator.MoveNextAsync())
                            {
                                current = state.Enumerator.Current;
                                goto IL_016d;
                            }
                        }
                        finally
                        {
                            state.Lock.Release();
                        }
                        break;
                        IL_016d:
                        if (!launchedNext)
                        {
                            launchedNext = true;
                            state.QueueWorkerIfDopAvailable();
                        }
                        await state.LoopBody(current, state.Cancellation.Token);
                    }
                }
                catch (Exception e)
                {
                    state.RecordException(e);
                }
            }
            finally
            {
                if (state.SignalWorkerCompletedIterating())
                {
                    try
                    {
                        await state.DisposeAsync();
                    }
                    catch (Exception e2)
                    {
                        state.RecordException(e2);
                    }
                    state.Complete();
                }
            }
        };
        try
        {
            AsyncForEachAsyncState<TSource> asyncForEachAsyncState = new AsyncForEachAsyncState<TSource>(source, taskBody, dop, scheduler, cancellationToken, body);
            asyncForEachAsyncState.QueueWorkerIfDopAvailable();
            return asyncForEachAsyncState.Task;
        }
        catch (Exception exception)
        {
            return Task.FromException(exception);
        }
    }

References System.cancellationToken, System.Threading.Tasks.Parallel.DefaultDegreeOfParallelism, System.exception, System.Threading.Tasks.Task< TResult >.FromCanceled(), System.Threading.Tasks.Task< TResult >.FromException(), System.scheduler, System.source, and System.state.