Solving Problems With Asynchrony #4

BatchedLoader.cs

protected IEnumerator<object> TextureLoaderTask () {
    var sleep = new Sleep(0.5); // Wait half a second for work items to build up.
    var batch = new List<PendingTextureLoad>();

    while (true) {
        // Pull a single work item from the queue. Our task will be put to sleep if
        //  the queue is empty.
        {
            var f = TextureLoadQueue.Dequeue();
            yield return f;
            batch.Add(f.Result);
        }

        // We were woken back up by a work item. Put our task back to sleep for a
        //  fixed period of time to allow more work items to build up.
        yield return sleep;

        // Now that we've slept for a while, more items may be in the queue. Attempt
        //  to pull up to 31 more items into the batch.
        TextureLoadQueue.DequeueMultiple(batch, 31);

        // Since the contents of the batch can be processed in any order, we need to
        //  wait until the callback has finished running as a whole, so we use a
        //  helper method to do that instead of waiting for the individual futures
        //  in the batch.
        yield return Future.RunInThread(
            () => LoadTextureBatch(batch)
        );

        // Make sure to empty out the batch before filling it with items in the next
        //  iteration.
        batch.Clear();
    }
}

// We've changed the callback to operate on a batch of work items instead of a single
//  work item.
protected static void LoadTextureBatch (List<PendingTextureLoad> batch) {
    // Sort the work items by their position on disk.
    batch.Sort((lhs, rhs) => {
        GetDiskLocation(lhs.Filename).CompareTo(
            GetDiskLocation(rhs.Filename)
        );
    });

    // Now walk through the batch in sorted order and load the textures.
    foreach (var ptl in batch) {
        try {
            var texture = new Texture(ptl.Filename);
            ptl.Future.SetResult(texture, null);
        } catch (Exception e) {
            ptl.Future.SetResult(null, e);
        }
    }
}

BlockingQueue.cs

public class BlockingQueue<T> {
    private readonly object _Lock = new object();
    private readonly Queue<Future<T>> _WaitingFutures = new Queue<Future<T>>();
    private readonly Queue<T> _Queue = new Queue<T>();

    public int Count {
        get {
            lock (_Lock) {
                return _Queue.Count - _WaitingFutures.Count;
            }
        }
    }

    public int DequeueMultiple (IList<T> output, int maximum) {
        for (int i = 0; i < maximum; i++) {
            lock (_Lock) {
                if (_Queue.Count == 0)
                    return i;

                output.Add(_Queue.Dequeue());
            }
        }

        return maximum;
    }

    public T[] DequeueAll () {
        lock (_Lock) {
            T[] result = new T[_Queue.Count];
            _Queue.CopyTo(result, 0);
            _Queue.Clear();
            return result;
        }
    }

    public Future<T> Dequeue () {
        var f = new Future<T>();
        lock (_Lock) {
            if (_Queue.Count > 0)
                f.Complete(_Queue.Dequeue());
            else
                _WaitingFutures.Enqueue(f);
        }
        return f;
    }

    public void Enqueue (T value) {
        Future<T> wf;

        while (true)
        lock (_Lock) {
            if (_WaitingFutures.Count > 0) {
                wf = _WaitingFutures.Dequeue();

                try {
                    wf.Complete(value);
                    if (wf.Completed)
                        return;
                } catch (FutureDisposedException) {
                }
            } else {
                _Queue.Enqueue(value);
                return;
            }
        }
    }
}

SequentialLoader.cs

public class PendingTextureLoad {
    public readonly Future<Texture> Future = new Future<Texture>();
    public string Filename;
}

protected readonly BlockingQueue<PendingTextureLoad> TextureLoadQueue = new BlockingQueue<PendingTextureLoad>();

protected IEnumerator<object> TextureLoaderTask () {
    PendingTextureLoad item;

    while (true) {
        {
            var f = TextureLoadQueue.Dequeue();
            yield return f;
            item = f.Result;
        }

        ThreadPool.QueueUserWorkItem(LoadSingleTexture, item);

        // Wait for the callback to finish loading the texture before we proceed to the next work
        //  item from the queue.
        yield return item.Future;
    }
}

// A callback to be run from the threadpool that loads our texture from disk.
// Implicitly we're assuming here that we don't need any synchronization, because we will only
//  ever be running one of these at a time. In practice, we probably will need some kind of
//  synchronization around the Texture constructor, depending on the API we're using.
protected static void LoadSingleTexture (object _) {
    var ptl = (PendingTextureLoad)_;
    try {
        var texture = new Texture(ptl.Filename);
        ptl.Future.SetResult(texture, null);
    } catch (Exception e) {
        ptl.Future.SetResult(null, e);
    }
}

public Future<Texture> LoadTexture (string filename) {
    var ptl = new PendingTextureLoad {
        Filename = filename
    };

    TextureLoadQueue.Enqueue(ptl);
    return ptl.Future;
}