Add Parallel.ForEachAsync (#46943)

* Add Parallel.ForEachAsync

* Address PR feedback and add a few more tests

Also clean up a few things in the src, adding a few more asserts and moving nullable suppressions closer to the point where they're relevant.

src/libraries/System.Threading.Tasks.Parallel/ref/System.Threading.Tasks.Parallel.cs

@@ -36,6 +36,12 @@ public static partial class Parallel
         public static System.Threading.Tasks.ParallelLoopResult ForEach<TSource, TLocal>(System.Collections.Generic.IEnumerable<TSource> source, System.Func<TLocal> localInit, System.Func<TSource, System.Threading.Tasks.ParallelLoopState, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }
         public static System.Threading.Tasks.ParallelLoopResult ForEach<TSource, TLocal>(System.Collections.Generic.IEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TLocal> localInit, System.Func<TSource, System.Threading.Tasks.ParallelLoopState, long, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }
         public static System.Threading.Tasks.ParallelLoopResult ForEach<TSource, TLocal>(System.Collections.Generic.IEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TLocal> localInit, System.Func<TSource, System.Threading.Tasks.ParallelLoopState, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, CancellationToken cancellationToken, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, CancellationToken cancellationToken, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
+        public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
         public static System.Threading.Tasks.ParallelLoopResult For<TLocal>(int fromInclusive, int toExclusive, System.Func<TLocal> localInit, System.Func<int, System.Threading.Tasks.ParallelLoopState, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }
         public static System.Threading.Tasks.ParallelLoopResult For<TLocal>(int fromInclusive, int toExclusive, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TLocal> localInit, System.Func<int, System.Threading.Tasks.ParallelLoopState, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }
         public static System.Threading.Tasks.ParallelLoopResult For<TLocal>(long fromInclusive, long toExclusive, System.Func<TLocal> localInit, System.Func<long, System.Threading.Tasks.ParallelLoopState, TLocal, TLocal> body, System.Action<TLocal> localFinally) { throw null; }

src/libraries/System.Threading.Tasks.Parallel/src/Resources/Strings.resx

@@ -64,13 +64,13 @@
     <value>This method requires the use of an OrderedPartitioner with the KeysNormalized property set to true.</value>
   </data>
   <data name="Parallel_ForEach_PartitionerNotDynamic" xml:space="preserve">
-    <value>The Partitioner used here must support dynamic partitioning.</value>
+    <value>The Partitioner must support dynamic partitioning.</value>
   </data>
   <data name="Parallel_ForEach_PartitionerReturnedNull" xml:space="preserve">
-    <value>The Partitioner used here returned a null partitioner source.</value>
+    <value>The Partitioner returned a null partitioner source.</value>
   </data>
   <data name="Parallel_ForEach_NullEnumerator" xml:space="preserve">
-    <value>The Partitioner source returned a null enumerator.</value>
+    <value>The source returned a null enumerator.</value>
   </data>
   <data name="ParallelState_Break_InvalidOperationException_BreakAfterStop" xml:space="preserve">
     <value>Break was called after Stop was called.</value>
@@ -81,4 +81,4 @@
   <data name="ParallelState_NotSupportedException_UnsupportedMethod" xml:space="preserve">
     <value>This method is not supported.</value>
   </data>
-</root>
+</root>

src/libraries/System.Threading.Tasks.Parallel/src/System.Threading.Tasks.Parallel.csproj

@@ -7,17 +7,20 @@
   </PropertyGroup>
   <ItemGroup>
     <Compile Include="System\Threading\Tasks\Parallel.cs" />
+    <Compile Include="System\Threading\Tasks\Parallel.ForEachAsync.cs" />
     <Compile Include="System\Threading\Tasks\ParallelETWProvider.cs" />
     <Compile Include="System\Threading\Tasks\ParallelLoopState.cs" />
     <Compile Include="System\Threading\Tasks\ParallelRangeManager.cs" />
     <Compile Include="System\Threading\Tasks\TaskReplicator.cs" />
   </ItemGroup>
   <ItemGroup>
+    <Reference Include="System.Collections" />
     <Reference Include="System.Collections.Concurrent" />
     <Reference Include="System.Diagnostics.Tracing" />
     <Reference Include="System.Runtime" />
     <Reference Include="System.Runtime.Extensions" />
     <Reference Include="System.Threading" />
     <Reference Include="System.Threading.Tasks" />
+    <Reference Include="System.Threading.ThreadPool" />
   </ItemGroup>
 </Project>