Dedup UnboundedChannel and UnboundedPriorityChannel

src/libraries/System.Threading.Channels/src/System.Threading.Channels.csproj

@@ -25,6 +25,7 @@ System.Threading.Channel&lt;T&gt;</PackageDescription>
     <Compile Include="System\Threading\Channels\Channel_1.cs" />
     <Compile Include="System\Threading\Channels\Channel_2.cs" />
     <Compile Include="System\Threading\Channels\IDebugEnumerator.cs" />
+    <Compile Include="System\Threading\Channels\IUnboundedChannelQueue.cs" />
     <Compile Include="System\Threading\Channels\SingleConsumerUnboundedChannel.cs" />
     <Compile Include="System\Threading\Channels\UnboundedChannel.cs" />
     <Compile Include="$(CommonPath)Internal\Padding.cs" Link="Common\Internal\Padding.cs" />
@@ -44,7 +45,6 @@ System.Threading.Channel&lt;T&gt;</PackageDescription>
     <Compile Include="System\Threading\Channels\AsyncOperation.netcoreapp.cs" />
     <Compile Include="System\Threading\Channels\Channel.netcoreapp.cs" />
     <Compile Include="System\Threading\Channels\ChannelOptions.netcoreapp.cs" />
-    <Compile Include="System\Threading\Channels\UnboundedPriorityChannel.cs" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == '$(NetCoreAppCurrent)'">

src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel.cs

@@ -1,6 +1,10 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+
 namespace System.Threading.Channels
 {
     /// <summary>Provides static methods for creating channels.</summary>
@@ -9,7 +13,7 @@ public static partial class Channel
         /// <summary>Creates an unbounded channel usable by any number of readers and writers concurrently.</summary>
         /// <returns>The created channel.</returns>
         public static Channel<T> CreateUnbounded<T>() =>
-            new UnboundedChannel<T>(runContinuationsAsynchronously: true);
+            new UnboundedChannel<T, UnboundedChannelConcurrentQueue<T>>(new(new()), runContinuationsAsynchronously: true);
 
         /// <summary>Creates an unbounded channel subject to the provided options.</summary>
         /// <typeparam name="T">Specifies the type of data in the channel.</typeparam>
@@ -27,7 +31,7 @@ public static Channel<T> CreateUnbounded<T>(UnboundedChannelOptions options)
                 return new SingleConsumerUnboundedChannel<T>(!options.AllowSynchronousContinuations);
             }
 
-            return new UnboundedChannel<T>(!options.AllowSynchronousContinuations);
+            return new UnboundedChannel<T, UnboundedChannelConcurrentQueue<T>>(new(new()), !options.AllowSynchronousContinuations);
         }
 
         /// <summary>Creates a channel with the specified maximum capacity.</summary>
@@ -71,5 +75,32 @@ public static Channel<T> CreateBounded<T>(BoundedChannelOptions options, Action<
 
             return new BoundedChannel<T>(options.Capacity, options.FullMode, !options.AllowSynchronousContinuations, itemDropped);
         }
+
+        /// <summary>Provides an <see cref="IUnboundedChannelQueue{T}"/> for a <see cref="ConcurrentQueue{T}"/>.</summary>
+        private readonly struct UnboundedChannelConcurrentQueue<T>(ConcurrentQueue<T> queue) : IUnboundedChannelQueue<T>
+        {
+            private readonly ConcurrentQueue<T> _queue = queue;
+
+            /// <inheritdoc/>
+            public bool IsThreadSafe => true;
+
+            /// <inheritdoc/>
+            public void Enqueue(T item) => _queue.Enqueue(item);
+
+            /// <inheritdoc/>
+            public bool TryDequeue([MaybeNullWhen(false)] out T item) => _queue.TryDequeue(out item);
+
+            /// <inheritdoc/>
+            public bool TryPeek([MaybeNullWhen(false)] out T item) => _queue.TryPeek(out item);
+
+            /// <inheritdoc/>
+            public int Count => _queue.Count;
+
+            /// <inheritdoc/>
+            public bool IsEmpty => _queue.IsEmpty;
+
+            /// <inheritdoc/>
+            public IEnumerator<T> GetEnumerator() => _queue.GetEnumerator();
+        }
     }
 }

src/libraries/System.Threading.Channels/src/System/Threading/Channels/Channel.netcoreapp.cs

