TAO_CosNotification: TAO_Notify_ThreadPool

TAO_BEGIN_VERSIONED_NAMESPACE_DECL TAO_Notify_ThreadPool_Task::TAO_Notify_ThreadPool_Task ( void )

Constructor.

Definition at line 18 of file ThreadPool_Task.cpp.

00019 : shutdown_ (false)
00020 {
00021 }

TAO_Notify_ThreadPool_Task::~TAO_Notify_ThreadPool_Task ( ) [virtual]

Destructor.

Definition at line 23 of file ThreadPool_Task.cpp.

00024 {
00025 }

TAO_Notify_Buffering_Strategy * TAO_Notify_ThreadPool_Task::buffering_strategy ( void )

Provide access to the underlying buffering strategy.

Definition at line 40 of file ThreadPool_Task.cpp.

References buffering_strategy_, and ACE_Auto_Basic_Ptr< X >::get().

Referenced by init().

00041 {
00042   return this->buffering_strategy_.get ();
00043 }

int TAO_Notify_ThreadPool_Task::close ( u_long flags ) [virtual]

release reference to my self.

Reimplemented from ACE_Task_Base.

Definition at line 187 of file ThreadPool_Task.cpp.

References TAO_Notify_Refcountable::_decr_refcnt().

00188 {
00189   // _incr_refcnt() for each spawned thread in init()
00190   this->_decr_refcnt();
00191   return 0;
00192 }

void TAO_Notify_ThreadPool_Task::execute ( TAO_Notify_Method_Request & method_request ) [virtual]

Queue the request.

Implements TAO_Notify_Worker_Task.

Definition at line 105 of file ThreadPool_Task.cpp.

References ACE_DEBUG, TAO_Notify_Method_Request::copy(), LM_DEBUG, shutdown_, and TAO_debug_level.

00106 {
00107   if (!shutdown_)
00108   {
00109     TAO_Notify_Method_Request_Queueable* request_copy = method_request.copy ();
00110 
00111     if (this->buffering_strategy_->enqueue (request_copy) == -1)
00112     {
00113       if (TAO_debug_level > 0)
00114         ACE_DEBUG ((LM_DEBUG, "NS_ThreadPool_Task (%P|%t) - "
00115         "failed to enqueue\n"));
00116     }
00117   }
00118 }

void TAO_Notify_ThreadPool_Task::init	(	const NotifyExt::ThreadPoolParams &	tp_params,
		const TAO_Notify_AdminProperties::Ptr &	admin_properties
	)

Activate the threadpool.

Definition at line 46 of file ThreadPool_Task.cpp.

References ACE_ASSERT, ACE_DEBUG, ACE_NEW_THROW_EX, ACE_NULL_SYNCH, ACE_TEXT(), ACE_THR_PRI_OTHER_DEF, ACE_Task_Base::activate(), buffering_strategy(), buffering_strategy_, TAO_Notify_Properties::instance(), ACE_OS::last_error(), LM_DEBUG, LM_ERROR, ACE_Task< ACE_NULL_SYNCH >::msg_queue(), TAO_Notify_Properties::orb(), ACE_Auto_Basic_Ptr< X >::reset(), TAO_Notify_Refcountable_Guard_T< T >::reset(), TAO_debug_level, THR_DETACHED, THR_NEW_LWP, timer(), and timer_.

