I made a simple-ish thread pool in C#. You can use it freely.
It basically maintains a number of threads and re-uses them. It uses a very simple circular array as a job queue. See GetNewJob(), AssignJob() and ExecuteJobs(). Read the notes at the top how to use it.
The Start() function contains one line to initialize the pool and one like to run a simple test. The test will max out your cpu for about 10 seconds or so.
It’s able to check periodically if all jobs are done… which is not real efficient or punctual, but it’s sort of the only way since I can’t really rely on Join() to wait for threads to end… they might end on their own beforehand so calling Join() on a closed thread hangs it. So it has to poll (in milliseconds increments) to see if jobs are done.
To assign functions you need to create a function which takes an object. The object will be cast from a JobData class. GetNewJob() gets access to a JobData instance to modify, into which you can set your function and any parameter data as fields. This is passed into the thread. Then the function will be called and the JobData will be passed to it, which it can then use to get parameter data into the function. Feel free to expand the JobData fields as needed (for all functions). You can assign multiple jobs before executing, and can stop the threads from running the new jobs, using the ThreadsCanSeeNewJobs() function (false to stop them, true to make them aware). You could theoretically keep adding jobs (until queue is full) while other jobs are running and either let the threads discover the jobs or, if they ended and didn’t see the jobs, signal them to go check by calling ExecuteJobs() again.
Let me know if you spot any bugs. I’m pretty happy with it so far. Props to the C# crowd on the web for the main principles and some basic sourcecode.
I’ve designed it to exclude WEBGL since that platform doesn’t support threads. If you target some other platform with no threading you might want to set the appropriate compiler directives. It should run single-threaded when there is no threading support (untested).
[edit]… I found a couple of bugs and it wasn’t keeping track of the oldest active job properly, nor was it reporting the amount of free space properly, so I had to recode some bits and add a few new functions. New version is below.[/edit]
using System.Collections;
using System.Collections.Generic;
using UnityEngine;
#if !UNITY_WEBGL
using System.Threading;
#endif
namespace TurboCharger{
public class ThreadPool : MonoBehaviour {
//Threads will search for jobs automatically, and if there are none, or if LookForNewJobs==false, they will go to sleep. They will not wake up until an ExecuteJobs() is called from somewhere.
//LookForNewJubs must == true for new jobs to be visible to the threads, and you also must ExecuteJobs() at some point unless you are assuming that threads will see your jobs when they are done working.
//Its possibly they will finish jobs before your new jobs are assigned and visible, so they may not necessarily run them automatically. If in doubt, call ExecuteJobs() after assigning work and making jobs visible.
//Note also that regardless of the sequence in which you assign jobs, although they begin running in exact sequence due to the fact that each thread has to lock access to the threadpool to get the 'next' job,
//how much cpu time is given to each thread or how the thread interacts with other parts of the o/s may vary, e.g. calls to Debug.Log() might print 'out of sequence' messages. Generally speaking
//jobs will execute in approximate order, with some random jitter. It also depends on the work to be done and how the o/s assigns cpu time.
//You can at least be certain that more recently added 'new jobs' will absolutely NOT be executed until older jobs are completed, and at least one thread has no older jobs left to run
//Basic usage:
//1) Upon Start(), threadool will be auto-initialized. Otherwise call InitializeThreadPool() with number of threads and max amount of jobs that can be queued at one time
//2) Call GetNewJob() to get reference to 'empty' JobData object (which will contain old data) which you can edit the fields of
//3) Modify fields of JobData object, store the function you want to be run, and any parameter data
//4) Call AssignJob() to 'lock in' the job and make the queue advance to the next empty job. Optionally execute the job now (which will also execute any other pending jobs).
//NB: After you call AssignJob(), any other references you have to the JobData object should be considered obsolete. Do not use them. If you try to re-use it later, you will break the threadpool. Always use GetNewJob()
//5) Optionally you can assign multiple jobs before running them. First do GetNewJob(), edit the job, AssignJob(false), for each of the jobs.... then at the end make one call to ExecuteJobs()
//NB: Threads may constantly be on the lookout for new jobs except when there are no jobs left, at which point they go to sleep and will only be woken to look for work by AssignJob(true) or ExecuteJobs()
//But if threads are already awake/working on older jobs, and you AssignJob(false), the job may instantly or soon be automatically executed because threads look for new work when they finish a job.
//If you prefer, you can stop threads looking for NEW jobs, with ThreadsSeeNewJobs(false); They will continue running existing jobs till done then go to sleep. You can then GetNewJob(), edit, and AssignJob()
//for several jobs, and then finally ThreadsSeeNewJobs(true) to allow them to be seen, and ExecuteJobs() to signal them to wake up and check for jobs.
#if !UNITY_WEBGL
public Thread[] Threads; //Array of actual threads
#endif
public int ThreadCount=0; //Maximum/total number of threads in the threadpool
public JobData[] Jobs; //Circular Array as a queue of temporary job data to be used by threadpool when it claims ownership of a job, containing the function to execute, variables/parameters to pass etc. Allows us to add new jobs onto the virtual 'end' while keeping job data in-place until it's done running, so that threads can pass the JobData class instance to the function being run without having to copy field data
public int MaximumJobs=0; //How many jobs can be stored in the Jobs array
public int JobsWaiting=0; //How many jobs are waiting to be consumed by the pool
public int JobsRunning=0; //How many jobs are currently still running
public int NextEmptyJobIndex=0; //Index of the next empty slot in the Jobs array, where we can set up a new job. Since Jobs is circular, the index wraps around to 0 when =Length
public int NextJobToRun=0; //Index of the next job that's set up ready to be run by a thread
public int OldestJob=0; //Index of the oldest job, in order to keep track of empty space and know where the space ends
public int JobsAvailable=0; //How many jobs are available for assigning new tasks
public bool LookForNewJobs=true; //Whether threads should be allowed to get new jobs. Until this is true, new jobs cannot be run, but running jobs will continue.
public int Default_MaximumPoolSize=16; //Default number of real threads in the theadpool - a good size is AT LEAST the number of CPU cores, or that number x2. Make sure to account for 'hyperthreading cores' as well which run 2 threads each (e.g. quad core reports as 8 cores, so try 16 threads). If the thread is stalling or waiting a lot or not using a lot of cpu time, you might want more threads to keep the pool busy.
public int Default_MaximumJobs=4096; //Default size of job queue - can't create more jobs than this at once. GetNewJob() will return null if there are none free. Or see GetJobSpace() and WaitForJobSpace().
public bool Default_LookForNewJobs=true; //Default for whether to look for new jobs or not
readonly object ThreadLocker = new object(); //A lock to use to isolate shared code/data in each thread and to give us the ability to send a signal to each thread to check for jobs
public delegate void Job(object obj); //Define the function pointer format for the job function - accepts a JobData object cast as an object
void Start(){
//Startup initialization
Debug.Log("Starting");
InitializeThreadPool(Default_MaximumPoolSize,Default_MaximumJobs); //Create a threadpool if there isn't one set up yet
Test();
}
public void Test(){
//Experiment to test if the threadpool is working
int index;
for (index=0;index<1000;index++){ //Add 1000 jobs
WaitForJobSpace(1,0); //Make sure we have room, in case the queue size is smaller than the number of jobs we're going to add. Don't need this if you know there is enough room for the number of jobs you're adding. You could just do a WaitForJobSpace(1000) outside the loop
JobData MyJob = GetNewJob(); //Get a new job to edit
MyJob.JobToRun = TestFunction; //The function we want to run on the threadpool
MyJob.Text1 = "Hello World"; //The parameter for the function
AssignJob(true); //Assign it and start executing it right now while we're still adding jobs - could just wait till all jobs are assigned but if queue size is < number of jobs being added, we will be stuck
}
ExecuteJobs(); //Run the jobs
//FinishAllJobs(1); //Finish all jobs and wait for them to be done
//Debug.Log("All Jobs are Done");
CloseThreadPool(); //Finish jobs and then close the pool
Debug.Log("Pool is closed");
}
public void TestFunction(object obj){
//Test function to print a string parameter and do 10 million calculations
JobData MyJob = (JobData)obj;
Debug.Log(MyJob.Text1+" Job: "+MyJob.JobIndex); //Say "Hello World"
int index;
int total=0;
for (index=0;index<1000000;index++){
total++; //Do work
}
}
public ThreadPool(){
//Constructor which will get called if someone does like: ThreadPool myPool = new ThreadPool();
//With no parameters, a default-size threadpool will be created
InitializeThreadPool(Default_MaximumPoolSize,Default_MaximumJobs); //Create a threadpool
}
public ThreadPool(int PoolSize){
//Constructor which will get called if someone does like: ThreadPool myPool = new ThreadPool(16);
//PoolSize parameter defines how many threads to allocate
//Default size of job queue will be created
InitializeThreadPool(PoolSize,Default_MaximumJobs); //Create a pool
}
public ThreadPool(int PoolSize, int MaxJobs){
//Constructor which will get called if someone does like: ThreadPool myPool = new ThreadPool(16,1024);
//PoolSize parameter defines how many threads to allocate
//MaxJobs defines what size job queue to create - maximum number of simultaneous job entries
InitializeThreadPool(PoolSize,MaxJobs); //Create a threadpool if there isn't one set up yet
}
~ThreadPool(){
//Destructor called automatically when object is being destroyed - you cannot call this manually
//To shut down the threadpool manaully call CloseThreadPool()
CloseThreadPool(); //Shut down the threadpool if it hasn't been done already. This will only finish jobs/close threads if Threads[] is != null
}
public void InitializeThreadPool(int PoolSize,int MaxJobs){
//Initialize the threadpool to a given number of threads
//This will erase any existing jobs or threads, which is dangerous if threads are still running jobs
//PoolSize is how many threads to allocate
//MaxJobs is how many jobs can be queued at one time - make sure there is enough room in the circular array to store currently executing jobs (for as many threads as there are in the pool) + room for new jobs (and at least enough space for new jobs 1 per thread so that we can close all the threads)
//Create room for storing jobs
if (MaxJobs<PoolSize*2){
MaxJobs=PoolSize*2; //Make sure it's at least big enough to store 1 currently running job per thread + 1 yet-to-run job per thread so that threads can be closed - this won't necessarily guarantee that there's enough space since the active job zone of the queue can be fragmented with randoly finished jobs
}
Jobs=new JobData[MaxJobs]; //Room for data for each job
int index;
for (index=0;index<MaxJobs;index++){
Jobs[index]=new JobData(); //Create a re-useable job data object which we can store new jobs in, and the reals thread can read for job info and parameters to pass to their functions
Jobs[index].JobIndex=index; //Store the index of each job in case functions want to refer to it
}
JobsWaiting=0; //No jobs to run yet
JobsRunning=0; //How many jobs are currently running
NextEmptyJobIndex=0; //Start at the beginning
NextJobToRun=0; //Next job to run is currently at end of Jobs array because there are no jobs yet, but if we do that, then when we assign a job it will be job 0 and nextjobtorun will be the previous job, so we need to fake it to 0 here assuming there are no jobs running yet so that the thread will get the first job. The first call to AssignJob will put a job at 0 and not change 'nextjob' so the next job to run will be at 0
OldestJob=0; //The oldest job is at 0 because there are no jobs yet
JobsAvailable=MaxJobs; //All jobs are now available
MaximumJobs=MaxJobs; //Largest number of jobs that can be stored at once
LookForNewJobs=Default_LookForNewJobs; //Whether to allow threads to get new jobs
//Create real threads on platforms that support them
#if !UNITY_WEBGL
Threads = new Thread[PoolSize]; //Room to store real threads
for (index=0;index<PoolSize;index++){
Threads[index]=new Thread(JobProcessor); //Create a new real thread for the threadpool to use, each will run a JobProcessor function which will look for and process jobs
Threads[index].Start(this); //Start the thread running. It will look for jobs. We pass in 'this' threadpool class instance so that all JobProcessors can access and share the threadpool data
}
ThreadCount=PoolSize; //We now have this many threads waiting for work
#else
ThreadCount=0; //No threads supported!
#endif
}
public void ThreadsSeeNewJobs(bool SeeJobs){
//Whether the threads can see and theefore consume new jobs from the queue (at some point after they are added), or whether to ignore them
//Calling this with 'true' will allow the threads to start working on jobs, after having finished their current job or after a call to Execute() which wakes them up from sleeping
//Just because threads are allowed to see new jobs doesn't mean they will run them yet
//Call this with 'false' allows you to let threads continue running their current jobs, but then go to sleep while you load a batch of new jobs with GetNewJob() and AssignJob().
//When you are then done assigning jobs you can call ThreadsSeeNewJobs(true) followed by EcecuteJobs()
//This allows you to 'delay' the running of new jobs but not interrupt existing jobs
//Default is that new jobs are always seen automatically
#if !UNITY_WEBGL
lock (ThreadLocker){ //Get exclusive access
#endif
LookForNewJobs=SeeJobs; //Set it
#if !UNITY_WEBGL
}
#endif
}
public JobData GetNewJob(){
//Get the JobData object for a new job, so that you can modify the data to assign a function, set parameters etc
//This doesn't need to be locked because no threads will touch empty jobs
//Will return null if there are no empty jobs available, so check for ==null after calling
#if !UNITY_WEBGL
lock (ThreadLocker){ //Get exclusive access
#endif
if (JobsAvailable>0){ //Fail if the next job is still running or we've used all our jobs up
return Jobs[NextEmptyJobIndex]; //Return reference to the next job that can be assigned
} else {
return null; //No job available
}
#if !UNITY_WEBGL
}
#endif
}
public bool AssignJob(bool ExecuteNow=false){
//Now that the JobData has been filled in by the user, advance the list of jobs to the next job
//This does not execute the jobs unless ExecuteNow==true. Either threads will find the jobs automatically when they're done with their current jobs, or you will need to call ExecuteJobs()
//Return true or false whether it was possible to assign the job. Will only be false if there is no more job space left for this job to be assigned (all jobs are busy or queued)
#if !UNITY_WEBGL
lock (ThreadLocker){ //Get exclusive access
#endif
if (JobsAvailable==0){ //Is there room left? This was checked in GetNewJob() but just to be sure we don't overwrite an active job
return false; //Can't assign the job, no space left in the queue
}
JobsWaiting++; //One more job is now waiting
JobsAvailable--; //One less job available
NextEmptyJobIndex++; //Next job
if (NextEmptyJobIndex==MaximumJobs){
NextEmptyJobIndex=0; //Warp around circular queue Job array
}
#if !UNITY_WEBGL
}
#endif
if (ExecuteNow==true){
ExecuteJobs(); //Run the job now (and any other pending jobs) - signal threads to look for jobs
}
return true; //This job was successfully assigned and possibly executed
}
public void ExecuteJobs(){
//Trigger any waiting real threads to wake up and check for work, since we just loaded some new jobs into the Jobs array
//If threading is disabled, the jobs will be executed in sequence on the main thread before returning from this function, otherwise they will run in parallel
//Note that LookForNewJobs must ==true otherwise the awoken threads will not see any work to do -- see ThreadsSeeNewJobs(true);
#if !UNITY_WEBGL
//Run on real threads
lock (ThreadLocker){ //Obtain exclusive access to the threadpool data
Monitor.PulseAll(ThreadLocker); //Trigger suspended threads to check for jobs
}
#else
//Run on main thread only
if (JobsWaiting>0){
int index;
int jobindex=NextJobToRun; //Start with this job
JobData ThisJob; //Its data
int jobcount=JobsWaiting; //Need to record copy of this here because JobsWaiting will change in loop
for (index=0;index<jobcount;index++){ //Go through all open jobs
//Get the job
ThisJob = Jobs[NextJobToRun]; //Get the data object
ThisJob.JobRunning=true; //Job is running
JobsRunning++; //One more job is running
JobsWaiting--; //One less job to do
NextJobToRun++; //Next job
if (NextJobToRun==MaximumJobs){ //At end of queue
NextJobToRun=0; //Wrap around circular queue array
}
//Check if this job is a dummy job asking us to close a thread
if (ThisJob.JobToRun==null){
//Skip it; return; //ThreadPool asked us to close this thread so 'return' from the JobProcessor function, which means the thread is done processing, which will close the thread (for threadless version, don't 'return')
} else {
//Otherwise can execute
ThisJob.JobToRun(ThisJob); //Call the function and pass parameter data to it
//Finish up
ThisJob.JobRunning=false; //This job is done
JobsRunning--; //One less job is running (not necessarily sequentually organized)
if (ThisJob.JobIndex==OldestJob){ //Check if this job is the oldest job, because if so then we need to now scan to see how many jobs after us also have ended and move the OldestJob marker to the oldest unfinished job
//Find the oldest active job
int index=OldestJob; //Current oldest position in the circular queue
int maxjobs=MaximumJobs; //For speed
bool finished=false; //Not done yet
while (finished==false){
if (Jobs[index].JobRunning==false){
//This job is not running, go to the next one
JobsAvailable++; //One more job has become available
index++; //Next job to check
if (index==maxjobs){ //If past the right end of the queue
index=0; //Wrap to the left
}
if ((index==NextJobToRun) || (index==NextEmptyJobIndex)){ //Did we get to the next non-running job that's actually a new job that hasn't been run yet? Or are there no new jobs and we just arrived at the next non-running empty job?
finished=true; //We're done
OldestJob=index; //This 'new' non-running job is (or will be) the OldestJob
}
} else {
//This job is running, so its the new oldest
finished=true;
OldestJob=index; //This index is the oldest active job
}
}
}
}
}
}
#endif
}
public void FinishAllJobs(int MillisecsPerCheck=1){
//Execute all assigned jobs and then wait for them all to finish before returning
//MillisecsPerCheck is how many millisecs to wait between checks, 0 will cooperatively multitask but may check more often, otherwise checking every 1 millisec may waste time if jobs are done soon after a check
if (MillisecsPerCheck<0){
MillisecsPerCheck=0; //Demand positive milliseconds
}
ThreadsSeeNewJobs(true); //Make sure the jobs will be seen if user for some reason hid jobs and didn't execute them yet
ExecuteJobs(); //Run the jobs
WaitForJobSpace(MaximumJobs,MillisecsPerCheck); //Check every 1ms until there are absolutely no jobs left or running. We could wait until only the termination jobs are left (=ThreadCount).
}
public void WaitForJobSpace(int SpaceNeeded=1, int MillisecsPerCheck=1){
//Since we keep threads alive and re-use them, the notion of waiting for jobs to be done can't be checked by using Join() unless the threads are going to permanently close
//Therefore we need to periodically check how many jobs are left, but we don't want to use a lot of cpu time constantly checking
//SpaceNeeded is how many jobs should be available for assignment before we return. If =MaximumJobs it will wait for all jobs to finish, or <MaximumJobs it will wait until there are that many jobs available
//e.g. if you need room for 16 jobs, and you have a Jobs queue size of 100 jobs, then waiting for 100-16=84 so it will wait until there are only 84 jobs assigned, leaving room for 16 more
//MillisecsPerCheck is how many milliseconds to wait between checks. Note that a lot of work can be done in 1 millisecond, so this could potentially waste time where the main thread is waiting unnecessarily
//But provided the job queue has plenty of busy work for threads to perform, hopefully most of the processing time is taken doing actual work and only >1 millisecond wasted
//Note then that each time you call this with a MillisecsPerCheck=1, you could be wasting up to 1 full millsecond of cpu time if all the jobs are done already
//Minimum time between checks is 0 millisecond. If you wait for 0 milliseconds, multitasking will hand over cpu time to the threads without a delay, but this will likely mean more frequent checks
//being performed by the main thread, which will reduce performance. However, the checks are small so giving up cpu timeslice quickly may still allow lots of work to be done
//If you want to immediately check how many jobs are left to run, call GetJobsRunning(), or how much space there is for new jobs call GetJobSpace().
//Note if you call with a high MillisecsPerCheck, there will be no way to break out of this until the timeout has expired, freezing Unity, because it puts the main thread to sleep
//Remember that a 1 millisecond wait means 1000 checks per second, or rather ~16 checks per frame at 60hz. If you feel you are wasting up to 1 millsecond of valuable cpu time, use MillisecsPerCheck=0
if (MillisecsPerCheck<0){
MillisecsPerCheck=0; //Demand positive milliseconds
}
if (SpaceNeeded<0){
SpaceNeeded=0; //Demand positive space left
}
bool done=false;
while (done==false){
#if !UNITY_WEBGL
lock(ThreadLocker){ //Get exclusive access
#endif
if (JobsAvailable>=SpaceNeeded){ //Get current actual space available for jobs, is it enough?
done=true; //We're at our threshold, there is enough room for the number of needed new jobs, can return now
}
#if !UNITY_WEBGL
}
#endif
if (done==false){ //If there are still too many jobs left...
Thread.Sleep(MillisecsPerCheck); //Go to sleep for a while/cooperatively multitask, then check how much space is left
}
}
}
private int GetJobsRunning(){
//Return number of jobs that are currently still running. This isn't necessarily the same as the amount of space available for new jobs because there is fragmentation in the
//block of currently running/recently finished jobs.
//This only tells you how many jobs are actually being processed, not how much space is left for new jobs. For that, use GetJobSpace()
#if !UNITY_WEBGL
lock(ThreadLocker){ //Get exclusive access
#endif
return JobsRunning; //Return number of running jobs
#if !UNITY_WEBGL
}
#endif
}
private int GetJobSpace(){
//Return how much space is left for new jobs. Unlike WaitForJobSpace this does not wait for a certain amount of space to become available, it just tells you how much space there currently is
#if !UNITY_WEBGL
lock(ThreadLocker){ //Get exclusive access
#endif
return JobsAvailable; //Return number of available jobs
#if !UNITY_WEBGL
}
#endif
}
public void CloseThreadPool(){
//Signal all threads to shut down by setting up as many dummy jobs as there are threads, so that each thread will grab 1 job and then close
//Even if there are jobs left to run you can call this and it will allow all remaining jobs to execute, even ones which were not seen yet, and only then it will shut down the system
//Note that if you call this inside of Unity Editor, and the functions you're excecuting contain something that requires Unity to respond (e.g. Debug.Log()), and you enter
//into a Join() state with the threads to wait for them to end, this main thread will not be able to return control to Unity and it won't be able to process the editor updates,
//which may result in a deadlock, freeze or crash. Same would be true if your functions depend on something like this from plugins or other behaviors that might cause a deadlock.
//However, removal of Debug.Log() or anything else causing Unity to hang will allow this to work. Threads will process and work will be done to the end, then closed.
#if !UNITY_WEBGL
if (Threads != null){ //Only try to shut down jobs/threads if we still have real threads created
//There may still be jobs running. If there aren't enough free jobs, 1 per thread, to tell them to close, we will have to wait somehow until enough jobs are free
WaitForJobSpace(ThreadCount,1); //Check every 1ms until there are at least enough jobs left to finish
int index;
JobData MyJob;
for (index=0;index<ThreadCount;index++){
MyJob=GetNewJob(); //Get a new job to modify
MyJob.JobToRun=null; //Set job to have no function, which will signal to thread to close permanently
AssignJob(false); //Next job
}
//At this point there may still be jobs left to run besides the termination jobs, since we didn't wait for JobsLeft=0, so we need to wait here before doing Join()s
FinishAllJobs(1); //Execute all remaining jobs and wait for them to finish
//It'd be more efficient to do Join() on each thread to wait until they are all done, but its possible they may already have exited due to our null functions, and trying to Join() a thread that is already closed will make it hang
//for (index=0;index<ThreadCount;index++){
// if (Threads[index].IsAlive==true){
// Threads[index].Join(); //Join main thread's program flow to the thread, which basically means wait until the thread exits and closes before continuing, so we will wait for all threads to exit
// }
//}
}
ThreadCount=0; //No threads left
JobsWaiting=0; //No jobs left
JobsAvailable=0; //No jobs available
MaximumJobs=0; //No jobs
OldestJob=0; //Oldest is first
NextJobToRun=0; //Next job is 0
NextEmptyJobIndex=0; //Next new job is 0
Jobs=null; //Release the job array
Threads=null; //Release the threads
#endif
}
void JobProcessor(object obj){
//This function is run by each individual real thread and looks for jobs to do in the array of jobs
//An instance of the threadpool class itself should be passed in, cast as an object, so that we can get access to the threadpool's state/data and interact with the jobs
//Use Thread.CurrentThread.ManagedThreadId to get the ID of this thread if you want it, then maybe assign it to a spare parameter in the JobData to make it accessible to the job function
ThreadPool ThePool=(ThreadPool)obj; //Cast the object back to a threadpool. This is now one of many references to the pool, so we will have to establish a lock or mutex to get safe access
JobData ThisJob; //Its data
//Run forever, or until a 'null' job function is found which triggers the thread to permanently close
while (true){ //Continue processing forever until the exit state is reached - exit state is when we create a job with a 'StopThread' flag set to true
lock (ThreadLocker){ //Obtain an exclusive lock before running the enclosed code so no other threads can interfere
//Wait for a new job
while ((ThePool.JobsWaiting==0) || (ThePool.LookForNewJobs==false)){ //If there are no jobs waiting, or there are jobs waiting but we're not allowed to look for new jobs yet, then we'll put this thread on hold and wait to be signalled
Monitor.Wait(ThreadLocker); //No jobs available to consume, so cancel this lock temporarily and wait for a signal from the ThreadPool that there are new jobs available
}
//Get the job
ThisJob = ThePool.Jobs[ThePool.NextJobToRun]; //Get the data object
ThisJob.JobRunning=true; //Job is running
ThePool.JobsRunning++; //One more job is running
ThePool.JobsWaiting--; //One less job to do
ThePool.NextJobToRun++; //Next job
if (ThePool.NextJobToRun==ThePool.MaximumJobs){ //At end of queue
ThePool.NextJobToRun=0; //Wrap around circular queue array
}
}
//Execute the function
if (ThisJob.JobToRun != null){ //Don't run the function if it's a null/terminate function
ThisJob.JobToRun(ThisJob); //Call the function and pass parameter data to it
}
//Finish up
lock (ThreadLocker){
ThisJob.JobRunning=false; //This job is done
ThePool.JobsRunning--; //One less job is running (not necessarily sequentually organized)
if (ThisJob.JobIndex==ThePool.OldestJob){ //Check if this job is the oldest job, because if so then we need to now scan to see how many jobs after us also have ended and move the OldestJob marker to the oldest unfinished job
//Find the oldest active job
int index=ThePool.OldestJob; //Current oldest position in the circular queue
int maxjobs=ThePool.MaximumJobs; //For speed
bool finished=false; //Not done yet
while (finished==false){
if (ThePool.Jobs[index].JobRunning==false){
//This job is not running, go to the next one
ThePool.JobsAvailable++; //One more job has become available
index++; //Next job to check
if (index==maxjobs){ //If past the right end of the queue
index=0; //Wrap to the left
}
if ((index==NextJobToRun) || (index==NextEmptyJobIndex)){ //Did we get to the next non-running job that's actually a new job that hasn't been run yet? Or are there no new jobs and we just arrived at the next non-running empty job?
finished=true; //We're done
ThePool.OldestJob=index; //This 'new' non-running job is (or will be) the OldestJob
}
} else {
//This job is running, so its the new oldest
finished=true;
ThePool.OldestJob=index; //This index is the oldest active job
}
}
}
if (ThisJob.JobToRun==null){
return; //This was a termination job, close the thread permanently!
}
}
}
}
public class JobData{
//Data associated with a job function to be consumed by the threadpool
//This class is nested inside the ThreadPool class to allow is to share the delegate definition of 'Job'
//Internal fields
public bool JobRunning=false; //Whether this is a running job that some thread has claimed ownership of. If so, you should not modify this job yet. DO NOT MANUALLY CHANGE THIS, only threads should change this internally
public int JobIndex=0; //Index of this job, in case functions wish to use this - DO NOT MANUALLY CHANGE THIS it will be assigned automatically. Also used to identify which index a job is at when figuring out if its the oldest job
//User fields
public Job JobToRun; //A function to be run by the thread. Set this to null to tell the thread to close itself permanently. Since this is a delegate, you can do JobToRun+=extrafunction as many times as you like to add multiple function calls, provided they will all share the same parameter data
public int Integer1; //Function parameter data in the form of integers - you only need to set the ones that the function will use
public int Integer2;
public int Integer3;
public int Float1; //Function parameter data in the form of floats - you only need to set the ones that the function will use
public int Float2;
public int Float3;
public string Text1; //Text parameter data in the form of strings - you only need to set the ones that the function will use
}
}
}