OpenDDS::DCPS::QueueTaskBase< T > Class Template Reference

A simple ACE task that manages a queue of request. More...

#include <QueueTaskBase_T.h>

Inheritance diagram for OpenDDS::DCPS::QueueTaskBase< T >:

[legend]Collaboration diagram for OpenDDS::DCPS::QueueTaskBase< T >:

[legend]List of all members.

Public Member Functions
	QueueTaskBase ()
	Constructor.
virtual	~QueueTaskBase ()
	Virtual Destructor.
int	add (const T &req)
virtual int	open (void *=0)
	Activate the worker threads.
virtual int	svc ()
	The "mainline" executed by the worker thread.
virtual int	close (u_long flag=0)
	Called when the thread exits.
virtual void	execute (T &req)=0
Private Types
typedef ACE_SYNCH_MUTEX	LockType
typedef ACE_Guard< LockType >	GuardType
typedef ACE_Condition< LockType >	ConditionType
typedef ACE_Unbounded_Queue< T >	Queue
Private Attributes
LockType	lock_
	Lock to protect the "state" (all of the data members) of this object.
Queue	queue_
	The request queue.
ConditionType	work_available_
bool	shutdown_initiated_
	Flag used to initiate a shutdown request to all worker threads.
bool	opened_
	Flag used to avoid multiple open() calls.
ACE_thread_t	thr_id_
	The id of the thread created by this task.

---

Detailed Description

template<typename T>
class OpenDDS::DCPS::QueueTaskBase< T >

A simple ACE task that manages a queue of request.

Definition at line 34 of file QueueTaskBase_T.h.

---

Member Typedef Documentation

template<typename T>

typedef ACE_Condition<LockType> OpenDDS::DCPS::QueueTaskBase< T >::ConditionType [private]

Definition at line 175 of file QueueTaskBase_T.h.

template<typename T>

typedef ACE_Guard<LockType> OpenDDS::DCPS::QueueTaskBase< T >::GuardType [private]

Definition at line 174 of file QueueTaskBase_T.h.

template<typename T>

typedef ACE_SYNCH_MUTEX OpenDDS::DCPS::QueueTaskBase< T >::LockType [private]

Definition at line 173 of file QueueTaskBase_T.h.

template<typename T>

typedef ACE_Unbounded_Queue<T> OpenDDS::DCPS::QueueTaskBase< T >::Queue [private]

Definition at line 177 of file QueueTaskBase_T.h.

---

Constructor & Destructor Documentation

template<typename T>

OpenDDS::DCPS::QueueTaskBase< T >::QueueTaskBase

(

)

[inline]

Constructor.

Definition at line 38 of file QueueTaskBase_T.h.

  : work_available_(lock_),
      shutdown_initiated_(false),
      opened_(false),
      thr_id_(ACE_OS::NULL_thread) {
    DBG_ENTRY("QueueTaskBase","QueueTaskBase");
  }

template<typename T>

virtual OpenDDS::DCPS::QueueTaskBase< T >::~QueueTaskBase

(

)

[inline, virtual]

Virtual Destructor.

Definition at line 47 of file QueueTaskBase_T.h.

                           {
    DBG_ENTRY("QueueTaskBase","~QueueTaskBase");
  }

---

Member Function Documentation

template<typename T>

int OpenDDS::DCPS::QueueTaskBase< T >::add

(

const T &

req

)

[inline]

Put the request to the request queue. Returns 0 if successful, -1 otherwise (it has been "rejected" or this task is shutdown).

Definition at line 54 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::TcpConnection::handle_close(), OpenDDS::DCPS::TcpTransport::passive_connection(), OpenDDS::DCPS::TransportImpl::release_link_resources(), and OpenDDS::DCPS::TcpConnection::relink_from_send().

                        {
    DBG_ENTRY("QueueTaskBase","add");
    GuardType guard(this->lock_);

    if (this->shutdown_initiated_)
      return -1;

    int result = this->queue_.enqueue_tail(req);

    if (result == 0) {
      this->work_available_.signal();

    } else
      ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: QueueTaskBase::add %p\n",
       ACE_TEXT("enqueue_tail")));

    return result;
  }

template<typename T>

virtual int OpenDDS::DCPS::QueueTaskBase< T >::close

(

u_long

flag = 0

)

[inline, virtual]

Called when the thread exits.

