source: branches/PersistenceSpeedUp/HeuristicLab.Persistence/3.3/Default/Xml/AsyncBuffer.cs @ 7654

using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;

namespace HeuristicLab.Persistence.Default.Xml {

  public class AsyncBuffer<T> : IEnumerable<T>, IEnumerable {

    private IEnumerator<T> it;

    private Queue<T> queue;
    private object queueLock;
    private AutoResetEvent elementAdded;
    private ManualResetEvent done;

    private Thread thread;

    private Exception exception;

    private class LockIndicator {
      public bool HasLock;
      public LockIndicator() { HasLock = false; }
    }

    public AsyncBuffer(IEnumerable<T> source) {
      it = source.GetEnumerator();
      queue = new Queue<T>();
      queueLock = new object();
      elementAdded = new AutoResetEvent(false);
      done = new ManualResetEvent(false);
      thread = new Thread(Run);
      thread.Start();
    }

    private void Run() {
      bool hasLock = false;
      try {
        while (it.MoveNext()) {
          T value = it.Current;
          lock (queueLock) {
            queue.Enqueue(value);
            elementAdded.Set();
          }
        }
      } catch (Exception x) {
        exception = x;
      } finally {
        if (hasLock)
          Monitor.Exit(queueLock);
      }
      done.Set();
    }

    private IEnumerator<T> RealGetEnumerator(LockIndicator li) {
      Monitor.Enter(queueLock, ref li.HasLock);
      while (!done.WaitOne(0) || queue.Count > 0) {
        while (queue.Count == 0 && !done.WaitOne(0)) {
          li.HasLock = false;
          Monitor.Exit(queueLock);
          AutoResetEvent.WaitAny(new WaitHandle[] { done, elementAdded });
          Monitor.Enter(queueLock, ref li.HasLock);
        }
        if (queue.Count > 0) {
          T element = queue.Dequeue();
          li.HasLock = false;
          Monitor.Exit(queueLock);
          yield return element;
          Monitor.Enter(queueLock, ref li.HasLock);
        }
      }
      li.HasLock = false;
      Monitor.Exit(queueLock);
      if (exception != null)
        throw exception;
    }

    #region IEnumerable<T> Members
    public IEnumerator<T> GetEnumerator() {
      LockIndicator li = new LockIndicator();
      try {
        return RealGetEnumerator(li);
      } catch {
        if (li.HasLock)
          Monitor.Exit(queueLock);
        throw;
      }
    }
    #endregion

    #region IEnumerable Members
    IEnumerator IEnumerable.GetEnumerator() {
      return GetEnumerator();
    }
    #endregion
  }
}