dispatching_modules.h

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

/* -*- C++ -*- */
//=============================================================================
/**
 *  @file Dispatching_Modules.h
 *
 *  Dispatching_Modules.h,v 1.27 2003/07/21 06:42:27 jwillemsen Exp
 *
 *  @author Tim Harrison ([e-mail])
 *
 *  This file holds the different Event Service dispatching
 *  mechanisms.  These include null-dispatching (EFD),
 *  single-threaded with (RTU) and without preemption (LAME), and a
 *  multithreaded implementation.
 *
 *
 */
//=============================================================================


#ifndef ACE_DISPATCHING_MODULES_H
#define ACE_DISPATCHING_MODULES_H
#include /**/ "ace/pre.h"

#include "ace/Reactor_Notification_Strategy.h"
#include "tao/Timeprobe.h"
#include "orbsvcs/Event/ReactorTask.h"
#include "orbsvcs/Event/Event_Manip.h"
#include "orbsvcs/Event/Event_Channel.h"

// ************************************************************

// Forward declarations.
class ACE_ES_Dispatch_Queue;
class ACE_ES_Dispatch_Request;

// ************************************************************

// Forward declarations.
class ACE_ES_Consumer_Module;

/**
 * @class ACE_ES_Dispatching_Base
 *
 * @brief Event Service Dispatch Module base class
 *
 * We inherit from ACE_Event_Handler so that we can be called back
 * by the ReactorEx when requests are queued.  The virtual
 * dispatch_event method allows ACE_ES_Dispatch_Requests to call
 * back the dispatching module when acting as command objects. When
 * this implementation is used by the Event Channel it forwards all
 * dispatch calls without any queuing.  Therefore, it can be
 * used to build an EFD.  It is also inherited by the Priority
 * Dispatching module.
 */
class TAO_RTOLDEvent_Export ACE_ES_Dispatching_Base : public ACE_Event_Handler
{
public:
  /// Default construction.
  ACE_ES_Dispatching_Base (ACE_EventChannel *channel);

  /// Link to adjacent modules.
  virtual void open (ACE_ES_Consumer_Module *up,
                     ACE_ES_Correlation_Module *down);

  /// Forward down_.
  virtual void connected (ACE_Push_Consumer_Proxy *consumer
                          ACE_ENV_ARG_DECL_NOT_USED);

  /// Forward down_.
  virtual void disconnecting (ACE_Push_Consumer_Proxy *consumer
                              ACE_ENV_ARG_DECL_NOT_USED);

  /// Release any unneeded dispatching resources.
  virtual void disconnected (ACE_Push_Consumer_Proxy *consumer);

  // = Not needed.
  // void connected (ACE_Push_Supplier_Proxy *supplier);
  // void disconnecting (ACE_Push_Supplier_Proxy *supplier);

  /// Forward up_.
  virtual void push (ACE_ES_Dispatch_Request *request
                     ACE_ENV_ARG_DECL_NOT_USED) = 0;

  /// Called by ACE_ES_Dispatch_Requests when dequeued by RT_Tasks.
  virtual int dispatch_event (ACE_ES_Dispatch_Request *request,
                              u_long &command_action);

  /// Called when all the threads of a <q> have exited.
  virtual void dispatch_queue_closed (ACE_ES_Dispatch_Queue *q);

  /// This is called by the Event Channel. It will create all the
  /// threads and only return once they are all up and running.
  virtual void activate (int threads_per_queue);

  /// This is called by the Event Channel.  This will attempt to shut
  /// down all of its threads gracefully.  Wish it luck.
  virtual void shutdown (void);

protected:
  /// Dat der channel.
  ACE_EventChannel *channel_;

  /// To synchronize thr_count_.
  ACE_ES_MUTEX lock_;

  /// The total number of threads in the Dispatching Module.  This will
  /// be the sum of all the Dispatch Queue threads.
  int thr_count_;

  /// Next module up.
  ACE_ES_Consumer_Module *up_;

  /// Next module down.
  ACE_ES_Correlation_Module *down_;
};

// ************************************************************

/**
 * @class ACE_ES_Dispatch_Request
 *
 * @brief ACE Event Service Dispatch Request
 *
 * Encapsulates a consumer and the events that will be sent to the
 * consumer.  Right now, this class keeps a single_event_ that can
 * be used when only one event is sent to the consumer.  Since this
 * is frequently the case (except for correlations), this
 * optimization reduces the amount of dynamic memory allocation is
 * necessary.  This class is also a GOF Command object since it can
 * be dequeued by an RT_Task to call back the dispatching module
 * for request dispatching.
 */
class TAO_RTOLDEvent_Export ACE_ES_Dispatch_Request : public ACE_RT_Task_Command
{
public:
  /// Default construction.
  ACE_ES_Dispatch_Request (void);

