#include <DcpsUpcalls.h>

Inheritance diagram for OpenDDS::DCPS::DcpsUpcalls:

[legend]

Collaboration diagram for OpenDDS::DCPS::DcpsUpcalls:

[legend]

Public Member Functions
	DcpsUpcalls (DataReaderCallbacks_rch drr, const GUID_t &reader, const WriterAssociation &wa, bool active, DataWriterCallbacks_rch dwr)

int	svc ()

void	writer_done ()

Public Member Functions inherited from ACE_Task_Base
	ACE_Task_Base (ACE_Thread_Manager *=0)

virtual	~ACE_Task_Base (void)

virtual int	open (void *args=0)

virtual int	close (u_long flags=0)

virtual int	module_closed (void)

virtual int	put (ACE_Message_Block , ACE_Time_Value =0)

virtual int	activate (long flags=THR_NEW_LWP\|THR_JOINABLE\|THR_INHERIT_SCHED, int n_threads=1, int force_active=0, long priority=ACE_DEFAULT_THREAD_PRIORITY, int grp_id=-1, ACE_Task_Base task=0, ACE_hthread_t thread_handles[]=0, void stack[]=0, size_t stack_size[]=0, ACE_thread_t thread_ids[]=0, const char *thr_name[]=0)

virtual int	wait (void)

virtual int	suspend (void)

virtual int	resume (void)

int	grp_id (void) const

void	grp_id (int)

ACE_Thread_Manager *	thr_mgr (void) const

void	thr_mgr (ACE_Thread_Manager *)

int	is_reader (void) const

int	is_writer (void) const

size_t	thr_count (void) const

ACE_thread_t	last_thread (void) const

Public Member Functions inherited from ACE_Service_Object
	ACE_Service_Object (ACE_Reactor *=0)

virtual	~ACE_Service_Object (void)

Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)

virtual ACE_HANDLE	get_handle (void) const

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_input (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_timeout (const ACE_Time_Value &current_time, const void *act=0)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

virtual int	handle_signal (int signum, siginfo_t =0, ucontext_t =0)

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

virtual Reference_Count	add_reference (void)

virtual Reference_Count	remove_reference (void)

Reference_Counting_Policy &	reference_counting_policy (void)

Public Member Functions inherited from ACE_Shared_Object
	ACE_Shared_Object (void)

virtual	~ACE_Shared_Object (void)

virtual int	init (int argc, ACE_TCHAR *argv[])

virtual int	fini (void)

virtual int	info (ACE_TCHAR **info_string, size_t length=0) const

Private Attributes
DataReaderCallbacks_wrch	drr_

const GUID_t &	reader_

const WriterAssociation &	wa_

bool	active_

DataWriterCallbacks_wrch	dwr_

bool	reader_done_

bool	writer_done_

ACE_Thread_Mutex	mtx_

ConditionVariable< ACE_Thread_Mutex >	cnd_

Additional Inherited Members
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Static Public Member Functions inherited from ACE_Task_Base
static ACE_THR_FUNC_RETURN	svc_run (void *)

static void	cleanup (void object, void params)

Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)

static int	register_stdin_handler (ACE_Event_Handler eh, ACE_Reactor reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)

static int	remove_stdin_handler (ACE_Reactor reactor, ACE_Thread_Manager thr_mgr)

Public Attributes inherited from ACE_Service_Object
	ACE_ALLOC_HOOK_DECLARE

Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

Protected Attributes inherited from ACE_Task_Base
size_t	thr_count_

ACE_Thread_Manager *	thr_mgr_

u_long	flags_

int	grp_id_

ACE_thread_t	last_thread_id_

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Definition at line 24 of file DcpsUpcalls.h.

Constructor & Destructor Documentation

◆ DcpsUpcalls()

OpenDDS::DCPS::DcpsUpcalls::DcpsUpcalls	(	DataReaderCallbacks_rch	drr,
		const GUID_t &	reader,
		const WriterAssociation &	wa,
		bool	active,
		DataWriterCallbacks_rch	dwr
	)

Definition at line 12 of file DcpsUpcalls.cpp.

   : drr_(drr)
   , reader_(reader)
   , wa_(wa)
   , active_(active)
   , dwr_(dwr)
   , reader_done_(false)
   , writer_done_(false)
   , cnd_(mtx_)
 {
 }

Member Function Documentation

◆ svc()

int OpenDDS::DCPS::DcpsUpcalls::svc ( void )

virtual

Reimplemented from ACE_Task_Base.

Definition at line 29 of file DcpsUpcalls.cpp.

