am-wd/common
1
0
Fork 0
Code Actions Activity

Added AsyncQueue

Browse Source
This commit is contained in:
Andreas Müller
2022-08-13 15:22:49 +02:00
parent d233bf6b0a
commit 6e6e9d975a
3 changed files with 795 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

437
AMWD.Common/Utilities/AsyncQueue.cs Normal file
View File

@@ -0,0 +1,437 @@
using System.Diagnostics.CodeAnalysis;
using System.Threading;
using System.Threading.Tasks;
namespace System.Collections.Generic
{
/// <summary>
/// Represents a first-in, first-out collection of objects.
/// </summary>
/// <typeparam name="T">Specifies the type of elements in the queue.</typeparam>
public class AsyncQueue<T>
{
#region Fields
private readonly Queue<T> queue;
private TaskCompletionSource<bool> dequeueTcs = new();
private TaskCompletionSource<bool> availableTcs = new();
#endregion Fields
#region Constructors
/// <summary>
/// Initializes a new instance of the <see cref="AsyncQueue{T}"/> class that is empty and has the default initial capacity.
/// </summary>
[ExcludeFromCodeCoverage]
public AsyncQueue()
{
queue = new Queue<T>();
}
/// <summary>
/// Initializes a new instance of the System.Collections.Generic.Queue`1 class that
/// contains elements copied from the specified collection and has sufficient capacity
/// to accommodate the number of elements copied.
/// </summary>
/// <param name="collection">The collection whose elements are copied to the new <see cref="AsyncQueue{T}"/>.</param>
/// <exception cref="ArgumentNullException"><paramref name="collection"/> is null.</exception>
[ExcludeFromCodeCoverage]
public AsyncQueue(IEnumerable<T> collection)
{
queue = new Queue<T>();
Enqueue(collection);
}
/// <summary>
/// Initializes a new instance of the <see cref="AsyncQueue{T}"/> class that is empty and has the specified initial capacity.
/// </summary>
/// <param name="capacity">The initial number of elements that the <see cref="AsyncQueue{T}"/> can contain.</param>
/// <exception cref="ArgumentOutOfRangeException"><paramref name="capacity"/> is less than zero.</exception>
[ExcludeFromCodeCoverage]
public AsyncQueue(int capacity)
{
queue = new Queue<T>(capacity);
}
#endregion Constructors
#region Properties
/// <summary>
/// Gets the number of elements contained in the <see cref="AsyncQueue{T}"/>.
/// </summary>
[ExcludeFromCodeCoverage]
public int Count
{
get
{
lock (queue)
{
return queue.Count;
}
}
}
#endregion Properties
#region Queue implementation
/// <summary>
/// Removes all objects from the <see cref="AsyncQueue{T}"/>.
/// </summary>
[ExcludeFromCodeCoverage]
public void Clear()
{
lock (queue)
{
queue.Clear();
}
}
/// <summary>
/// Determines whether an element is in the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="item">The object to locate in the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
/// <returns>true if item is found in the <see cref="AsyncQueue{T}"/>; otherwise, false.</returns>
[ExcludeFromCodeCoverage]
public bool Contains(T item)
{
lock (queue)
{
return queue.Contains(item);
}
}
/// <summary>
/// Copies the <see cref="AsyncQueue{T}"/> elements to an existing one-dimensional <see cref="Array"/>, starting at the specified array index.
/// </summary>
/// <param name="array">The one-dimensional <see cref="Array"/> that is the destination of the elements copied from <see cref="AsyncQueue{T}"/>. The <see cref="Array"/> must have zero-based indexing.</param>
/// <param name="arrayIndex">The zero-based index in array at which copying begins.</param>
/// <exception cref="ArgumentNullException"><paramref name="array"/> is null.</exception>
/// <exception cref="ArgumentOutOfRangeException"><paramref name="arrayIndex"/> is less than zero.</exception>
/// <exception cref="ArgumentException">The number of elements in the source <see cref="AsyncQueue{T}"/> is greater than the available space from <paramref name="arrayIndex"/> to the end of the destination array.</exception>
[ExcludeFromCodeCoverage]
public void CopyTo(T[] array, int arrayIndex)
{
lock (queue)
{
queue.CopyTo(array, arrayIndex);
}
}
/// <summary>
/// Removes and returns the object at the beginning of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <returns>The object that is removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
/// <exception cref="InvalidOperationException">The <see cref="AsyncQueue{T}"/> is empty.</exception>
[ExcludeFromCodeCoverage]
public T Dequeue()
{
lock (queue)
{
return queue.Dequeue();
}
}
/// <summary>
/// Adds an object to the end of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="item">The object to add to the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
public void Enqueue(T item)
{
lock (queue)
{
queue.Enqueue(item);
SetToken(dequeueTcs);
SetToken(availableTcs);
}
}
/// <summary>
/// Returns the object at the beginning of the <see cref="AsyncQueue{T}"/> without removing it.
/// </summary>
/// <returns>The object at the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
/// <exception cref="InvalidOperationException">The <see cref="AsyncQueue{T}"/> is empty.</exception>
[ExcludeFromCodeCoverage]
public T Peek()
{
lock (queue)
{
return queue.Peek();
}
}
/// <summary>
/// Copies the <see cref="AsyncQueue{T}"/> elements to a new array.
/// </summary>
/// <returns>A new array containing elements copied from the <see cref="AsyncQueue{T}"/>.</returns>
[ExcludeFromCodeCoverage]
public T[] ToArray()
{
lock (queue)
{
return queue.ToArray();
}
}
/// <summary>
/// Sets the capacity to the actual number of elements in the <see cref="AsyncQueue{T}"/>, if that number is less than 90 percent of current capacity.
/// </summary>
[ExcludeFromCodeCoverage]
public void TrimExcess()
{
lock (queue)
{
queue.TrimExcess();
}
}
#endregion Queue implementation
#region Async implementation
/// <summary>
/// Removes and returns all available objects at the beginning of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="maxCount">The maximum number of objects to return. Zero means no limit.</param>
/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
/// <returns>The objects that are removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
public async Task<T[]> DequeueAvailableAsync(int maxCount = 0, CancellationToken cancellationToken = default)
{
while (true)
{
TaskCompletionSource<bool> internalDequeueTcs;
lock (queue)
{
if (queue.Count > 0)
{
int count = queue.Count;
if (maxCount > 0 && count > maxCount)
count = maxCount;
var items = new T[count];
for (int i = 0; i < count; i++)
items[i] = queue.Dequeue();
return items;
}
internalDequeueTcs = ResetToken(ref dequeueTcs);
}
await WaitAsync(internalDequeueTcs, cancellationToken);
}
}
/// <summary>
/// Removes and returns objects at the beginning of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="count">The number of objects to return.</param>
/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
/// <returns>The objects that are removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
public async Task<T[]> DequeueManyAsync(int count, CancellationToken cancellationToken = default)
{
if (count < 0)
throw new ArgumentOutOfRangeException(nameof(count));
while (true)
{
TaskCompletionSource<bool> internalDequeueTcs;
lock (queue)
{
if (count <= queue.Count)
{
var items = new T[count];
for (int i = 0; i < count; i++)
items[i] = queue.Dequeue();
return items;
}
internalDequeueTcs = ResetToken(ref dequeueTcs);
}
await WaitAsync(internalDequeueTcs, cancellationToken);
}
}
/// <summary>
/// Removes and returns the object at the beginning of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
/// <returns>The object that is removed from the beginning of the <see cref="AsyncQueue{T}"/>.</returns>
public async Task<T> DequeueAsync(CancellationToken cancellationToken = default)
{
while (true)
{
TaskCompletionSource<bool> internalDequeueTcs;
lock (queue)
{
if (queue.Count > 0)
return queue.Dequeue();
internalDequeueTcs = ResetToken(ref dequeueTcs);
}
await WaitAsync(internalDequeueTcs, cancellationToken);
}
}
/// <summary>
/// Waits asynchonously until at least one object is available in the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="cancellationToken">A cancellation token used to propagate notification that this operation should be canceled.</param>
/// <returns>An awaitable task.</returns>
public async Task WaitAsync(CancellationToken cancellationToken = default)
{
TaskCompletionSource<bool> internalAvailableTcs;
lock (queue)
{
if (queue.Count > 0)
return;
internalAvailableTcs = ResetToken(ref availableTcs);
}
await WaitAsync(internalAvailableTcs, cancellationToken);
}
#endregion Async implementation
#region Additional features
/// <summary>
/// Removes the object at the beginning of the <see cref="AsyncQueue{T}"/>, and copies it to the <paramref name="result"/> parameter.
/// </summary>
/// <param name="result">The removed object.</param>
/// <returns>true if the object is successfully removed; false if the <see cref="AsyncQueue{T}"/> is empty.</returns>
public bool TryDequeue(out T result)
{
try
{
result = Dequeue();
return true;
}
catch
{
result = default;
return false;
}
}
/// <summary>
/// Returns a value that indicates whether there is an object at the beginning of
/// the <see cref="AsyncQueue{T}"/>, and if one is present, copies it to the
/// <paramref name="result"/> parameter. The object is not removed from the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="result">If present, the object at the beginning of the <see cref="AsyncQueue{T}"/>; otherwise, the default value of <typeparamref name="T"/>.</param>
/// <returns>true if there is an object at the beginning of the <see cref="AsyncQueue{T}"/>; false if the <see cref="AsyncQueue{T}"/> is empty.</returns>
public bool TryPeek(out T result)
{
try
{
result = Peek();
return true;
}
catch
{
result = default;
return false;
}
}
/// <summary>
/// Removes the first occurrence of a specific object from the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="item">The object to remove from the <see cref="AsyncQueue{T}"/>. The value can be null for reference types.</param>
/// <returns>true if item is successfully removed; otherwise, false. This method also returns false if item was not found in the <see cref="AsyncQueue{T}"/>.</returns>
public bool Remove(T item)
{
lock (queue)
{
var copy = new Queue<T>(queue);
queue.Clear();
bool found = false;
int count = copy.Count;
for (int i = 0; i < count; i++)
{
var element = copy.Dequeue();
if (found)
{
queue.Enqueue(element);
continue;
}
if ((element == null && item == null) || element?.Equals(item) == true)
{
found = true;
continue;
}
queue.Enqueue(element);
}
return found;
}
}
/// <summary>
/// Adds objects to the end of the <see cref="AsyncQueue{T}"/>.
/// </summary>
/// <param name="collection">The objects to add to the <see cref="AsyncQueue{T}"/>.</param>
public void Enqueue(IEnumerable<T> collection)
{
lock (queue)
{
bool hasElements = false;
foreach (var element in collection)
{
hasElements = true;
queue.Enqueue(element);
}
if (hasElements)
{
SetToken(dequeueTcs);
SetToken(availableTcs);
}
}
}
#endregion Additional features
#region Helper
[ExcludeFromCodeCoverage]
private static void SetToken(TaskCompletionSource<bool> tcs)
{
tcs.TrySetResult(true);
}
[ExcludeFromCodeCoverage]
private static TaskCompletionSource<bool> ResetToken(ref TaskCompletionSource<bool> tcs)
{
if (tcs.Task.IsCompleted)
{
tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
}
return tcs;
}
[ExcludeFromCodeCoverage]
private static async Task WaitAsync(TaskCompletionSource<bool> tcs, CancellationToken cancellationToken)
{
if (await Task.WhenAny(tcs.Task, Task.Delay(-1, cancellationToken)) == tcs.Task)
{
await tcs.Task;
return;
}
cancellationToken.ThrowIfCancellationRequested();
}
#endregion Helper
}
}

