Implements a Thread Pool Worker Task. More...

#include <ThreadPool_Task.h>

Inheritance diagram for TAO_Notify_ThreadPool_Task:

Collaboration diagram for TAO_Notify_ThreadPool_Task:

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

Detailed Description

Implements a Thread Pool Worker Task.

Definition at line 41 of file ThreadPool_Task.h.

Constructor & Destructor Documentation

TAO_Notify_ThreadPool_Task::TAO_Notify_ThreadPool_Task ( void )

Constructor.

Definition at line 18 of file ThreadPool_Task.cpp.

: shutdown_ (false)
{
}

TAO_Notify_ThreadPool_Task::~TAO_Notify_ThreadPool_Task ( ) [virtual]

Destructor.

Definition at line 23 of file ThreadPool_Task.cpp.

{
}

Member Function Documentation

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.

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

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.

{
  // _incr_refcnt() for each spawned thread in init()
  this->_decr_refcnt();
  return 0;
}

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

Queue the request.

Definition at line 105 of file ThreadPool_Task.cpp.

{
  if (!shutdown_)
  {
    TAO_Notify_Method_Request_Queueable* request_copy = method_request.copy ();

    if (this->buffering_strategy_->enqueue (request_copy) == -1)
    {
      if (TAO_debug_level > 0)
        ACE_DEBUG ((LM_DEBUG, "NS_ThreadPool_Task (%P|%t) - "
        "failed to enqueue\n"));
    }
  }
}

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.

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

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.

{
  ACE_ASSERT (this->timer_.get() == 0);

  TAO_Notify_Timer_Queue* timer = 0;
  ACE_NEW_THROW_EX (timer,
    TAO_Notify_Timer_Queue (),
    CORBA::NO_MEMORY ());
  this->timer_.reset (timer);

  TAO_Notify_Buffering_Strategy* buffering_strategy = 0;
  ACE_NEW_THROW_EX (buffering_strategy,
    TAO_Notify_Buffering_Strategy (*msg_queue (), admin_properties),
    CORBA::NO_MEMORY ());
  this->buffering_strategy_.reset (buffering_strategy);

  long flags = THR_NEW_LWP | THR_DETACHED;
  CORBA::ORB_var orb =
    TAO_Notify_PROPERTIES::instance()->orb ();

  flags |=
    orb->orb_core ()->orb_params ()->thread_creation_flags ();

  // Guards the thread for auto-deletion; paired with close.
  // This is done in the originating thread before the spawn to
  // avoid any race conditions.
  for ( CORBA::ULong i = 0; i < tp_params.static_threads; ++i )
    {
      this->_incr_refcnt();
    }

  // Become an active object.
  if (this->ACE_Task <ACE_NULL_SYNCH>::activate (flags,
    tp_params.static_threads,
    0,
    ACE_THR_PRI_OTHER_DEF) == -1)
  {
    // Undo the ref counts on error
    for ( CORBA::ULong i = 0; i < tp_params.static_threads; ++i )
      {
        this->_decr_refcnt();
      }

    if (ACE_OS::last_error () == EPERM)
      ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%P|%t) Insufficient privilege.\n")));
    else if (ACE_OS::last_error () == EAGAIN)
    {
      ACE_DEBUG ((LM_DEBUG,
      ACE_TEXT ("(%P|%t) task activation at priority %d failed %p\n"),
      tp_params.default_priority, "activate"));
      throw CORBA::NO_RESOURCES ();
    }

    throw CORBA::BAD_PARAM ();
  }
}

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

Release.

Definition at line 181 of file ThreadPool_Task.cpp.

{
  delete this;
}

void TAO_Notify_ThreadPool_Task::shutdown ( void ) [virtual]

Shutdown task.

Definition at line 168 of file ThreadPool_Task.cpp.

{
  if (this->shutdown_)
  {
    return;
  }

  this->shutdown_ = true;

  this->buffering_strategy_->shutdown ();
}

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.

{
  TAO_Notify_Method_Request_Queueable* method_request = 0;

  while (!shutdown_)
  {
    try
    {
      ACE_Time_Value* dequeue_blocking_time = 0;
      ACE_Time_Value earliest_time;

      if (!this->timer_->impl().is_empty ())
      {
        earliest_time = this->timer_->impl().earliest_time ();
        dequeue_blocking_time = &earliest_time;
      }

      // Dequeue 1 item
      int const result = buffering_strategy_->dequeue (method_request, dequeue_blocking_time);

      if (result > 0)
      {
        method_request->execute ();

        ACE_Message_Block::release (method_request);
      }
      else if (errno == ETIME)
      {
        this->timer_->impl ().expire ();
      }
      else
      {
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG, "(%P|%t)ThreadPool_Task dequeue failed\n"));
      }
    }
    catch (const CORBA::Exception& ex)
    {
      ex._tao_print_exception (
        "ThreadPool_Task (%P|%t) exception in method request\n");
    }
  } /* while */

  return 0;
}

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

The object used by clients to register timers.

Definition at line 34 of file ThreadPool_Task.cpp.

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

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

Update QoS Properties.

Definition at line 195 of file ThreadPool_Task.cpp.

{
  this->buffering_strategy_->update_qos_properties (qos_properties);
}

Friends And Related Function Documentation

friend class TAO_Notify_Method_Request_Shutdown [friend]

Definition at line 45 of file ThreadPool_Task.h.

Member Data Documentation

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.

bool TAO_Notify_ThreadPool_Task::shutdown_ [private]

Shutdown.

Definition at line 90 of file ThreadPool_Task.h.

TAO_Notify_Timer_Queue::Ptr TAO_Notify_ThreadPool_Task::timer_ [private]

The Queue based timer.

Definition at line 93 of file ThreadPool_Task.h.

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

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