vpr::ThreadPool Class Reference

A pool of threads to process user jobs. More...

#include <vpr/Thread/ThreadPool.h>

Collaboration diagram for vpr::ThreadPool:

[legend]List of all members.

Public Member Functions
	ThreadPool (int numToStartWith=1)
	Constructor.
	~ThreadPool ()
void	startFunc (thread_func_t func, void *arg=NULL)
	Gives a function to the processes.
void	startFunc (BaseThreadFunctor *theFunctor, void *argument)
void	startFunc (BaseThreadFunctor *theFunctor, const bool deleteFunctor=false)
OneThread *	addThread ()
void	threadLoop (void *theThreadAsVoid)
	Body of a general-purpose child process.
void	threadSleep (OneThread *theThread)
	Waits for work to do.
OneThread *	getThread ()
	Gets a process to run.
void	wait ()
	Waits until all threads are done doing their work.
void	printList ()

---

Detailed Description

A pool of threads to process user jobs.

Outline of thread pool:

|-Need to initialize all thread given a parameter to the constructor of the # to use
|-Need a function to block until all work is done that has started
|-Just repetiviely add work to the process

Date:

February 5, 1997

Definition at line 104 of file ThreadPool.h.

---

Constructor & Destructor Documentation

vpr::ThreadPool::ThreadPool

(

int

numToStartWith = 1

)

Constructor.

Definition at line 74 of file ThreadPool.cpp.

References addThread(), vpr::MutexPosix::release(), vprDBG_ALL(), vprDBG_DETAILED_LVL, vprDBG_HVERB_LVL, vprDEBUG, and vprDEBUG_FLUSH.

                                          : readyThreads(0)
{
   //DebugLock.acquire();
   vprDEBUG(vprDBG_ALL, vprDBG_DETAILED_LVL)
      << "vprThreadPool::vprThreadPool: Entering.\n" << vprDEBUG_FLUSH;
   vprDEBUG(vprDBG_ALL, vprDBG_HVERB_LVL)
      << "\tvprThreadPool::vprThreadPool: Number threads: " << numToStartWith
      << std::endl << vprDEBUG_FLUSH;
   //DebugLock.release();

   listHead = NULL;
   workingCount = 0;
   listLock.release();             // release threadList
   finishedLock.release();         // Initialize if to threads being done

   //-- Start the initial # of threads ---//
   for ( int index=0;index < numToStartWith;index++ )
   {
      addThread();
   }
}

vpr::ThreadPool::~ThreadPool

(

)

Definition at line 96 of file ThreadPool.cpp.

References vpr::OneThread::next, and vpr::OneThread::thread.

{
   OneThread* cur_thread = listHead;
   while ( cur_thread != NULL )
   {
      delete cur_thread->thread;
      cur_thread->thread = NULL;
      OneThread* old_thread = cur_thread;
      cur_thread = cur_thread->next;
      delete old_thread;
   }
}

---

Member Function Documentation

void vpr::ThreadPool::startFunc	(	thread_func_t	func,
		void *	arg = NULL
	)			[inline]

Gives a function to the processes.

Start a function going asynchronously. Called by master process.

Definition at line 116 of file ThreadPool.h.

Referenced by startFunc().

   {
      this->startFunc(new ThreadNonMemberFunctor(func, arg), true);
   }

void vpr::ThreadPool::startFunc	(	BaseThreadFunctor *	theFunctor,
		void *	argument
	)			[inline]

Definition at line 121 of file ThreadPool.h.

References vpr::BaseThreadFunctor::setArg(), and startFunc().

   {
      theFunctor->setArg(argument);
      this->startFunc(theFunctor);
   }

void vpr::ThreadPool::startFunc	(	BaseThreadFunctor *	theFunctor,
		const bool	deleteFunctor = false
	)			[inline]

Definition at line 127 of file ThreadPool.h.

References vpr::OneThread::deleteFunctor, vpr::OneThread::functor, getThread(), vpr::SemaphorePosix::release(), and vpr::OneThread::threadWait.

   {
      OneThread* theThread = getThread();

      theThread->deleteFunctor = deleteFunctor;
      theThread->functor = theFunctor;     /* set address of func to exec */
      theThread->threadWait.release();     /* wake up sleeping process */
      //delete theFunctor;
   }

OneThread * vpr::ThreadPool::addThread

(

)

Definition at line 222 of file ThreadPool.cpp.

References vpr::OneThread::next, printList(), vpr::ThreadPosix::self(), vpr::OneThread::thread, threadLoop(), vprDBG_ALL(), vprDBG_DETAILED_LVL, vprDBG_HVERB_LVL, vprDEBUG, and vprDEBUG_FLUSH.

Referenced by ThreadPool().

{
   static int numTimes = 0;
//    DebugLock.acquire();
   vprDEBUG(vprDBG_ALL, vprDBG_DETAILED_LVL) << Thread::self()
      << " vpr::ThreadPool::addThread: Entering: " << ++numTimes << std::endl
      << vprDEBUG_FLUSH;
//    DebugLock.release();

   Guard<Mutex> guard(listLock);   // Protect the head

   //OneThread* newThread = new (this->getMyMemPool()->allocate(sizeof(OneThread))) OneThread;    // Used placement new
   OneThread* newThread = new OneThread;
   newThread->next = NULL;

//    ThreadMemberFunctor<ThreadPool>* memberFunctor = new ThreadMemberFunctor<ThreadPool>(this, ThreadPool::threadLoop, (void*)newThread);
   ThreadMemberFunctor<ThreadPool>* memberFunctor = new ThreadMemberFunctor<ThreadPool>(this, &ThreadPool::threadLoop, (void*)newThread);

   newThread->thread = new Thread(memberFunctor);

//    DebugLock.acquire();
   vprDEBUG(vprDBG_ALL, vprDBG_HVERB_LVL) << newThread->thread
      << " vprThreadPool::addThread: List at end\n" << vprDEBUG_FLUSH;
   printList();
//    DebugLock.release();

   return listHead;
}