4
CHANGELOG.md
View File

@@ -5,7 +5,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased](https://git.am-wd.de/AM.WD/common/compare/v1.8.1...master) - 0000-00-00
_nothing changed yet_
### Added
- `AsyncQueue<T>`
## [v1.8.1](https://git.am-wd.de/AM.WD/common/compare/v1.8.0...v1.8.1) - 2022-08-07

355
UnitTests/Common/Utilities/AsyncQueueTests.cs Normal file
View File

@@ -0,0 +1,355 @@
using System;
using System.Collections.Generic;
using System.Reflection;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Moq;
namespace UnitTests.Common.Utilities
{
[TestClass]
[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
public class AsyncQueueTests
{
private Queue<TestElement> internalQueue;
private TestElement queueElement1;
private TestElement queueElement2;
private TestElement queueElement3;
[TestInitialize]
public void InitializeTest()
{
queueElement1 = new TestElement
{
Number = 111,
Text = "one"
};
queueElement2 = new TestElement
{
Number = 222,
Text = "two"
};
queueElement3 = new TestElement
{
Number = 333,
Text = "three"
};
internalQueue = new Queue<TestElement>();
internalQueue.Enqueue(queueElement1);
internalQueue.Enqueue(queueElement2);
internalQueue.Enqueue(queueElement3);
}
[TestMethod]
public void ShouldEnqueueItem()
{
// arrange
var element = new TestElement { Number = 1, Text = "Hello" };
internalQueue.Clear();
var queue = GetQueue();
// act
queue.Enqueue(element);
// assert
Assert.AreEqual(1, internalQueue.Count);
Assert.AreEqual(internalQueue.Count, queue.Count);
}
[TestMethod]
public void ShouldEnqueueItemAndResetAvailableToken()
{
// arrange
var element = new TestElement { Number = 1, Text = "Hello" };
bool available = false;
internalQueue.Clear();
var queue = GetQueue();
// act
var task = Task.Run(async () =>
{
await queue.WaitAsync();
available = true;
});
queue.Enqueue(element);
task.Wait();
// assert
Assert.IsTrue(available);
}
[TestMethod]
public void ShouldEnqueueItemAndResetDequeueToken()
{
// arrange
var element = new TestElement { Number = 1, Text = "Hello" };
TestElement callback = null;
internalQueue.Clear();
var queue = GetQueue();
// act
var task = Task.Run(async () =>
{
callback = await queue.DequeueAsync();
});
queue.Enqueue(element);
task.Wait();
// assert
Assert.IsNotNull(callback);
Assert.AreEqual(element, callback);
}
[TestMethod]
public void ShouldEnqueueMultipleItems()
{
// arrange
var elements = new TestElement[]
{
new TestElement { Number = 1, Text = "Hello" },
new TestElement { Number = 2, Text = "World" },
};
internalQueue.Clear();
var queue = GetQueue();
// act
queue.Enqueue(elements);
// assert
Assert.AreEqual(2, internalQueue.Count);
Assert.AreEqual(queue.Count, internalQueue.Count);
}
[TestMethod]
public void ShouldPeekAValue()
{
// arrange
var queue = GetQueue();
// act
bool isSuccess = queue.TryPeek(out var item);
// assert
Assert.IsTrue(isSuccess);
Assert.IsNotNull(item);
Assert.AreEqual(queueElement1, item);
Assert.AreEqual(3, queue.Count);
}
[TestMethod]
public void ShouldNotPeekAValue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
// act
bool isSuccess = queue.TryPeek(out var item);
// assert
Assert.IsFalse(isSuccess);
Assert.IsNull(item);
}
[TestMethod]
public void ShouldDequeueAValue()
{
// arrange
var queue = GetQueue();
// act
bool isSuccess = queue.TryDequeue(out var item);
// assert
Assert.IsTrue(isSuccess);
Assert.IsNotNull(item);
Assert.AreEqual(queueElement1, item);
Assert.AreEqual(2, queue.Count);
}
[TestMethod]
public void ShouldNotDequeueAValue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
// act
bool isSuccess = queue.TryDequeue(out var item);
// assert
Assert.IsFalse(isSuccess);
Assert.IsNull(item);
}
[TestMethod]
public void ShouldRemoveAValue()
{
// arrange
var queue = GetQueue();
// act
queue.Remove(queueElement2);
var item1 = queue.Dequeue();
var item2 = queue.Dequeue();
// assert
Assert.AreEqual(0, queue.Count);
Assert.AreEqual(queueElement1, item1);
Assert.AreEqual(queueElement3, item2);
}
[TestMethod]
public void ShouldNotRemoveAValue()
{
// arrange
var queue = GetQueue();
// act
queue.Remove(null);
var item1 = queue.Dequeue();
var item2 = queue.Dequeue();
var item3 = queue.Dequeue();
// assert
Assert.AreEqual(0, queue.Count);
Assert.AreEqual(queueElement1, item1);
Assert.AreEqual(queueElement2, item2);
Assert.AreEqual(queueElement3, item3);
}
[TestMethod]
public async Task ShouldAwaitOneDequeue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
var task = Task.Run(async () =>
{
await Task.Delay(1000);
queue.Enqueue(new[] { queueElement1, queueElement2, queueElement3 });
});
// act
var item = await queue.DequeueAsync();
// assert
Assert.AreEqual(2, queue.Count);
Assert.IsNotNull(item);
Assert.AreEqual(queueElement1, item);
}
[TestMethod]
public async Task ShouldAwaitManyDequeue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
var task = Task.Run(async () =>
{
await Task.Delay(1000);
queue.Enqueue(new[] { queueElement1, queueElement2, queueElement3 });
});
// act
var items = await queue.DequeueManyAsync(2);
// assert
Assert.AreEqual(1, queue.Count);
Assert.IsNotNull(items);
Assert.AreEqual(2, items.Length);
Assert.AreEqual(queueElement1, items[0]);
Assert.AreEqual(queueElement2, items[1]);
}
[TestMethod]
public async Task ShouldAwaitAllDequeue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
var task = Task.Run(async () =>
{
await Task.Delay(1000);
queue.Enqueue(new[] { queueElement1, queueElement2, queueElement3 });
});
// act
var items = await queue.DequeueAvailableAsync();
// assert
Assert.AreEqual(0, queue.Count);
Assert.IsNotNull(items);
Assert.AreEqual(3, items.Length);
Assert.AreEqual(queueElement1, items[0]);
Assert.AreEqual(queueElement2, items[1]);
Assert.AreEqual(queueElement3, items[2]);
}
[TestMethod]
public async Task ShouldAwaitAvailableDequeue()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
var task = Task.Run(async () =>
{
await Task.Delay(1000);
queue.Enqueue(new[] { queueElement1, queueElement2, queueElement3 });
});
// act
var items = await queue.DequeueAvailableAsync(2);
// assert
Assert.AreEqual(1, queue.Count);
Assert.IsNotNull(items);
Assert.AreEqual(2, items.Length);
Assert.AreEqual(queueElement1, items[0]);
Assert.AreEqual(queueElement2, items[1]);
}
[TestMethod]
[ExpectedException(typeof(ArgumentOutOfRangeException))]
public async Task ShouldThrowArumentOutOfRangeException()
{
// arrange
internalQueue.Clear();
var queue = GetQueue();
// act
await queue.DequeueManyAsync(-2);
// assert - ArgumentOutOfRangeException expected
Assert.Fail();
}
private AsyncQueue<TestElement> GetQueue()
{
var asyncQueue = new AsyncQueue<TestElement>();
var field = asyncQueue.GetType().GetField("queue", BindingFlags.Instance | BindingFlags.NonPublic);
field.SetValue(asyncQueue, internalQueue);
return asyncQueue;
}
private class TestElement
{
public int Number { get; set; }
public string Text { get; set; }
}
}
}