#include <DispatchService.h>

Inheritance diagram for OpenDDS::DCPS::DispatchService:

Collaboration diagram for OpenDDS::DCPS::DispatchService:

[legend]

Public Types
typedef bool	DispatchStatus
	Typedef for Dispatch Return Values. More...

typedef long	TimerId
	Typedef for Schedule Return Values. More...

typedef void(*	FunPtr) (void *)

typedef std::pair< FunPtr, void * >	FunArgPair

Public Member Functions
typedef	OPENDDS_DEQUE (FunArgPair) EventQueue

	DispatchService (size_t count=1)

virtual	~DispatchService ()

void	shutdown (bool immediate=false, EventQueue *pending=0)

DispatchStatus	dispatch (FunPtr fun, void *arg=0)

template<typename T >
DispatchStatus	dispatch (T &ref)
	Helper function to dispatch arbitrary function objects (see fun_ptr_proxy) More...

TimerId	schedule (FunPtr fun, void *arg=0, const MonotonicTimePoint &expiration=MonotonicTimePoint::now())

template<typename T >
TimerId	schedule (T &ref, const MonotonicTimePoint &expiration=MonotonicTimePoint::now())
	Helper function to schedule arbitrary function objects (see fun_ptr_proxy) More...

size_t	cancel (TimerId id, void **arg=0)

size_t	cancel (FunPtr fun, void *arg=0)

template<typename T >
size_t	cancel (T &ref)
	Helper function to cancel arbitrary function objects (see fun_ptr_proxy) More...

Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Static Public Member Functions
template<typename T >
static void	fun_ptr_proxy (void *arg)
	Helper function for adapting arbitrary function objects (with operator()) More...

Static Public Attributes
static const bool	DS_SUCCESS = true
	DispatchStatus Success Constant. More...

static const bool	DS_ERROR = false
	DispatchStatus Error Constant. More...

static const long	TI_FAILURE = -1
	TimerId Failure Constant. More...

Private Types
typedef std::pair< FunArgPair, TimerId >	TimerPair

Private Member Functions
void	run_event_loop ()

typedef	OPENDDS_MULTIMAP (MonotonicTimePoint, TimerPair) TimerQueueMap

typedef	OPENDDS_MAP (TimerId, TimerQueueMap::iterator) TimerIdMap

Static Private Member Functions
static ACE_THR_FUNC_RETURN	run (void *arg)

Private Attributes
ACE_Thread_Mutex	mutex_

ConditionVariable< ACE_Thread_Mutex >	cv_

bool	allow_dispatch_

bool	stop_when_empty_

bool	running_

size_t	running_threads_

EventQueue	event_queue_

TimerQueueMap	timer_queue_map_

TimerIdMap	timer_id_map_

TimerId	max_timer_id_

ThreadPool	pool_

Additional Inherited Members
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Detailed Description

Definition at line 24 of file DispatchService.h.

Member Typedef Documentation

◆ DispatchStatus

typedef bool OpenDDS::DCPS::DispatchService::DispatchStatus

Typedef for Dispatch Return Values.

Definition at line 36 of file DispatchService.h.

◆ FunArgPair

typedef std::pair<FunPtr, void*> OpenDDS::DCPS::DispatchService::FunArgPair

Definition at line 51 of file DispatchService.h.

◆ FunPtr

typedef void(* OpenDDS::DCPS::DispatchService::FunPtr) (void *)

Definition at line 50 of file DispatchService.h.

◆ TimerId

typedef long OpenDDS::DCPS::DispatchService::TimerId

Typedef for Schedule Return Values.

Definition at line 45 of file DispatchService.h.

◆ TimerPair

typedef std::pair<FunArgPair, TimerId> OpenDDS::DCPS::DispatchService::TimerPair

private

Definition at line 129 of file DispatchService.h.

Constructor & Destructor Documentation

◆ DispatchService()

OpenDDS::DCPS::DispatchService::DispatchService ( size_t count = 1 )

explicit

Create a DispatchService

Parameters