00048 {
00049   ACE_ASSERT (this->timer_.get() == 0);
00050 
00051   TAO_Notify_Timer_Queue* timer = 0;
00052   ACE_NEW_THROW_EX (timer,
00053     TAO_Notify_Timer_Queue (),
00054     CORBA::NO_MEMORY ());
00055   this->timer_.reset (timer);
00056 
00057   TAO_Notify_Buffering_Strategy* buffering_strategy = 0;
00058   ACE_NEW_THROW_EX (buffering_strategy,
00059     TAO_Notify_Buffering_Strategy (*msg_queue (), admin_properties),
00060     CORBA::NO_MEMORY ());
00061   this->buffering_strategy_.reset (buffering_strategy);
00062 
00063   long flags = THR_NEW_LWP | THR_DETACHED;
00064   CORBA::ORB_var orb =
00065     TAO_Notify_PROPERTIES::instance()->orb ();
00066 
00067   flags |=
00068     orb->orb_core ()->orb_params ()->thread_creation_flags ();
00069 
00070   // Guards the thread for auto-deletion; paired with close.
00071   // This is done in the originating thread before the spawn to
00072   // avoid any race conditions.
00073   for ( CORBA::ULong i = 0; i < tp_params.static_threads; ++i )
00074     {
00075       this->_incr_refcnt();
00076     }
00077 
00078   // Become an active object.
00079   if (this->ACE_Task <ACE_NULL_SYNCH>::activate (flags,
00080     tp_params.static_threads,
00081     0,
00082     ACE_THR_PRI_OTHER_DEF) == -1)
00083   {
00084     // Undo the ref counts on error
00085     for ( CORBA::ULong i = 0; i < tp_params.static_threads; ++i )
00086       {
00087         this->_decr_refcnt();
00088       }
00089 
00090     if (TAO_debug_level > 0)
00091     {
00092       if (ACE_OS::last_error () == EPERM)
00093         ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Insufficient privilege.\n")));
00094       else
00095         ACE_DEBUG ((LM_ERROR,
00096         ACE_TEXT ("(%t) task activation at priority %d failed\n")
00097         ACE_TEXT ("exiting!\n%a"),
00098         tp_params.default_priority));
00099     }
00100     throw CORBA::BAD_PARAM ();
00101   }
00102 }

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

Call the base class init.

Definition at line 28 of file ThreadPool_Task.cpp.

References ACE_Shared_Object::init().

Referenced by TAO_Notify_Builder::apply_thread_pool_concurrency().

00029 {
00030   return this->ACE_Task<ACE_NULL_SYNCH>::init (argc, argv);
00031 }

void TAO_Notify_ThreadPool_Task::release ( void ) [private, virtual]

Release.

Implements TAO_Notify_Refcountable.

Definition at line 181 of file ThreadPool_Task.cpp.

00182 {
00183   delete this;
00184 }

void TAO_Notify_ThreadPool_Task::shutdown ( ) [virtual]

Shutdown task.

Implements TAO_Notify_Worker_Task.

Definition at line 168 of file ThreadPool_Task.cpp.

References buffering_strategy_, and shutdown_.

00169 {
00170   if (this->shutdown_)
00171   {
00172     return;
00173   }
00174 
00175   this->shutdown_ = true;
00176 
00177   this->buffering_strategy_->shutdown ();
00178 }

int TAO_Notify_ThreadPool_Task::svc ( void ) [protected, virtual]

Task svc.

Reimplemented from ACE_Task_Base.

Definition at line 121 of file ThreadPool_Task.cpp.

References CORBA::Exception::_tao_print_exception(), ACE_DEBUG, buffering_strategy_, TAO_Notify_Method_Request::execute(), LM_DEBUG, ACE_Message_Block::release(), shutdown_, TAO_debug_level, and timer_.

00122 {
00123   TAO_Notify_Method_Request_Queueable* method_request = 0;
00124 
00125   while (!shutdown_)
00126   {
00127     try
00128     {
00129       ACE_Time_Value* dequeue_blocking_time = 0;
00130       ACE_Time_Value earliest_time;
00131 
00132       if (!this->timer_->impl().is_empty ())
00133       {
00134         earliest_time = this->timer_->impl().earliest_time ();
00135         dequeue_blocking_time = &earliest_time;
00136       }
00137 
00138       // Dequeue 1 item
00139       int const result = buffering_strategy_->dequeue (method_request, dequeue_blocking_time);
00140 
00141       if (result > 0)
00142       {
00143         method_request->execute ();
00144 
00145         ACE_Message_Block::release (method_request);
00146       }
00147       else if (errno == ETIME)
00148       {
00149         this->timer_->impl ().expire ();
00150       }
00151       else
00152       {
00153         if (TAO_debug_level > 0)
00154           ACE_DEBUG ((LM_DEBUG, "ThreadPool_Task dequeue failed\n"));
00155       }
00156     }
00157     catch (const CORBA::Exception& ex)
00158     {
00159       ex._tao_print_exception (
00160         "ThreadPool_Task (%P|%t) exception in method request\n");
00161     }
00162   } /* while */
00163 
00164   return 0;
00165 }

