OpenDDS::DCPS::ThreadPool Class Reference

#include <ThreadPool.h>

Collaboration diagram for OpenDDS::DCPS::ThreadPool:

Public Types
typedef ACE_THR_FUNC_RETURN(*	FunPtr) (void *)
	A typedef for the starting point of the ThreadPool. More...

Public Member Functions
	ThreadPool (size_t count, FunPtr fun, void *arg=0)
virtual	~ThreadPool ()
bool	contains (ACE_thread_t id) const

Static Public Member Functions
static ACE_THR_FUNC_RETURN	run (void *arg)
	A static helper function used to redirect to requested thread start point. More...

Private Member Functions
void	join_all ()
	OPENDDS_VECTOR (ACE_hthread_t) ids_
	OPENDDS_SET (ACE_thread_t) id_set_

Private Attributes
FunPtr	fun_
void *	arg_
ACE_Thread_Mutex	mutex_
ConditionVariable< ACE_Thread_Mutex >	cv_
ThreadStatusManager	tsm_
size_t	active_threads_

Detailed Description

ThreadPool is a light-weight utility class for starting a group of threads

ThreadPool creates several threads at construction and attempts to join them at destruction. Users of ThreadPool are responsible for making sure the running threads are in a joinable state before the destruction of ThreadPool

Definition at line 34 of file ThreadPool.h.

Member Typedef Documentation

FunPtr

typedef ACE_THR_FUNC_RETURN(* OpenDDS::DCPS::ThreadPool::FunPtr) (void *)

A typedef for the starting point of the ThreadPool.

Definition at line 39 of file ThreadPool.h.

Constructor & Destructor Documentation

ThreadPool()

OpenDDS::DCPS::ThreadPool::ThreadPool	(	size_t	count,
		FunPtr	fun,
		void *	arg = 0
	)

Creates a ThreadPool with the specifed size, starting point, and argument

Parameters

count	number of threads
fun	starting point for the threads of the ThreadPool
arg	an optional argument to pass to the starting function

Definition at line 22 of file ThreadPool.cpp.

References active_threads_, cv_, mutex_, run(), ACE_Thread::spawn(), tsm_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

 : fun_(fun)
 , arg_(arg)
 , mutex_()
 , cv_(mutex_)
 , active_threads_(0)
#ifdef OPENDDS_NO_THREAD_JOIN
 , finished_threads_(0)
#endif
 , ids_(count, ACE_thread_t())
{
  {
    ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
    for (size_t i = 0; i < count; ++i) {
      ACE_Thread::spawn(run, this, THR_NEW_LWP | THR_JOINABLE, 0, &(ids_[i]));
    }
  }
  if (count) {
    ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
    while (active_threads_ != count) {
      cv_.wait(tsm_);
    }
  }
}

~ThreadPool()

OpenDDS::DCPS::ThreadPool::~ThreadPool ( )

virtual

Definition at line 47 of file ThreadPool.cpp.

References join_all().

{
  join_all();
}

Member Function Documentation

contains()

bool OpenDDS::DCPS::ThreadPool::contains

(

ACE_thread_t

id

)

const

Check if a specific thread id belongs to this ThreadPool

Parameters

id	thread id to check

Returns

true if the ThreadPool contains thread with specified id, false otherwise

Definition at line 73 of file ThreadPool.cpp.

References mutex_.

Referenced by OpenDDS::DCPS::DispatchService::shutdown().

{
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  return id_set_.count(id);
}

join_all()

void OpenDDS::DCPS::ThreadPool::join_all ( )

private

Definition at line 79 of file ThreadPool.cpp.

References cv_, ACE_Thread::join(), mutex_, OPENDDS_ASSERT, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OPENDDS_VECTOR(), tsm_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

Referenced by ~ThreadPool().

{
#ifdef OPENDDS_NO_THREAD_JOIN
  ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
  while (finished_threads_ != ids_.size()) {
    cv_.wait(tsm_);
  }
#else
  OPENDDS_VECTOR(ACE_hthread_t) ids;
  {
    ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
    ids = ids_;
  }

  for (size_t i = 0; i < ids.size(); ++i) {
    const int result = ACE_Thread::join(ids[i], 0);
    ACE_UNUSED_ARG(result);
    OPENDDS_ASSERT(result == 0);
  }
#endif
}

OPENDDS_SET()

OpenDDS::DCPS::ThreadPool::OPENDDS_SET ( ACE_thread_t  )

private

OPENDDS_VECTOR()

OpenDDS::DCPS::ThreadPool::OPENDDS_VECTOR ( ACE_hthread_t  )

private

Referenced by join_all().

run()

ACE_THR_FUNC_RETURN OpenDDS::DCPS::ThreadPool::run ( void *  arg )

static

A static helper function used to redirect to requested thread start point.

Definition at line 52 of file ThreadPool.cpp.

References active_threads_, arg_, cv_, fun_, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OpenDDS::DCPS::ConditionVariable< Mutex >::notify_one(), ACE_Thread::self(), tsm_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

Referenced by ThreadPool().

{
  ThreadPool& pool = *static_cast<ThreadPool*>(arg);
  {
    ACE_Guard<ACE_Thread_Mutex> guard(pool.mutex_);
    pool.id_set_.insert(ACE_Thread::self());
    ++pool.active_threads_;
    pool.cv_.notify_all();
    while (pool.active_threads_ != pool.ids_.size()) {
      pool.cv_.wait(pool.tsm_);
    }
  }
  (*pool.fun_)(pool.arg_);
#ifdef OPENDDS_NO_THREAD_JOIN
  ACE_Guard<ACE_Thread_Mutex> guard(pool.mutex_);
  ++pool.finished_threads_;
  pool.cv_.notify_one();
#endif
  return 0;
}

Member Data Documentation

active_threads_

size_t OpenDDS::DCPS::ThreadPool::active_threads_

private

Definition at line 69 of file ThreadPool.h.

Referenced by run(), and ThreadPool().

arg_

void* OpenDDS::DCPS::ThreadPool::arg_

private

Definition at line 65 of file ThreadPool.h.

Referenced by run().

cv_

ConditionVariable<ACE_Thread_Mutex> OpenDDS::DCPS::ThreadPool::cv_

mutableprivate

Definition at line 67 of file ThreadPool.h.

Referenced by join_all(), run(), and ThreadPool().

fun_

FunPtr OpenDDS::DCPS::ThreadPool::fun_

private

Definition at line 64 of file ThreadPool.h.

Referenced by run().

mutex_

ACE_Thread_Mutex OpenDDS::DCPS::ThreadPool::mutex_

mutableprivate

Definition at line 66 of file ThreadPool.h.

Referenced by contains(), join_all(), run(), and ThreadPool().

tsm_

ThreadStatusManager OpenDDS::DCPS::ThreadPool::tsm_

private

Definition at line 68 of file ThreadPool.h.

Referenced by join_all(), run(), and ThreadPool().

---

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

• ThreadPool.h
• ThreadPool.cpp