References ACE_ERROR, ACE_GUARD_RETURN, active_, cnd_, OpenDDS::DCPS::CvStatus_Error, OpenDDS::DCPS::CvStatus_NoTimeout, OpenDDS::DCPS::CvStatus_Timeout, OpenDDS::DCPS::DCPS_debug_level, drr_, LM_ERROR, OpenDDS::DCPS::WeakRcHandle< T >::lock(), mtx_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_one(), OpenDDS::DCPS::TimePoint_T< MonotonicClock >::now(), reader_, reader_done_, TheServiceParticipant, OpenDDS::DCPS::ThreadStatusManager::thread_status_interval(), OpenDDS::DCPS::ThreadStatusManager::update_thread_status(), wa_, OpenDDS::DCPS::ConditionVariable< Mutex >::wait(), OpenDDS::DCPS::ConditionVariable< Mutex >::wait_until(), and writer_done_.

 {
   ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   const TimeDuration thread_status_interval = thread_status_manager.thread_status_interval();
   const bool update_thread_status = thread_status_manager.update_thread_status();
 
   ThreadStatusManager::Start s(thread_status_manager, "DcpsUpcalls");
 
   MonotonicTimePoint expire = MonotonicTimePoint::now() + thread_status_interval;
 
   DataReaderCallbacks_rch drr = drr_.lock();
   if (!drr) {
     return 0;
   }
   drr->add_association(reader_, wa_, active_);
 
   {
     ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mtx_, -1);
     reader_done_ = true;
     cnd_.notify_one();
     while (!writer_done_) {
       if (update_thread_status) {
         switch (cnd_.wait_until(expire, thread_status_manager)) {
         case CvStatus_NoTimeout:
           break;
 
         case CvStatus_Timeout:
           expire = MonotonicTimePoint::now() + thread_status_interval;
           break;
 
         case CvStatus_Error:
           if (DCPS_debug_level) {
             ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DcpsUpcalls::svc: error in wait_utill\n"));
           }
           return -1;
         }
       } else if (cnd_.wait(thread_status_manager) == CvStatus_Error) {
         if (DCPS_debug_level) {
           ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: DcpsUpcalls::svc: error in wait\n"));
         }
         return -1;
       }
     }
   }
 
   return 0;
 }

◆ writer_done()

void OpenDDS::DCPS::DcpsUpcalls::writer_done ( )

Definition at line 77 of file DcpsUpcalls.cpp.

References ACE_GUARD, cnd_, mtx_, OpenDDS::DCPS::ConditionVariable< Mutex >::notify_one(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, TheServiceParticipant, ACE_Task_Base::wait(), and writer_done_.

Referenced by OpenDDS::DCPS::StaticEndpointManager::match_continue().

 {
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mtx_);
     writer_done_ = true;
     cnd_.notify_one();
   }
 
   ThreadStatusManager& thread_status_manager = TheServiceParticipant->get_thread_status_manager();
   ThreadStatusManager::Sleeper sleeper(thread_status_manager);
   wait(); // ACE_Task_Base::wait does not accept a timeout
 }

Member Data Documentation

◆ active_

bool OpenDDS::DCPS::DcpsUpcalls::active_

private

Definition at line 40 of file DcpsUpcalls.h.

Referenced by svc().

◆ cnd_

ConditionVariable<ACE_Thread_Mutex> OpenDDS::DCPS::DcpsUpcalls::cnd_

private

Definition at line 44 of file DcpsUpcalls.h.

Referenced by svc(), and writer_done().

◆ drr_

DataReaderCallbacks_wrch OpenDDS::DCPS::DcpsUpcalls::drr_

private

Definition at line 37 of file DcpsUpcalls.h.

Referenced by svc().

◆ dwr_

DataWriterCallbacks_wrch OpenDDS::DCPS::DcpsUpcalls::dwr_

private

Definition at line 41 of file DcpsUpcalls.h.

◆ mtx_

ACE_Thread_Mutex OpenDDS::DCPS::DcpsUpcalls::mtx_

private

Definition at line 43 of file DcpsUpcalls.h.

Referenced by svc(), and writer_done().

◆ reader_

const GUID_t& OpenDDS::DCPS::DcpsUpcalls::reader_

private

Definition at line 38 of file DcpsUpcalls.h.

Referenced by svc().

◆ reader_done_

bool OpenDDS::DCPS::DcpsUpcalls::reader_done_

private

Definition at line 42 of file DcpsUpcalls.h.

Referenced by svc().

◆ wa_

const WriterAssociation& OpenDDS::DCPS::DcpsUpcalls::wa_

private

Definition at line 39 of file DcpsUpcalls.h.

Referenced by svc().

◆ writer_done_

bool OpenDDS::DCPS::DcpsUpcalls::writer_done_

private

Definition at line 42 of file DcpsUpcalls.h.

Referenced by svc(), and writer_done().

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

Public Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ DcpsUpcalls()

Member Function Documentation

◆ svc()

◆ writer_done()

Member Data Documentation

◆ active_

◆ cnd_

◆ drr_

◆ dwr_

◆ mtx_

◆ reader_

◆ reader_done_

◆ wa_

◆ writer_done_