count the requested size of the internal thread pool

Definition at line 22 of file DispatchService.cpp.

  : cv_(mutex_)
  , allow_dispatch_(true)
  , stop_when_empty_(false)
  , running_(true)
  , running_threads_(0)
  , max_timer_id_(LONG_MAX)
  , pool_(count, run, this)
 {
 }

◆ ~DispatchService()

OpenDDS::DCPS::DispatchService::~DispatchService ( )

virtual

Definition at line 33 of file DispatchService.cpp.

References shutdown().

 {
   shutdown();
 }

Member Function Documentation

◆ cancel() [1/3]

size_t OpenDDS::DCPS::DispatchService::cancel	(	DispatchService::TimerId	id,
		void **	arg = `0`
	)

Cancel a scheduled event by id

Parameters

id	the scheduled timer id to cancel
arg	if specified, arg will be set to the arg value passed in at time of scheduling

Returns: the number of timers successfully canceled (0 or 1)

Definition at line 114 of file DispatchService.cpp.

References cv_, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), timer_id_map_, and timer_queue_map_.

Referenced by OpenDDS::DCPS::ServiceEventDispatcher::cancel().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   TimerIdMap::iterator pos = timer_id_map_.find(id);
   if (pos != timer_id_map_.end()) {
     if (pos->second == timer_queue_map_.begin()) {
       cv_.notify_all();
     }
     if (arg) {
       *arg = pos->second->second.first.second;
     }
     timer_queue_map_.erase(pos->second);
     timer_id_map_.erase(pos);
     return 1;
   }
   return 0;
 }

◆ cancel() [2/3]

size_t OpenDDS::DCPS::DispatchService::cancel	(	FunPtr	fun,
		void *	arg = `0`
	)

Cancel a scheduled event by function pointer

Parameters

fun	the function pointer of the event/s to cancel
arg	the argument passed at time of scheduling

Returns: the number of timers successfully canceled (potentially several)

Definition at line 132 of file DispatchService.cpp.

References cv_, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OPENDDS_ASSERT, timer_id_map_, and timer_queue_map_.

 {
   OPENDDS_ASSERT(fun);
   size_t count = 0;
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   for (TimerQueueMap::iterator it = timer_queue_map_.begin(); it != timer_queue_map_.end();) {
     if (it->second.first.first == fun && it->second.first.second == arg) {
       if (it == timer_queue_map_.begin()) {
         cv_.notify_all();
       }
       timer_id_map_.erase(it->second.second);
       timer_queue_map_.erase(it++);
       ++count;
     } else {
       ++it;
     }
   }
   return count;
 }

◆ cancel() [3/3]

template<typename T >

size_t OpenDDS::DCPS::DispatchService::cancel ( T & ref )

inline

Helper function to cancel arbitrary function objects (see fun_ptr_proxy)

Definition at line 119 of file DispatchService.h.

   {
     return cancel(fun_ptr_proxy<T>, &ref);
   }

◆ dispatch() [1/2]

DispatchService::DispatchStatus OpenDDS::DCPS::DispatchService::dispatch	(	FunPtr	fun,
		void *	arg = `0`
	)

Dispatch an event

Parameters

fun	the function pointer to dispatch
arg	the argument to pass during dispatch

Returns: true if event successfully enqueue, otherwise false

Definition at line 74 of file DispatchService.cpp.

References allow_dispatch_, cv_, DS_ERROR, DS_SUCCESS, event_queue_, mutex_, and OpenDDS::DCPS::ConditionVariable< Mutex >::notify_one().

Referenced by OpenDDS::DCPS::ServiceEventDispatcher::dispatch().

 {
   if (!fun) {
     return DS_ERROR;
   }
 
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   if (allow_dispatch_) {
     event_queue_.push_back(std::make_pair(fun, arg));
     cv_.notify_one();
     return DS_SUCCESS;
   }
   return DS_ERROR;
 }

◆ dispatch() [2/2]

template<typename T >

DispatchStatus OpenDDS::DCPS::DispatchService::dispatch ( T & ref )

