How should you use C# 5's async to represent a sequence of asynchronous tasks? For example, if we wanted to download numbered files from a server and return each one as we get it, how can we implement a method such as this?
public async IEnumerable<File> DownloadPictures() {
const string format = "http://example.com/files/{0}.png";
for (int i = 0; i++; ) {
yield return await DownloadFile(string.Format(format, i));
}
}
A true sequence doesn't work well directly with async/await, because tasks only return a single value. You need an actual enumerable type, such as IAsyncEnumerator<T> in Ix-Async (or AsyncEx). The design of IAsyncEnumerator<T> is described in this Channel9 video.
It seems to me you want something very similar to BlockingCollection<T>, that uses Tasks and awaiting instead of blocking.
Specifically, something that you can add to without blocking or waiting. But when you try to remove an item when none is available at the moment, you can await until some item is available.
The public interface could look like this:
public class AsyncQueue<T>
{
public bool IsCompleted { get; }
public Task<T> DequeueAsync();
public void Enqueue(T item);
public void FinishAdding();
}
FinishAdding() is necessary, so that we know when to end dequeuing.
With this, your code could look like this (m_queue is AsyncQueue<File>):
var tasks = Enumerable.Range(0, 10)
.Select(i => DownloadAndEnqueue(i))
.ToArray();
Task.WhenAll(tasks).ContinueWith(t => m_queue.FinishAdding());
…
static async Task DownloadAndEnqueue(string url)
{
m_queue.Enqueue(await DownloadFile(url));
}
It's not as nice as what you imagined could work, but it should work.
And the implementation of AsyncQueue<T>? There are two queues. One is for completed work, that hasn't been dequeued yet. The other is for Tasks (actually, TaskCompletionSource<T>) that were already dequeued, but that don't have any result yet.
When you dequeue and there is some completed work in the queue, just return work from there (using Task.FromResult()). If the queue is empty, create new Task, add it to the other queue and return it.
When you enqueue some completed work and there are some Tasks in the queue, remove one and finish it using the result we have now. If the Task queue is empty, add the work to the first queue.
With this, you can dequeue and enqueue as many times as you want, and it will work correctly. When you know there won't be any new work, call FinishAdding(). If there are any waiting Tasks, they will throw an exception.
In other words:
public class AsyncQueue<T>
{
private readonly object m_lock = new object();
private bool m_finishedAdding = false;
private readonly Queue<T> m_overflowQueue = new Queue<T>();
private readonly Queue<TaskCompletionSource<T>> m_underflowQueue =
new Queue<TaskCompletionSource<T>>();
public bool IsCompleted
{
get { return m_finishedAdding && m_overflowQueue.Count == 0; }
}
public Task<T> DequeueAsync()
{
Task<T> result;
lock (m_lock)
{
if (m_overflowQueue.Count > 0)
result = Task.FromResult(m_overflowQueue.Dequeue());
else if (!m_finishedAdding)
{
var tcs = new TaskCompletionSource<T>();
m_underflowQueue.Enqueue(tcs);
result = tcs.Task;
}
else
throw new InvalidOperationException();
}
return result;
}
public void Enqueue(T item)
{
lock (m_lock)
{
if (m_finishedAdding)
throw new InvalidOperationException();
if (m_underflowQueue.Count > 0)
{
var tcs = m_underflowQueue.Dequeue();
tcs.SetResult(item);
}
else
m_overflowQueue.Enqueue(item);
}
}
public void FinishAdding()
{
lock (m_lock)
{
m_finishedAdding = true;
while (m_underflowQueue.Count > 0)
{
var tcs = m_underflowQueue.Dequeue();
tcs.SetException(new InvalidOperationException());
}
}
}
}
If you wanted to limit size of the work queue (and thus limiting producers, if they are too fast), you could make Enqueue() return Task too, which would require another queue.
I know it's been a while, but I've written something to closely emulate "yield return" for async enumerables here. No complicated code needed.
You use it like:
public IAsyncEnumerable<File> DownloadPictures() {
const string format = "http://example.com/files/{0}.png";
return AsyncEnumerable.Create(async y =>
{
for (int i = 0; i++; ) {
await y.YieldReturn(await DownloadFile(string.Format(format, i)));
}
};
}
I normally avoid advertising my own code here but this is a clearly needed feature in C# 6.0, so I hope you find it useful in C# 5.0 if you're still stuck on this.
If you only had a finite number of URLs, you could do this:
public async Task<IEnumerable<File>> DownloadPictures()
{
const string format = "http://example.com/files/{0}.png";
var urls = Enumerable.Range(0, 999).Select(i => String.Format(format, i));
var tasks = urls.Select(u => DownloadFile(u));
var results = Task.WhenAll(tasks);
return await results;
}
The key is to get a list of Tasks and then call Task.WhenAll on that list.
The benefit of async is the calling method can invoke several blocking operations in parallel and only block once the returned value is needed. This same capability is possible in this scenario with yield/return by using return type IEnumerable<Task>.
public IEnumerable<Task<File>> DownloadPictures() {
const string format = "http://example.com/files/{0}.png";
for (int i = 0; i++; ) {
yield return DownloadFileAsync(string.Format(format, i));
}
}
In a similar fashion to async/await the calling method can now continue to execute until it needs the next value at which point await/.Result can be called on the next task. The following extension method demonstrates this:
public static IEnumerable<T> Results<T>(this IEnumerable<Task<T>> tasks)
{
foreach (var task in tasks)
yield return task.Result;
}
If the calling method would like to ensure that all the IEnumerable Tasks are created and running in parallel then an extension method such as the following may be beneficial (this, and the aforementioned method, are likely already in a standard lib):
public static IEnumerable<T> ResultsParallel<T>(this IEnumerable<Task<T>> tasks)
{
foreach (var task in tasks.ToArray())
yield return task.Result;
}
Notice how the responsibility for the concern of what runs in parallel is transferred to the calling method just as it was with async/await. In the event that there is concern regarding the creation of the Tasks blocking, an extension method such as the following could be created:
public static Task<IEnumerable<T>> ResultsAsync<T>(this IEnumerable<Task<T>> tasks)
{
var startedTasks = new ConcurrentQueue<Task<T>>();
var writerTask = new Task(() =>
{
foreach (var task in tasks)
{
startedTasks.Enqueue(task);
}
});
writerTask.Start();
var readerTask = new Task<IEnumerable<T>>(() =>
{
return ResultsSequential(startedTasks, () => writerTask.IsCompleted);
});
readerTask.Start();
return readerTask;
}
private static IEnumerable<T> ResultsSequential<T>(ConcurrentQueue<Task<T>> tasks, Func<bool> isDone)
{
while (true)
{
Task<T> task;
if (isDone.Invoke())
{
if (tasks.TryDequeue(out task))
{
yield return task.Result;
}
else
{
yield break;
}
} else if (tasks.TryDequeue(out task))
{
yield return task.Result;
}
}
}
This implementation is not very efficient. An efficient implementation is too large to fit in the margin.
