ThrottledQueue<T>

    /// <summary>
    /// Throttles the processing of a collection of items, based on how many threads are specified
    /// Substitute for when PLinq AsParallel() doesn't work for whatever reasons
    /// <example>
    /// new ThrottledQueue<int>(2,Enumerable.Range(1,100),(i) => Console.WriteLine(i + " " + Thread.CurrentThread.ManagedThreadId))
    /// .Wait();
    /// </example>
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class ThrottledQueue<T>:IDisposable {

        readonly ConcurrentQueue<T> _queue;
        readonly int _numberOfThreads;
        readonly CountdownEvent _waitCountdown;
        readonly CountdownEvent _startCountdown;
        readonly ManualResetEvent _startEvent;
        volatile bool _cancelled;
        Action<T> _processAction;

        public int NumberOfThreads {
            get { return _numberOfThreads; }
        }

        public bool IsCancelled {
            get { return _cancelled; }
        }

        public void Cancel() {
            if(_cancelled)
                throw new InvalidOperationException("Already cancelled");
            _cancelled = true;
        }

        public void Wait() {
            _waitCountdown.Wait();
        }

        public ThrottledQueue(int numberOfThreads,IEnumerable<T> collection,Action<T> processAction) {
            //TODO: Argument

            _numberOfThreads = numberOfThreads;
            _processAction = processAction;
            _queue = new ConcurrentQueue<T>(collection);
            _waitCountdown = new CountdownEvent(collection.Count());
            _startCountdown = new CountdownEvent(numberOfThreads);
            _startEvent = new ManualResetEvent(false);

            for(int i = 0;i < numberOfThreads;i++)
                ThreadPool.QueueUserWorkItem(RunningWorker,i);

            //wait until all threads are started
            _startCountdown.Wait();

            //after all threads are started then signal them to start processing the queue
            _startEvent.Set();
        }

        void RunningWorker(object id) {

            int workerid = (int)id;

            Console.WriteLine(string.Format("Worker #{0:N0} started",workerid));

            _startCountdown.Signal();
            _startEvent.WaitOne();

            while(!_cancelled) {
                T item;
                if(_queue.TryDequeue(out item)) {
                    bool quit = false;
                    try {
                        _processAction(item);
                    }
                    catch { }
                    finally {
                        if(_waitCountdown.Signal())
                            quit = true;
                    }
                    if(quit)
                        return;
                }
                else
                    Thread.Sleep(10);
            }
        }

        public void Dispose() {
            if(!IsCancelled)
                Cancel();
        }
    }