Definition at line 144 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::TransportImpl::shutdown(), OpenDDS::DCPS::TcpConnection::shutdown(), OpenDDS::DCPS::TcpTransport::shutdown_i(), OpenDDS::DCPS::TcpConnection::transfer(), OpenDDS::DCPS::TcpConnection::~TcpConnection(), and OpenDDS::DCPS::TcpTransport::~TcpTransport().

                                     {
    DBG_ENTRY("QueueTaskBase","close");

    if (flag == 0)
      return 0;

    {
      GuardType guard(this->lock_);

      if (this->shutdown_initiated_)
        return 0;

      // Set the shutdown flag to true.
      this->shutdown_initiated_ = true;
      this->work_available_.signal();
    }

    if (this->opened_ && !ACE_OS::thr_equal(this->thr_id_, ACE_OS::thr_self()))
      this->wait();

    return 0;
  }

template<typename T>

virtual void OpenDDS::DCPS::QueueTaskBase< T >::execute

(

T &

req

)

[pure virtual]

The subclass should implement this function to handle the dequeued request.

Implemented in OpenDDS::DCPS::TcpReconnectTask.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::svc().

template<typename T>

virtual int OpenDDS::DCPS::QueueTaskBase< T >::open

(

void *

= 0

)

[inline, virtual]

Activate the worker threads.

Definition at line 74 of file QueueTaskBase_T.h.

                              {
    DBG_ENTRY("QueueTaskBase","open");

    GuardType guard(this->lock_);

    // We can assume that we are in the proper state to handle this open()
    // call as long as we haven't been open()'ed before.
    if (this->opened_) {
      ACE_ERROR_RETURN((LM_ERROR,
                        "(%P|%t) QueueTaskBase failed to open.  "
                        "Task has previously been open()'ed.\n"),
                       -1);
    }

    // Activate this task object with one worker thread.
    if (this->activate(THR_NEW_LWP | THR_JOINABLE, 1) != 0) {
      // Assumes that when activate returns non-zero return code that
      // no threads were activated.
      ACE_ERROR_RETURN((LM_ERROR,
                        "(%P|%t) QueueTaskBase failed to activate "
                        "the worker threads.\n"),
                       -1);
    }

    // Now we have past the point where we can say we've been open()'ed before.
    this->opened_ = true;

    return 0;
  }

template<typename T>

virtual int OpenDDS::DCPS::QueueTaskBase< T >::svc

(

)

[inline, virtual]

The "mainline" executed by the worker thread.

Definition at line 105 of file QueueTaskBase_T.h.

                    {
    DBG_ENTRY("QueueTaskBase","svc");

    this->thr_id_ = ACE_OS::thr_self();

    // Start the "GetWork-And-PerformWork" loop for the current worker thread.
    while (!this->shutdown_initiated_) {
      T req;
      {
        GuardType guard(this->lock_);

        if (this->queue_.is_empty()) {
          this->work_available_.wait();
        }

        if (this->shutdown_initiated_)
          break;

        int result = queue_.dequeue_head(req);

        if (result != 0) {
          //I'm not sure why this thread got more signals than actual signals
          //when using thread_per_connection and the user application thread
          //send requests without interval. We just need ignore the dequeue
          //failure.
          //ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: QueueTaskBase::svc  %p\n",
          //  ACE_TEXT("dequeue_head")));
          continue;
        }
      }

      this->execute(req);
    }

    // This will never get executed.
    return 0;
  }

---

Member Data Documentation

template<typename T>

LockType OpenDDS::DCPS::QueueTaskBase< T >::lock_ [private]

Lock to protect the "state" (all of the data members) of this object.

Definition at line 180 of file QueueTaskBase_T.h.

template<typename T>

bool OpenDDS::DCPS::QueueTaskBase< T >::opened_ [private]

Flag used to avoid multiple open() calls.

Definition at line 195 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::open().

template<typename T>

Queue OpenDDS::DCPS::QueueTaskBase< T >::queue_ [private]

The request queue.

Definition at line 183 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::add(), and OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::svc().

template<typename T>

bool OpenDDS::DCPS::QueueTaskBase< T >::shutdown_initiated_ [private]

Flag used to initiate a shutdown request to all worker threads.

Definition at line 192 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::close().

template<typename T>

ACE_thread_t OpenDDS::DCPS::QueueTaskBase< T >::thr_id_ [private]

The id of the thread created by this task.

Definition at line 198 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::svc().

template<typename T>

ConditionType OpenDDS::DCPS::QueueTaskBase< T >::work_available_ [private]

Condition used to signal the worker threads that they may be able to find a request in the queue_ that needs to be executed. This condition will be signal()'ed each time a request is added to the queue_, and also when this task is shutdown.

Definition at line 189 of file QueueTaskBase_T.h.

Referenced by OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::add(), and OpenDDS::DCPS::QueueTaskBase< ReconnectOpType >::close().

---

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

• QueueTaskBase_T.h

---