Created
January 22, 2014 07:59
-
-
Save Acapla/8555049 to your computer and use it in GitHub Desktop.
simple awaitable queue
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
public class AwaitableQueue<T> | |
{ | |
private readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>(); | |
private long _count = 0; | |
private readonly ConcurrentQueue<TaskCompletionSource<T>> _pending = new ConcurrentQueue<TaskCompletionSource<T>>(); | |
/// <summary> | |
/// Enqueue the specified item. Blocking operation. | |
/// </summary> | |
/// <param name="item">Item.</param> | |
public async void Enqueue(T item) | |
{ | |
var count = Interlocked.Increment(ref _count); | |
if (count <= 0) | |
{ | |
Console.WriteLine("already wait"); | |
TaskCompletionSource<T> t; | |
while (!_pending.TryDequeue(out t)) | |
{ | |
await Task.Yield(); | |
} | |
t.SetResultAsync(item); | |
} | |
else | |
{ | |
Console.WriteLine("no wait"); | |
_queue.Enqueue(item); | |
} | |
} | |
public async Task<T> DequeueAsync() | |
{ | |
var count = Interlocked.Decrement(ref _count); | |
if (count < 0) | |
{ | |
Console.WriteLine("empty"); | |
var t = new TaskCompletionSource<T>(); | |
_pending.Enqueue(t); | |
return t.Task.IsCompleted ? t.Task.Result : await t.Task.ConfigureAwait(false); | |
} | |
else | |
{ | |
Console.WriteLine("not empty"); | |
T result; | |
while (!_queue.TryDequeue(out result)) | |
{ | |
await Task.Yield(); | |
} | |
return result; | |
} | |
} | |
public bool TryDequeue(out T t) | |
{ | |
return _queue.TryDequeue(out t); | |
} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
看起来很完美,要挑毛病的话,就是那个Task.Yield(),不过考虑到while循环的次数不会太多,也没有什么问题了。