inline

Helper function to dispatch arbitrary function objects (see fun_ptr_proxy)

Definition at line 80 of file DispatchService.h.

References OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now().

   {
     return dispatch(fun_ptr_proxy<T>, &ref);
   }

◆ fun_ptr_proxy()

template<typename T >

static void OpenDDS::DCPS::DispatchService::fun_ptr_proxy ( void * arg )

inlinestatic

Helper function for adapting arbitrary function objects (with operator())

Definition at line 29 of file DispatchService.h.

References OpenDDS::DCPS::ref().

   {
     T& ref = *static_cast<T*>(arg);
     ref();
   }

◆ OPENDDS_DEQUE()

typedef OpenDDS::DCPS::DispatchService::OPENDDS_DEQUE ( FunArgPair )

◆ OPENDDS_MAP()

typedef OpenDDS::DCPS::DispatchService::OPENDDS_MAP	(	TimerId	,
		TimerQueueMap::iterator
	)

private

◆ OPENDDS_MULTIMAP()

typedef OpenDDS::DCPS::DispatchService::OPENDDS_MULTIMAP	(	MonotonicTimePoint	,
		TimerPair
	)

private

◆ run()

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

staticprivate

Definition at line 152 of file DispatchService.cpp.

References run_event_loop().

 {
   DispatchService& dispatcher = *static_cast<DispatchService*>(arg);
   dispatcher.run_event_loop();
   return 0;
 }

◆ run_event_loop()

void OpenDDS::DCPS::DispatchService::run_event_loop ( void )

private

Definition at line 159 of file DispatchService.cpp.

References allow_dispatch_, cv_, event_queue_, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, running_, running_threads_, stop_when_empty_, TheServiceParticipant, timer_id_map_, timer_queue_map_, OpenDDS::DCPS::ConditionVariable< Mutex >::wait(), and OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until().

Referenced by run().

 {
   ACE_Reverse_Lock<ACE_Thread_Mutex> rev_lock(mutex_);
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   ++running_threads_;
   while (running_) {
 
     // Logical Order:
     // - Move expired timer events into normal event queue
     // - Wait appropriate length if there's nothing to do
     // - Check for early exit before execution
     // - Run first task from event queue
 
     if (allow_dispatch_ && !timer_queue_map_.empty()) {
       const MonotonicTimePoint now = MonotonicTimePoint::now();
 
       TimerQueueMap::iterator last = timer_queue_map_.upper_bound(now), pos = last;
       while (pos != timer_queue_map_.begin()) {
         --pos;
         event_queue_.push_back(pos->second.first);
         timer_id_map_.erase(pos->second.second);
       }
       if (last != timer_queue_map_.begin()) {
         timer_queue_map_.erase(timer_queue_map_.begin(), last);
       }
     }
 
     if (event_queue_.empty()) {
       if (stop_when_empty_) {
         running_ = false;
         cv_.notify_all();
       } else if (allow_dispatch_ && timer_queue_map_.size()) {
         MonotonicTimePoint deadline(timer_queue_map_.begin()->first);
         cv_.wait_until(deadline, TheServiceParticipant->get_thread_status_manager());
       } else {
         cv_.wait(TheServiceParticipant->get_thread_status_manager());
       }
     }
 
     if (!running_ || event_queue_.empty()) continue;
 
     FunArgPair pair = event_queue_.front();
     event_queue_.pop_front();
     ACE_Guard<ACE_Reverse_Lock<ACE_Thread_Mutex> > rev_guard(rev_lock);
     pair.first(pair.second);
   }
   --running_threads_;
   cv_.notify_all();
 }

◆ schedule() [1/2]

DispatchService::TimerId OpenDDS::DCPS::DispatchService::schedule	(	FunPtr	fun,
		void *	arg = `0`,
		const MonotonicTimePoint &	expiration = `MonotonicTimePoint::now()`
	)

Schedule the future dispatch of an event

Parameters