  /// Default destruction.
  virtual ~ACE_ES_Dispatch_Request (void);

  /// All the events must be added after construction to the
  /// event_set.
  ACE_ES_Dispatch_Request (ACE_Push_Consumer_Proxy *consumer,
                           RtecScheduler::handle_t rt_info);

  /// Set consumer_ to <consumer> and copy <event_set> to event_set_.
  /// <rt_info> describes the method receiving this dispatch.
  ACE_ES_Dispatch_Request (ACE_Push_Consumer_Proxy *consumer,
                           const TAO_EC_Event_Array &event_set,
                           RtecScheduler::handle_t rt_info);

  /**
   * Set consumer_ to <consumer> and sets
   * single_event_.header.creation_time to <time>.  Sets
   * use_single_event_ to 1.  <rt_info> describes the method
   * receiving this dispatch.
   */
  ACE_ES_Dispatch_Request (ACE_Push_Consumer_Proxy *consumer,
                           const RtecEventComm::Time &time,
                           RtecScheduler::handle_t rt_info);

  /**
   * Sets consumer_ and the first slot of event_set_.  We use the
   * event_set_ instead of the single_event_ so that we can just carry
   * around the pointer to <event>.  <rt_info> describes the method
   * receiving this dispatch.
   */
  ACE_ES_Dispatch_Request (ACE_Push_Consumer_Proxy *consumer,
                           const TAO_EC_Event &event,
                           RtecScheduler::handle_t rt_info);

  /// Description of the method receiving this request.
  RtecScheduler::handle_t rt_info (void);

  /**
   * For multi-threaded implementations, <dispatching_module> is
   * called back when a request is dequeued.  <priority> is the
   * dispatch priority of the event.  <sub_priority> is the enqueue
   * priority of the event and will be forwarded to
   * ACE_Message_Block.
   */
  void set (ACE_ES_Dispatching_Base *dispatching_module,
            RtecScheduler::OS_Priority priority,
            RtecScheduler::Preemption_Subpriority_t sub_priority);

  /// Consumer accessor.
  ACE_Push_Consumer_Proxy *consumer (void) const;

  /// If accessed, make_copy will use event_set_.
  const TAO_EC_Event_Array &event_set (void) const;

  /// If accessed, make_copy will use event_set_.
  TAO_EC_Event_Array &event_set (void);

  /// Append an event to the list of events in the Request.
  void append_event (const TAO_EC_Event& event);

  /// Returns 1 if we're using single_event, or event_set_.size ().
  CORBA::ULong number_of_events (void) const;

  /// Copy single_event or event_set into <dest>.
  /// @@ Change the name to something more meaningful...
  void make_copy (RtecEventComm::EventSet &dest) const;

  /// Calls dispatching_module_->dispatch_event.
  virtual int execute (u_long &command_action);

  /// Priority accessor.
  RtecScheduler::OS_Priority priority (void);

#if 0
  // @@ Memory pools
  // @@ This cannot be done: the object would be allocated using this
  // class operator new, but it will be removed using the
  // ACE_Message_Block operator delete!
  /// Allocates memory from a thread-specific memory pool.
  void *operator new (size_t nbytes);

  /// Returns memory to a thread-specific memory pool.
  void operator delete (void *);
#endif

protected:
  RtecScheduler::OS_Priority priority_;

  /// Describes the method receiving this dispatch.
  RtecScheduler::handle_t rt_info_;

  /// The dispatching module called back when we're dequeued by a
  /// thread.
  ACE_ES_Dispatching_Base *dispatching_module_;

  /// Is true if we're using a single event.  Is 0 is we're using
  /// event_set_.
  int use_single_event_;

  /// The final destination for single_event_ or event_set_.
  ACE_Push_Consumer_Proxy *consumer_;

  /// This is used for single event dispatches.
  TAO_EC_Event single_event_;

  /// This is used for event sets that need to be dispatched.
  TAO_EC_Event_Array event_set_;
};

// ************************************************************

#if defined (ACE_WIN32)
/**
 * @class ACE_ES_ReactorEx_NS
 *
 * @brief Event Service ReactorEx Notification Strategy
 *
 * Integrates the ACE_Message_Queue notification to signal a
 * handle that will wake up the ACE_ES_Priority_Dispatching
 * module.  This is used in place of the
 * ACE_ReactorEx_Notification_Strategy to avoid any queueing by
 * the ReactorEx::notify mechanism.
 */
class TAO_RTOLDEvent_Export ACE_ES_ReactorEx_NS : public ACE_Notification_Strategy
{
public:
  /// Stores away <eh> for when this->open is called.
  ACE_ES_ReactorEx_NS (ACE_Event_Handler *eh,
                       TAO_EC_Timer_Module *tm);

  /// Registers eh_ with the ReactorEx to be notified when this->notify
  /// is called.