@@ -2,20 +2,22 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 
 namespace System.Threading.Channels
 {
     /// <summary>Provides static methods for creating channels.</summary>
     public static partial class Channel
     {
         /// <summary>Creates an unbounded prioritized channel usable by any number of readers and writers concurrently.</summary>
+        /// <typeparam name="T">Specifies the type of data in the channel.</typeparam>
         /// <returns>The created channel.</returns>
         /// <remarks>
         /// <see cref="Comparer{T}.Default"/> is used to determine priority of elements.
         /// The next item read from the channel will be the element available in the channel with the lowest priority value.
         /// </remarks>
         public static Channel<T> CreateUnboundedPrioritized<T>() =>
-            new UnboundedPrioritizedChannel<T>(runContinuationsAsynchronously: true, comparer: null);
+            new UnboundedChannel<T, UnboundedChannelPriorityQueue<T>>(new(new()), runContinuationsAsynchronously: true);
 
         /// <summary>Creates an unbounded prioritized channel subject to the provided options.</summary>
         /// <typeparam name="T">Specifies the type of data in the channel.</typeparam>
@@ -30,7 +32,45 @@ public static Channel<T> CreateUnboundedPrioritized<T>(UnboundedPrioritizedChann
         {
             ArgumentNullException.ThrowIfNull(options);
 
-            return new UnboundedPrioritizedChannel<T>(!options.AllowSynchronousContinuations, options.Comparer);
+            return new UnboundedChannel<T, UnboundedChannelPriorityQueue<T>>(new(new(options.Comparer)), !options.AllowSynchronousContinuations);
+        }
+
+        /// <summary>Provides an <see cref="IUnboundedChannelQueue{T}"/> for a <see cref="PriorityQueue{TElement, TPriority}"/>.</summary>
+        private readonly struct UnboundedChannelPriorityQueue<T>(PriorityQueue<bool, T> queue) : IUnboundedChannelQueue<T>
+        {
+            private readonly PriorityQueue<bool, T> _queue = queue;
+
+            /// <inheritdoc/>
+            public bool IsThreadSafe => false;
+
+            /// <inheritdoc/>
+            public void Enqueue(T item) => _queue.Enqueue(true, item);
+
+            /// <inheritdoc/>
+            public bool TryDequeue([MaybeNullWhen(false)] out T item) => _queue.TryDequeue(out _, out item);
+
+            /// <inheritdoc/>
+            public bool TryPeek([MaybeNullWhen(false)] out T item) => _queue.TryPeek(out _, out item);
+
+            /// <inheritdoc/>
+            public int Count => _queue.Count;
+
+            /// <inheritdoc/>
+            public bool IsEmpty => _queue.Count == 0;
+
+            /// <inheritdoc/>
+            public IEnumerator<T> GetEnumerator()
+            {
+                List<T> list = [];
+                foreach ((bool _, T Priority) item in _queue.UnorderedItems)
+                {
+                    list.Add(item.Priority);
+                }
+
+                list.Sort(_queue.Comparer);
+
+                return list.GetEnumerator();
+            }
         }
     }
 }

src/libraries/System.Threading.Channels/src/System/Threading/Channels/IDebugEnumerator.cs

@@ -11,7 +11,7 @@ internal interface IDebugEnumerable<T>
         IEnumerator<T> GetEnumerator();
     }
 
-    internal sealed class DebugEnumeratorDebugView<T>
+    internal class DebugEnumeratorDebugView<T>
     {
         public DebugEnumeratorDebugView(IDebugEnumerable<T> enumerable)
         {
@@ -26,4 +26,6 @@ public DebugEnumeratorDebugView(IDebugEnumerable<T> enumerable)
         [DebuggerBrowsable(DebuggerBrowsableState.RootHidden)]
         public T[] Items { get; }
     }
+
+    internal sealed class DebugEnumeratorDebugView<T, TOther>(IDebugEnumerable<T> enumerable) : DebugEnumeratorDebugView<T>(enumerable);
 }