fun	the function pointer to schedule
arg	the argument to pass during dispatch
expiration	the requested dispatch time (no earlier than)

Returns: -1 on a failure, otherwise the timer id for the future dispatch

Definition at line 89 of file DispatchService.cpp.

References allow_dispatch_, cv_, max_timer_id_, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_one(), TI_FAILURE, timer_id_map_, and timer_queue_map_.

Referenced by OpenDDS::DCPS::ServiceEventDispatcher::schedule().

 {
   if (!fun) {
     return TI_FAILURE;
   }
 
   TimerId id = 0;
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   if (allow_dispatch_) {
     TimerQueueMap::iterator pos = timer_queue_map_.insert(std::make_pair(expiration, std::make_pair(std::make_pair(fun, arg), 0)));
     // Make it a loop in case we ever recycle timer ids
     const TimerId starting_id = max_timer_id_;
     do {
       id = max_timer_id_ = max_timer_id_ == LONG_MAX ? 1 : max_timer_id_ + 1;
       if (id == starting_id) {
         return TI_FAILURE; // all ids in use ?!
       }
       pos->second.second = id;
     } while (timer_id_map_.insert(std::make_pair(id, pos)).second == false);
     cv_.notify_one();
     return id;
   }
   return TI_FAILURE;
 }

◆ schedule() [2/2]

template<typename T >

TimerId OpenDDS::DCPS::DispatchService::schedule	(	T &	ref,
		const MonotonicTimePoint &	expiration = `MonotonicTimePoint::now()`
	)

inline

Helper function to schedule arbitrary function objects (see fun_ptr_proxy)

Definition at line 96 of file DispatchService.h.

   {
     return schedule(fun_ptr_proxy<T>, &ref, expiration);
   }

◆ shutdown()

void OpenDDS::DCPS::DispatchService::shutdown	(	bool	immediate = `false`,
		EventQueue *	pending = `0`
	)

Request shutdown of this DispatchService, which prevents sucessful future calls to either dispatch or schedule and cancels all scheduled events.

Parameters

immediate	prevent any further dispatches from event queue, otherwise allow current queue to empty
pending	An EventQueue object in which to store canceled events in case extra processing is required

Definition at line 38 of file DispatchService.cpp.

References ACE_ERROR, allow_dispatch_, OpenDDS::DCPS::ThreadPool::contains(), cv_, OpenDDS::DCPS::LogLevel::Error, event_queue_, LM_ERROR, OpenDDS::DCPS::log_level, mutex_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_all(), pool_, running_, running_threads_, ACE_Thread::self(), stop_when_empty_, TheServiceParticipant, timer_id_map_, timer_queue_map_, and OpenDDS::DCPS::ConditionVariable< Mutex >::wait().

Referenced by OpenDDS::DCPS::ServiceEventDispatcher::shutdown(), and ~DispatchService().

 {
   ACE_Guard<ACE_Thread_Mutex> guard(mutex_);
   allow_dispatch_ = false;
   stop_when_empty_ = true;
   running_ = running_ && !immediate; // && with existing state in case shutdown has already been called
   cv_.notify_all();
 
   if (pool_.contains(ACE_Thread::self())) {
     if (log_level >= LogLevel::Error) {
       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR :: DispatchService::shutdown: Contained Thread Attempting To Call Shutdown.\n"));
     }
     if (pending) {
       pending->clear();
     }
     return;
   }
 
   while (running_threads_) {
     cv_.wait(TheServiceParticipant->get_thread_status_manager());
   }
 
   if (pending) {
     pending->clear();
     pending->swap(event_queue_);
     const TimerQueueMap& cmap = timer_queue_map_;
     for (TimerQueueMap::const_iterator it = cmap.begin(), limit = cmap.end(); it != limit; ++it) {
       pending->push_back(it->second.first);
     }
   } else {
     event_queue_.clear();
   }
   timer_queue_map_.clear();
   timer_id_map_.clear();
 }

Member Data Documentation

◆ allow_dispatch_

bool OpenDDS::DCPS::DispatchService::allow_dispatch_

