RateLimitedThreading.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace rate_limited_threadpool
{
    class Program
    {
        static void Main(string[] args)
        {
            //make a garbage list, simulates our input
            List<int> SomeList = new List<int>();
            for (int x = 0; x < 101; x++)
            {
                SomeList.Add(x);
            }

            //user variables to set
            TimeSpan MaxTime = new TimeSpan(0, 0, 3); //some time in hour,min,seconds
            var MaxTasksInTimeFrame = 10; //the maximum number of tasks to complete in 3 seconds
            var MaxConcurrentThreads = 5; //how many threads you want to run at the same time            
            //end user variables

            //the output collection
            BlockingCollection<int> bc = new BlockingCollection<int>();
            //ManualResetEvents are the events which signal a thread is done to the thread pool
            ManualResetEvent[] manualEvents = new ManualResetEvent[MaxTasksInTimeFrame];
            
            while (SomeList.Count > 0)
            {
                var startTime = DateTime.Now;
                ThreadPool.SetMaxThreads(MaxConcurrentThreads, MaxConcurrentThreads);
                for (int x = 0; x < MaxTasksInTimeFrame; x++)
                {
                    //surround in try/catch to handle when we have an empty list                    
                    try
                    {
                        //pop and element out of the list
                        var element = SomeList[0];
                        SomeList.RemoveAt(0);
                        //set the manual reset event for this thread to false (unsignaled)
                        manualEvents[x] = new ManualResetEvent(false);
                        //create a new state which is made up of our input (x), the reset event (manualEvent), and the output collection
                        State state = new State()
                        {
                            x = element,
                            manualEvent = manualEvents[x],
                            bc = bc
                        };
                        //queue the thread to start, calling the DoSomething method with 'state' as an input
                        ThreadPool.QueueUserWorkItem(new WaitCallback(DoSomething), state);
                    }
                    //catch empty lists
                    catch(System.ArgumentOutOfRangeException e)
                    {
                        break;
                    }                   
                }
                //wait for all the threads in the queue to finish
                WaitHandle.WaitAll(manualEvents);
                //if the elapsedTime is less than our rate limit, sleep for the remainder
                var elapsedTime = DateTime.Now - startTime;
                if(elapsedTime < MaxTime)
                {
                    Console.WriteLine("sleping for " + (MaxTime - elapsedTime).TotalSeconds.ToString());
                    Thread.Sleep(MaxTime - elapsedTime);
                }
            }
            //print our output!
            foreach(var item in bc)
            {
                Console.WriteLine(item);
            }
            //we're done
            Console.WriteLine("DONE!");
            Console.ReadLine();
        }


        public static void DoSomething(object input)
        {
            //cast input to a state class
            var state = (State)input;
            //write x in our thread
            Console.WriteLine(state.x);
            //add x to the output collection
            state.bc.Add(state.x);
            //sleep for 100ms to simulate more work
            Thread.Sleep(100);
            Console.WriteLine(String.Format("Thread {0} finishing", state.x));
            //signal that we're done doing our thing, allowing the threadpool to start a new thread
            state.manualEvent.Set();            
        }
    }

    class State
    {
        public int x { get; set; }
        public ManualResetEvent manualEvent { get; set; }
        public BlockingCollection<int> bc { get; set; }
    }
}