void vpr::ThreadPool::threadLoop

(

void *

theThreadAsVoid

)

Body of a general-purpose child process.

The argument, which must be declared void* to match the func prototype, is the vpr::OneThread structure that represents this process. The contents of that struct, in particular threadWait(), MUST be initialized by the parent.

Definition at line 118 of file ThreadPool.cpp.

References vpr::MutexPosix::acquire(), vpr::OneThread::functor, vpr::MutexPosix::release(), vpr::ThreadPosix::self(), threadSleep(), vprDBG_ALL(), vprDBG_DETAILED_LVL, vprDEBUG, and vprDEBUG_FLUSH.

Referenced by addThread().

{
//   DebugLock.acquire();
   vprDEBUG(vprDBG_ALL, vprDBG_DETAILED_LVL) << Thread::self()
      << " vpr::ThreadPool::threadLoop: Entering."
      << std::endl << vprDEBUG_FLUSH;
//      vprDEBUG(vprDBG_ALL, vprDBG_HVERB_LVL) << Thread::self()
//      << " vpr::ThreadPool::threadLoop: theThreadAsVoid:"
//      << theThreadAsVoid << endl << vprDEBUG_FLUSH;
//   DebugLock.release();

   listLock.acquire();
   listLock.release();     // Do this to make sure addThread is done

   OneThread* myThread = (OneThread*)theThreadAsVoid;

#ifdef VPR_USE_IRIX_SPROC

   // --- SIGNAL Handlers ---- //
   prctl(PR_TERMCHILD);       // What should I do with. FIX - Allen
#endif

   for ( ;; )
   {
      // --- WAIT FOR WORK --- //
      threadSleep(myThread);
      // ASSERT:  We now have work to do...
      // --- PROCESS ENTRY OVERHEAD --- //
      workingCountLock.acquire();     // Get access to the working thread count
      if ( workingCount == 0 )
      {
         finishedLock.acquire();       // Now there are threads working
      }
      workingCount = workingCount + 1;    // Update thread count
      workingCountLock.release();

      // --- DO THE WORK --- //
      myThread->functor->operator()();

      // --- PROCESS EXIT OVERHEAD --- //
      workingCountLock.acquire();     // Get access to the working count
      workingCount = workingCount - 1;
      if ( workingCount == 0 )
      {
         finishedLock.release();       // Now there are no threads working
      }
      workingCountLock.release();
   }
}

void vpr::ThreadPool::threadSleep

(

OneThread *

theThread

)

Waits for work to do.

Put a vpr::OneThread structure on the ready list and sleep on it. Called by a child process when its work is done.

Definition at line 172 of file ThreadPool.cpp.

References vpr::SemaphorePosix::acquire(), vpr::MutexPosix::acquire(), vpr::OneThread::next, vpr::SemaphorePosix::release(), vpr::MutexPosix::release(), and vpr::OneThread::threadWait.

Referenced by threadLoop().

{
   listLock.acquire();               // acquire exclusive rights to threadList
   theThread->next = listHead;   // put self on head of the list
   listHead = theThread;
   listLock.release();               // release threadList

   readyThreads.release();           // notify master, at least 1 on the list

   theThread->threadWait.acquire();  // sleep until master needs/releases me
}

OneThread * vpr::ThreadPool::getThread

(

)

Gets a process to run.

Acquire a vpr::OneThread structure from the ready list, waiting if necessary. Called by the master process as part of dispatching a thread.

Definition at line 189 of file ThreadPool.cpp.

References vpr::MutexPosix::acquire(), vpr::SemaphorePosix::acquire(), vpr::OneThread::next, and vpr::MutexPosix::release().

Referenced by startFunc().

{
   OneThread* theThread;

   readyThreads.acquire();           // wait until at least 1 thread is free

   listLock.acquire();               // acquire exclusive rights to threadList
   theThread = listHead;         // get address of first free OneThread
   listHead = theThread->next;   // make next in list, the head of list
   listLock.release();               // release threadList

   return theThread;
}

void vpr::ThreadPool::wait

(

)

[inline]

Waits until all threads are done doing their work.

Note:

This was renamed from barrier() in version 0.92.1 to deal with a case where that name is a preprocessor macro.

Since:

0.92.1

Definition at line 169 of file ThreadPool.h.

References vpr::MutexPosix::acquire(), and vpr::MutexPosix::release().

   {
      finishedLock.acquire();      // Get the lock that means threads done
      finishedLock.release();      // Reset it to done
   }

void vpr::ThreadPool::printList

(

)

Definition at line 205 of file ThreadPool.cpp.

References vpr::OneThread::next.

Referenced by addThread().

{
   OneThread* curThread = listHead;
   int counter = 0;

   std::cerr << "----- Thread List -----\n";

   while ( curThread != NULL )
   {
      std::cerr << "Thread: " << counter++ << std::endl;
      std::cerr << "\tpid: " << *curThread << std::endl;
      curThread = curThread->next;
   }
}

---

The documentation for this class was generated from the following files:

• ThreadPool.h
• ThreadPool.cpp

---