Using ThreadPool

[Roy Chastain]

I have a set of treads that I have called a Parallel Thread Pool.
This is a pool of threads. Each thread processes item from a specific
queue. Once the thread specific processing of an item is complete, I
want to move the execution of methods associated with this object to a
pool of non-specific threads.

The obvious choice is to simple do a QueueUserWorkItem to the
ThreadPool. However,...

In my model an object might be queued to thread pool more than once.
(By that I mean that the thread specific processing could complete
more than one time before the pool of general threads could complete
the first set of post processing work. For those of you with
operating system experience, think of it as interrupt pre-processing
and interrupt post-processing.)

The state of the object will be fine. I have handled that. What I
need is a method of making sure that a 2nd work item based on the same
object is not run before the first one is completed by the post
processing thread pool.

I keep thinking that something with signals such that the
pre-processing thread (my Parallel Thread Pool thread) would put the
item onto a wait queue that is waiting for a signal. When a post
processing thread completes with an object, it would signal and the
wait any thread would then do the QueueUserWorkItem so that any thread
in the pool could actually do the processing.

Questions are
1) - Does anyone have a better idea
2) - The documentation keeps saying things like "One thread monitors
the status of several wait operations..." This is the part that I
don't understand. How do I get the ThreadPool to monitor a wait. In
my proposed scheme I would want it to monitor the signal that a
completing thread sets.

Thanks

 

[Ayende Rahien]

I have a set of treads that I have called a Parallel Thread Pool.
This is a pool of threads. Each thread processes item from a specific
queue. Once the thread specific processing of an item is complete, I
want to move the execution of methods associated with this object to a
pool of non-specific threads.

Do something like this:
(Untested & uncompiled but it should work)

class ThreadPool
{
int threadLimit;
ArrayList WorkToDo = new ArrayList();
ArrayList FreeThreads = new ArrayList();
ArrayList AllThreads = new ArrayList();
bool KeepProccessing=true;

public class Work
{
public delegate Method;
public object [] args;
}

public int ThreadLimit
{
get{return threadLimi;}
set{threadLimit=value;}
}

public StopWorking()
{
KeepProccess = false;
lock(AllThreads)
{
foreach(Thread t in AllThreads)
{
//Safe approach
if (t.ThreadState = ThreadState.Suspended)//will return instantly
t.Resume();
t.Join();//Will wait until the thread is done

/*Quick approach
t.Abort();//Usually don't let the thread time to free resources
*/
}
}
}

public AddWork(Work w)
{
lock(WorkToDo)
{
WorkToDo.Add(w);
}
//Don't care if it's changed in the middle here, because nothing bad happens
if it does
if(FreeThreads.Count>0)
{
lock(FreeThreads)
{
Thread t = (Thread)FreeThreads[0];
FreeThreads.Remove(t);
}
t.Resume();
}
//Don't care if it's changed in the middle here, because nothing bad happens
if it does
else if(AllThreads.Count<threadLimit)
{
Thread t = new Thread(new ThreadStart(DoWork));
lock(AllThreads)
{
AllThreads.Add(t);
}
t.Start();
}
}

private void DoWork()
{
Work w;
while(this.KeepWorking)
{
w = null;
lock(WorkToDo)
{
//Avoid a case where two threads accessing WorkToDo.Count at the same time
and trying to get the current Work, but would fail if there is only one Work
//Would cause null exception
if(WorkToDo.Count>0)
{
w = (Work)WorkToDo[0];
WorkToDo.Remove(w);
}
}
if(w!=null)
w.Method.DynamicInvoke(w.args);
else
Thread.CurrentThread.Suspend();
}

}
}