private

Definition at line 135 of file DispatchService.h.

Referenced by dispatch(), run_event_loop(), schedule(), and shutdown().

◆ cv_

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

mutableprivate

Definition at line 134 of file DispatchService.h.

Referenced by cancel(), dispatch(), run_event_loop(), schedule(), and shutdown().

◆ DS_ERROR

const bool OpenDDS::DCPS::DispatchService::DS_ERROR = false

static

DispatchStatus Error Constant.

Definition at line 42 of file DispatchService.h.

Referenced by dispatch().

◆ DS_SUCCESS

const bool OpenDDS::DCPS::DispatchService::DS_SUCCESS = true

static

DispatchStatus Success Constant.

Definition at line 39 of file DispatchService.h.

Referenced by dispatch().

◆ event_queue_

EventQueue OpenDDS::DCPS::DispatchService::event_queue_

private

Definition at line 139 of file DispatchService.h.

Referenced by dispatch(), run_event_loop(), and shutdown().

◆ max_timer_id_

TimerId OpenDDS::DCPS::DispatchService::max_timer_id_

private

Definition at line 142 of file DispatchService.h.

Referenced by schedule().

◆ mutex_

ACE_Thread_Mutex OpenDDS::DCPS::DispatchService::mutex_

mutableprivate

Definition at line 133 of file DispatchService.h.

Referenced by cancel(), dispatch(), run_event_loop(), schedule(), and shutdown().

◆ pool_

ThreadPool OpenDDS::DCPS::DispatchService::pool_

private

Definition at line 143 of file DispatchService.h.

Referenced by shutdown().

◆ running_

bool OpenDDS::DCPS::DispatchService::running_

private

Definition at line 137 of file DispatchService.h.

Referenced by run_event_loop(), and shutdown().

◆ running_threads_

size_t OpenDDS::DCPS::DispatchService::running_threads_

private

Definition at line 138 of file DispatchService.h.

Referenced by run_event_loop(), and shutdown().

◆ stop_when_empty_

bool OpenDDS::DCPS::DispatchService::stop_when_empty_

private

Definition at line 136 of file DispatchService.h.

Referenced by run_event_loop(), and shutdown().

◆ TI_FAILURE

const long OpenDDS::DCPS::DispatchService::TI_FAILURE = -1

static

TimerId Failure Constant.

Definition at line 48 of file DispatchService.h.

Referenced by schedule().

◆ timer_id_map_

TimerIdMap OpenDDS::DCPS::DispatchService::timer_id_map_

private

Definition at line 141 of file DispatchService.h.

Referenced by cancel(), run_event_loop(), schedule(), and shutdown().

◆ timer_queue_map_

TimerQueueMap OpenDDS::DCPS::DispatchService::timer_queue_map_

private

Definition at line 140 of file DispatchService.h.

Referenced by cancel(), run_event_loop(), schedule(), and shutdown().

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

Public Types

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Types

Private Member Functions

Static Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ DispatchStatus

◆ FunArgPair

◆ FunPtr

◆ TimerId

◆ TimerPair

Constructor & Destructor Documentation

◆ DispatchService()

◆ ~DispatchService()

Member Function Documentation

◆ cancel() [1/3]

◆ cancel() [2/3]

◆ cancel() [3/3]

◆ dispatch() [1/2]

◆ dispatch() [2/2]

◆ fun_ptr_proxy()

◆ OPENDDS_DEQUE()

◆ OPENDDS_MAP()

◆ OPENDDS_MULTIMAP()

◆ run()

◆ run_event_loop()

◆ schedule() [1/2]

◆ schedule() [2/2]

◆ shutdown()

Member Data Documentation

◆ allow_dispatch_

◆ cv_

◆ DS_ERROR

◆ DS_SUCCESS

◆ event_queue_

◆ max_timer_id_

◆ mutex_

◆ pool_

◆ running_

◆ running_threads_

◆ stop_when_empty_

◆ TI_FAILURE

◆ timer_id_map_

◆ timer_queue_map_