src/libraries/System.Threading.Channels/src/System/Threading/Channels/IUnboundedChannelQueue.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.
+
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+
+namespace System.Threading.Channels
+{
+    /// <summary>Representation of the queue data structure used by <see cref="UnboundedChannel{T, TQueue}"/>.</summary>
+    internal interface IUnboundedChannelQueue<T> : IDebugEnumerable<T>
+    {
+        /// <summary>Gets whether the other members are safe to use concurrently with each other and themselves.</summary>
+        bool IsThreadSafe { get; }
+
+        /// <summary>Enqueues an item into the queue.</summary>
+        /// <param name="item">The item to enqueue.</param>
+        void Enqueue(T item);
+
+        /// <summary>Dequeues an item from the queue, if possible.</summary>
+        /// <param name="item">The dequeued item, or default if the queue was empty.</param>
+        /// <returns>Whether an item was dequeued.</returns>
+        bool TryDequeue([MaybeNullWhen(false)] out T item);
+
+        /// <summary>Peeks at the next item from the queue that would be dequeued, if possible.</summary>
+        /// <param name="item">The peeked item, or default if the queue was empty.</param>
+        /// <returns>Whether an item was peeked.</returns>
+        bool TryPeek([MaybeNullWhen(false)] out T item);
+
+        /// <summary>Gets the number of elements in the queue.</summary>
+        int Count { get; }
+
+        /// <summary>Gets whether the queue is empty.</summary>
+        bool IsEmpty { get; }
+    }
+}

src/libraries/System.Threading.Channels/src/System/Threading/Channels/UnboundedChannel.cs

