Move RoslynParallel and ProducerConsumer to threading source package (#79474)

Author: tmat

src/Dependencies/Threading/ChannelReaderExtensions.cs

@@ -0,0 +1,27 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+#nullable enable
+
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+
+namespace System.Threading.Channels;
+
+internal static class RoslynChannelReaderExtensions
+{
+#if NET // binary compatibility
+    public static IAsyncEnumerable<T> ReadAllAsync<T>(ChannelReader<T> reader, CancellationToken cancellationToken)
+        => reader.ReadAllAsync(cancellationToken);
+#else
+    public static async IAsyncEnumerable<T> ReadAllAsync<T>(this ChannelReader<T> reader, [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
+        {
+            while (reader.TryRead(out var item))
+                yield return item;
+        }
+    }
+#endif
+}

src/Dependencies/Threading/IAsyncEnumerableExtensions.cs

@@ -0,0 +1,35 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.CodeAnalysis.PooledObjects;
+
+namespace Microsoft.CodeAnalysis;
+
+internal static class IAsyncEnumerableExtensions
+{
+    public static async Task<ImmutableArray<T>> ToImmutableArrayAsync<T>(this IAsyncEnumerable<T> values, CancellationToken cancellationToken)
+    {
+        using var _ = ArrayBuilder<T>.GetInstance(out var result);
+
+        await foreach (var value in values.WithCancellation(cancellationToken).ConfigureAwait(false))
+            result.Add(value);
+
+        return result.ToImmutableAndClear();
+    }
+
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+#pragma warning disable VSTHRD200 // Use "Async" suffix for async methods
+    public static async IAsyncEnumerable<TSource> AsAsyncEnumerable<TSource>(this IEnumerable<TSource> source)
+#pragma warning restore VSTHRD200 // Use "Async" suffix for async methods
+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
+    {
+        foreach (var item in source)
+            yield return item;
+    }
+}

src/Dependencies/Threading/Microsoft.CodeAnalysis.Threading.Package.csproj

@@ -2,7 +2,7 @@
 <!-- Licensed to the .NET Foundation under one or more agreements. The .NET Foundation licenses this file to you under the MIT license. See the LICENSE file in the project root for more information. -->
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <TargetFramework>netstandard2.0</TargetFramework>
+    <TargetFrameworks>$(NetRoslyn);netstandard2.0</TargetFrameworks>
     <GenerateDocumentationFile>false</GenerateDocumentationFile>
     <DebugType>none</DebugType>
     <GenerateDependencyFile>false</GenerateDependencyFile>
@@ -21,7 +21,9 @@
   </PropertyGroup>
   <ItemGroup>
     <PackageReference Include="System.Collections.Immutable" />
+    <PackageReference Include="System.Threading.Channels" />
     <PackageReference Include="System.Threading.Tasks.Extensions" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" />
   </ItemGroup>
   <Import Project="..\Collections\Microsoft.CodeAnalysis.Collections.projitems" Label="Shared" />
   <Import Project="..\PooledObjects\Microsoft.CodeAnalysis.PooledObjects.projitems" Label="Shared" />

src/Dependencies/Threading/Microsoft.CodeAnalysis.Threading.projitems

@@ -23,6 +23,12 @@
     <Compile Include="$(MSBuildThisFileDirectory)TestHooks\IAsynchronousOperationListenerProvider.cs" />
     <Compile Include="$(MSBuildThisFileDirectory)TestHooks\IAsynchronousOperationListener.cs" />
     <Compile Include="$(MSBuildThisFileDirectory)YieldAwaitableExtensions.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)ParallelExtensions.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)ParallelExtensions.NetFramework.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)IAsyncEnumerableExtensions.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)ChannelReaderExtensions.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)ProducerConsumer.cs" />
+    <Compile Include="$(MSBuildThisFileDirectory)ProducerConsumerOptions.cs" />
   </ItemGroup>
   <ItemGroup Condition="'$(DefaultLanguageSourceExtension)' != '' AND '$(BuildingInsideVisualStudio)' != 'true'">
     <ExpectedCompile Include="$(MSBuildThisFileDirectory)**\*$(DefaultLanguageSourceExtension)" />

src/Workspaces/SharedUtilitiesAndExtensions/Compiler/Core/Utilities/RoslynParallel.NetFramework.cs renamed to src/Dependencies/Threading/ParallelExtensions.NetFramework.cs

@@ -17,17 +17,13 @@
 // With only changes to make the code work on NetFx.  Where changes have been made, the original code is kept around in
 // an ifdef'ed block to see what it was doing.
 
-using System;
 using System.Collections.Generic;
 using System.Diagnostics;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.CodeAnalysis.Threading;
-using Roslyn.Utilities;
 
-namespace Microsoft.CodeAnalysis.Shared.Utilities;
+namespace System.Threading.Tasks;
 
-internal static partial class RoslynParallel
+internal static partial class RoslynParallelExtensions
 {
     private static class NetFramework
     {

src/Dependencies/Threading/ParallelExtensions.cs

@@ -0,0 +1,43 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+#nullable enable
+
+#pragma warning disable CA1068 // CancellationToken parameters must come last
+
+using System.Collections.Generic;
+
+namespace System.Threading.Tasks;
+
+internal static partial class RoslynParallelExtensions
+{
+#if NET // binary compatibility
+    public static Task ForEachAsync<TSource>(IEnumerable<TSource> source, CancellationToken cancellationToken, Func<TSource, CancellationToken, ValueTask> body)
+        => Parallel.ForEachAsync(source, cancellationToken, body);
+
+    public static Task ForEachAsync<TSource>(IEnumerable<TSource> source, ParallelOptions parallelOptions, Func<TSource, CancellationToken, ValueTask> body)
+        => Parallel.ForEachAsync(source, parallelOptions, body);
+
+    public static Task ForEachAsync<TSource>(IAsyncEnumerable<TSource> source, CancellationToken cancellationToken, Func<TSource, CancellationToken, ValueTask> body)
+        => Parallel.ForEachAsync(source, cancellationToken, body);
+
+    public static Task ForEachAsync<TSource>(IAsyncEnumerable<TSource> source, ParallelOptions parallelOptions, Func<TSource, CancellationToken, ValueTask> body)
+        => Parallel.ForEachAsync(source, parallelOptions, body);
+#else
+    extension(Parallel)
+    {
+        public static Task ForEachAsync<TSource>(IEnumerable<TSource> source, CancellationToken cancellationToken, Func<TSource, CancellationToken, ValueTask> body)
+            => NetFramework.ForEachAsync(source, cancellationToken, body);
+
+        public static Task ForEachAsync<TSource>(IEnumerable<TSource> source, ParallelOptions parallelOptions, Func<TSource, CancellationToken, ValueTask> body)
+            => NetFramework.ForEachAsync(source, parallelOptions, body);
+
+        public static Task ForEachAsync<TSource>(IAsyncEnumerable<TSource> source, CancellationToken cancellationToken, Func<TSource, CancellationToken, ValueTask> body)
+            => NetFramework.ForEachAsync(source, cancellationToken, body);
+
+        public static Task ForEachAsync<TSource>(IAsyncEnumerable<TSource> source, ParallelOptions parallelOptions, Func<TSource, CancellationToken, ValueTask> body)
+            => NetFramework.ForEachAsync(source, parallelOptions, body);
+    }
+#endif
+}

src/Workspaces/SharedUtilitiesAndExtensions/Compiler/Core/Utilities/ProducerConsumer.cs renamed to src/Dependencies/Threading/ProducerConsumer.cs

@@ -2,38 +2,19 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
+#nullable enable
+
 using System;
 using System.Collections.Generic;
 using System.Collections.Immutable;
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.CodeAnalysis.PooledObjects;
-using Microsoft.CodeAnalysis.Shared.Extensions;
 using Microsoft.CodeAnalysis.Threading;
+using Roslyn.Utilities;
 
-namespace Microsoft.CodeAnalysis.Shared.Utilities;
-
-internal readonly record struct ProducerConsumerOptions
-{
-    /// <summary>
-    /// Used when the consumeItems routine will only pull items on a single thread (never concurrently). produceItems
-    /// can be called concurrently on many threads.
-    /// </summary>
-    public static readonly ProducerConsumerOptions SingleReaderOptions = new() { SingleReader = true };
-
-    /// <summary>
-    /// Used when the consumeItems routine will only pull items on a single thread (never concurrently). produceItems
-    /// can be called on a single thread as well (never concurrently).
-    /// </summary>
-    public static readonly ProducerConsumerOptions SingleReaderWriterOptions = new() { SingleReader = true, SingleWriter = true };
-
-    /// <inheritdoc cref="ChannelOptions.SingleWriter"/>
-    public bool SingleWriter { get; init; }
-
-    /// <inheritdoc cref="ChannelOptions.SingleReader"/>
-    public bool SingleReader { get; init; }
-}
+namespace Microsoft.CodeAnalysis.Threading;
 
 internal static class ProducerConsumer<TItem>
 {
@@ -248,7 +229,7 @@ private static Task<TResult> RunParallelChannelAsync<TSource, TArgs, TResult>(
             // We're running in parallel, so we def have multiple writers
             ProducerConsumerOptions.SingleReaderOptions,
             produceItems: static (callback, args, cancellationToken) =>
-                RoslynParallel.ForEachAsync(
+                Parallel.ForEachAsync(
                     args.source,
                     cancellationToken,
                     async (source, cancellationToken) =>
@@ -310,7 +291,7 @@ public static IAsyncEnumerable<TItem> RunParallelStreamAsync<TSource, TArgs>(
     {
         return RunAsync(
             static (callback, args, cancellationToken) =>
-                RoslynParallel.ForEachAsync(
+                Parallel.ForEachAsync(
                     args.source, cancellationToken,
                     async (source, cancellationToken) => await args.produceItems(
                         source, callback, args.args, cancellationToken).ConfigureAwait(false)),

src/Dependencies/Threading/ProducerConsumerOptions.cs

@@ -0,0 +1,34 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+#nullable enable
+
+using System.Threading.Channels;
+
+namespace Microsoft.CodeAnalysis.Threading;
+
+internal readonly record struct ProducerConsumerOptions
+{
+    /// <summary>
+    /// Used when the consumeItems routine will only pull items on a single thread (never concurrently). produceItems
+    /// can be called concurrently on many threads.
+    /// </summary>
+    public static readonly ProducerConsumerOptions SingleReaderOptions = new() { SingleReader = true };
+
+    /// <summary>
+    /// Used when the consumeItems routine will only pull items on a single thread (never concurrently). produceItems
+    /// can be called on a single thread as well (never concurrently).
+    /// </summary>
+    public static readonly ProducerConsumerOptions SingleReaderWriterOptions = new() { SingleReader = true, SingleWriter = true };
+
+#if NET
+    /// <inheritdoc cref="ChannelOptions.SingleWriter"/>
+#endif
+    public bool SingleWriter { get; init; }
+
+#if NET
+    /// <inheritdoc cref="ChannelOptions.SingleReader"/>
+#endif
+    public bool SingleReader { get; init; }
+}

src/EditorFeatures/Core/ExternalAccess/VSTypeScript/VSTypeScriptFindUsagesContext.cs

@@ -27,7 +27,7 @@ public ValueTask OnDefinitionFoundAsync(VSTypeScriptDefinitionItem definition, C
         => UnderlyingObject.OnDefinitionFoundAsync(definition.UnderlyingObject, cancellationToken);
 
     public ValueTask OnReferenceFoundAsync(VSTypeScriptSourceReferenceItem reference, CancellationToken cancellationToken)
-        => UnderlyingObject.OnReferencesFoundAsync(IAsyncEnumerableExtensions.SingletonAsync(reference.UnderlyingObject), cancellationToken);
+        => UnderlyingObject.OnReferencesFoundAsync(AsyncEnumerableFactory.SingletonAsync(reference.UnderlyingObject), cancellationToken);
 
     public ValueTask OnCompletedAsync(CancellationToken cancellationToken)
         => UnderlyingObject.OnCompletedAsync(cancellationToken);

src/EditorFeatures/Core/ExternalAccess/VSTypeScript/VSTypeScriptFindUsagesService.cs

@@ -46,7 +46,7 @@ public ValueTask OnDefinitionFoundAsync(VSTypeScriptDefinitionItem definition, C
             => _context.OnDefinitionFoundAsync(definition.UnderlyingObject, cancellationToken);
 
         public ValueTask OnReferenceFoundAsync(VSTypeScriptSourceReferenceItem reference, CancellationToken cancellationToken)
-            => _context.OnReferencesFoundAsync(IAsyncEnumerableExtensions.SingletonAsync(reference.UnderlyingObject), cancellationToken);
+            => _context.OnReferencesFoundAsync(AsyncEnumerableFactory.SingletonAsync(reference.UnderlyingObject), cancellationToken);
 
         public ValueTask OnCompletedAsync(CancellationToken cancellationToken)
             => ValueTaskFactory.CompletedTask;