TAO_Notify_Timer * TAO_Notify_ThreadPool_Task::timer ( void ) [virtual]

The object used by clients to register timers.

Implements TAO_Notify_Worker_Task.

Definition at line 34 of file ThreadPool_Task.cpp.

References TAO_Notify_Refcountable_Guard_T< T >::get(), and timer_.

Referenced by init().

00035 {
00036   return this->timer_.get();
00037 }

void TAO_Notify_ThreadPool_Task::update_qos_properties ( const TAO_Notify_QoSProperties & qos_properties ) [virtual]

Update QoS Properties.

Reimplemented from TAO_Notify_Worker_Task.

Definition at line 195 of file ThreadPool_Task.cpp.

References buffering_strategy_.

00196 {
00197   this->buffering_strategy_->update_qos_properties (qos_properties);
00198 }

friend class TAO_Notify_Method_Request_Shutdown [friend]

Definition at line 45 of file ThreadPool_Task.h.

ACE_Auto_Ptr< TAO_Notify_Buffering_Strategy > TAO_Notify_ThreadPool_Task::buffering_strategy_ [private]

The buffering strategy to use.

Definition at line 87 of file ThreadPool_Task.h.

Referenced by buffering_strategy(), init(), shutdown(), svc(), and update_qos_properties().

bool TAO_Notify_ThreadPool_Task::shutdown_ [private]

Shutdown.

Definition at line 90 of file ThreadPool_Task.h.

Referenced by execute(), shutdown(), and svc().

TAO_Notify_Timer_Queue::Ptr TAO_Notify_ThreadPool_Task::timer_ [private]

The Queue based timer.

Definition at line 93 of file ThreadPool_Task.h.

Referenced by init(), svc(), and timer().

TAO_Notify_ThreadPool_Task Class Reference

Public Member Functions

Protected Member Functions

Private Member Functions

Private Attributes

Friends

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation

Member Data Documentation


Public Member Functions
	TAO_Notify_ThreadPool_Task (void)
	Constructor.
virtual	~TAO_Notify_ThreadPool_Task ()
	Destructor.
virtual int	init (int argc, ACE_TCHAR **argv)
	Call the base class init.
virtual int	close (u_long flags)
	release reference to my self.
void	init (const NotifyExt::ThreadPoolParams &tp_params, const TAO_Notify_AdminProperties::Ptr &admin_properties)
	Activate the threadpool.
virtual void	execute (TAO_Notify_Method_Request &method_request)
	Queue the request.
virtual void	shutdown ()
	Shutdown task.
virtual void	update_qos_properties (const TAO_Notify_QoSProperties &qos_properties)
	Update QoS Properties.
virtual TAO_Notify_Timer *	timer (void)
	The object used by clients to register timers.
TAO_Notify_Buffering_Strategy *	buffering_strategy (void)
	Provide access to the underlying buffering strategy.
Protected Member Functions
virtual int	svc (void)
	Task svc.
Private Member Functions
virtual void	release (void)
	Release.
Private Attributes
ACE_Auto_Ptr< TAO_Notify_Buffering_Strategy >	buffering_strategy_
	The buffering strategy to use.
bool	shutdown_
	Shutdown.
TAO_Notify_Timer_Queue::Ptr	timer_
	The Queue based timer.
Friends
class	TAO_Notify_Method_Request_Shutdown