@@ -5,19 +5,20 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
+using System.Runtime.CompilerServices;
 using System.Threading.Tasks;
 
 namespace System.Threading.Channels
 {
     /// <summary>Provides a buffered channel of unbounded capacity.</summary>
     [DebuggerDisplay("Items = {ItemsCountForDebugger}, Closed = {ChannelIsClosedForDebugger}")]
-    [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<>))]
-    internal sealed class UnboundedChannel<T> : Channel<T>, IDebugEnumerable<T>
+    [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<,>))]
+    internal sealed class UnboundedChannel<T, TQueue> : Channel<T>, IDebugEnumerable<T> where TQueue : struct, IUnboundedChannelQueue<T>
     {
         /// <summary>Task that indicates the channel has completed.</summary>
         private readonly TaskCompletionSource _completion;
         /// <summary>The items in the channel.</summary>
-        private readonly ConcurrentQueue<T> _items = new ConcurrentQueue<T>();
+        private readonly TQueue _items;
         /// <summary>Readers blocked reading from the channel.</summary>
         private readonly Deque<AsyncOperation<T>> _blockedReaders = new Deque<AsyncOperation<T>>();
         /// <summary>Whether to force continuations to be executed asynchronously from producer writes.</summary>
@@ -29,23 +30,24 @@ internal sealed class UnboundedChannel<T> : Channel<T>, IDebugEnumerable<T>
         private Exception? _doneWriting;
 
         /// <summary>Initialize the channel.</summary>
-        internal UnboundedChannel(bool runContinuationsAsynchronously)
+        internal UnboundedChannel(TQueue items, bool runContinuationsAsynchronously)
         {
+            _items = items;
             _runContinuationsAsynchronously = runContinuationsAsynchronously;
             _completion = new TaskCompletionSource(runContinuationsAsynchronously ? TaskCreationOptions.RunContinuationsAsynchronously : TaskCreationOptions.None);
             Reader = new UnboundedChannelReader(this);
             Writer = new UnboundedChannelWriter(this);
         }
 
         [DebuggerDisplay("Items = {Count}")]
-        [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<>))]
+        [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<,>))]
         private sealed class UnboundedChannelReader : ChannelReader<T>, IDebugEnumerable<T>
         {
-            internal readonly UnboundedChannel<T> _parent;
+            internal readonly UnboundedChannel<T, TQueue> _parent;
             private readonly AsyncOperation<T> _readerSingleton;
             private readonly AsyncOperation<bool> _waiterSingleton;
 
-            internal UnboundedChannelReader(UnboundedChannel<T> parent)
+            internal UnboundedChannelReader(UnboundedChannel<T, TQueue> parent)
             {
                 _parent = parent;
                 _readerSingleton = new AsyncOperation<T>(parent._runContinuationsAsynchronously, pooled: true);
@@ -68,8 +70,8 @@ public override ValueTask<T> ReadAsync(CancellationToken cancellationToken)
                 }
 
                 // Dequeue an item if we can.
-                UnboundedChannel<T> parent = _parent;
-                if (parent._items.TryDequeue(out T? item))
+                UnboundedChannel<T, TQueue> parent = _parent;
+                if (parent._items.IsThreadSafe && parent._items.TryDequeue(out T? item))
                 {
                     CompleteIfDone(parent);
                     return new ValueTask<T>(item);
@@ -112,24 +114,60 @@ public override ValueTask<T> ReadAsync(CancellationToken cancellationToken)
 
             public override bool TryRead([MaybeNullWhen(false)] out T item)
             {
-                UnboundedChannel<T> parent = _parent;
+                UnboundedChannel<T, TQueue> parent = _parent;
+                return parent._items.IsThreadSafe ?
+                    LockFree(parent, out item) :
+                    Locked(parent, out item);
 
-                // Dequeue an item if we can
-                if (parent._items.TryDequeue(out item))
+                static bool LockFree(UnboundedChannel<T, TQueue> parent, [MaybeNullWhen(false)] out T item)
                 {
-                    CompleteIfDone(parent);
-                    return true;
+                    if (parent._items.TryDequeue(out item))
+                    {
+                        CompleteIfDone(parent);
+                        return true;
+                    }
+
+                    item = default;
+                    return false;
                 }
 
-                item = default;
-                return false;
+                static bool Locked(UnboundedChannel<T, TQueue> parent, [MaybeNullWhen(false)] out T item)
+                {
+                    lock (parent.SyncObj)
+                    {
+                        if (parent._items.TryDequeue(out item))
+                        {
+                            CompleteIfDone(parent);
+                            return true;
+                        }
+                    }
+
+                    item = default;
+                    return false;
+                }
             }
 
-            public override bool TryPeek([MaybeNullWhen(false)] out T item) =>
-                _parent._items.TryPeek(out item);
+            public override bool TryPeek([MaybeNullWhen(false)] out T item)
+            {
+                UnboundedChannel<T, TQueue> parent = _parent;
+                return parent._items.IsThreadSafe ?
+                    parent._items.TryPeek(out item) :
+                    Locked(parent, out item);
+
+                // Separated out to keep the try/finally from preventing TryPeek from being inlined
+                static bool Locked(UnboundedChannel<T, TQueue> parent, [MaybeNullWhen(false)] out T item)
+                {
+                    lock (parent.SyncObj)
+                    {
+                        return parent._items.TryPeek(out item);
+                    }
+                }
+            }
 
-            private static void CompleteIfDone(UnboundedChannel<T> parent)
+            private static void CompleteIfDone(UnboundedChannel<T, TQueue> parent)
             {
+                Debug.Assert(parent._items.IsThreadSafe || Monitor.IsEntered(parent.SyncObj));
+
                 if (parent._doneWriting != null && parent._items.IsEmpty)
                 {
                     // If we've now emptied the items queue and we're not getting any more, complete.
@@ -144,12 +182,12 @@ public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationTo
                     return new ValueTask<bool>(Task.FromCanceled<bool>(cancellationToken));
                 }
 
-                if (!_parent._items.IsEmpty)
+                if (_parent._items.IsThreadSafe && !_parent._items.IsEmpty)
                 {
                     return new ValueTask<bool>(true);
                 }
 
-                UnboundedChannel<T> parent = _parent;
+                UnboundedChannel<T, TQueue> parent = _parent;
 
                 lock (parent.SyncObj)
                 {
@@ -192,15 +230,15 @@ public override ValueTask<bool> WaitToReadAsync(CancellationToken cancellationTo
         }
 
         [DebuggerDisplay("Items = {ItemsCountForDebugger}")]
-        [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<>))]
+        [DebuggerTypeProxy(typeof(DebugEnumeratorDebugView<,>))]
         private sealed class UnboundedChannelWriter : ChannelWriter<T>, IDebugEnumerable<T>
         {
-            internal readonly UnboundedChannel<T> _parent;
-            internal UnboundedChannelWriter(UnboundedChannel<T> parent) => _parent = parent;
+            internal readonly UnboundedChannel<T, TQueue> _parent;
+            internal UnboundedChannelWriter(UnboundedChannel<T, TQueue> parent) => _parent = parent;
 
             public override bool TryComplete(Exception? error)
             {
-                UnboundedChannel<T> parent = _parent;
+                UnboundedChannel<T, TQueue> parent = _parent;
                 bool completeTask;
 
                 lock (parent.SyncObj)
@@ -240,7 +278,7 @@ public override bool TryComplete(Exception? error)
 
             public override bool TryWrite(T item)
             {
-                UnboundedChannel<T> parent = _parent;
+                UnboundedChannel<T, TQueue> parent = _parent;
                 while (true)
                 {
                     AsyncOperation<T>? blockedReader = null;
@@ -321,7 +359,7 @@ public override ValueTask WriteAsync(T item, CancellationToken cancellationToken
         }
 
         /// <summary>Gets the object used to synchronize access to all state on this instance.</summary>
-        private object SyncObj => _items;
+        private object SyncObj => _blockedReaders;
 
         [Conditional("DEBUG")]
         private void AssertInvariants()