lb_loadmanager.cpp

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

#include "LB_LoadManager.h"
#include "LB_MemberLocator.h"
#include "LB_LoadAlert_Handler.h"
#include "LB_RoundRobin.h"
#include "LB_Random.h"
#include "LB_LeastLoaded.h"
#include "LB_conf.h"

#include "orbsvcs/PortableGroup/PG_Property_Utils.h"
#include "orbsvcs/PortableGroup/PG_conf.h"

#include "tao/Messaging/Messaging.h"
#include "tao/debug.h"
#include "tao/ORB_Constants.h"

#include "ace/Reactor.h"
#include "ace/OS_NS_sys_time.h"
#include "ace/Reverse_Lock_T.h"
#include "ace/OS_NS_stdio.h"

ACE_RCSID (LoadBalancing,
           LB_LoadManager,
           "LB_LoadManager.cpp,v 1.20 2003/12/30 00:10:17 dhinton Exp")


TAO_LB_LoadManager::TAO_LB_LoadManager (void)
  : reactor_ (0),
    poa_ (),
    root_poa_ (),
    monitor_lock_ (),
    load_lock_ (),
    load_alert_lock_ (),
    lock_ (),
    monitor_map_ (TAO_PG_MAX_LOCATIONS),
    load_map_ (TAO_PG_MAX_LOCATIONS),
    load_alert_map_ (TAO_PG_MAX_LOCATIONS),
    object_group_manager_ (),
    property_manager_ (object_group_manager_),
    generic_factory_ (object_group_manager_, property_manager_),
    pull_handler_ (),
    timer_id_ (-1),
    lm_ref_ (),
    round_robin_ (),
    random_ (),
    least_loaded_ (),
    built_in_balancing_strategy_info_name_ (1),
    built_in_balancing_strategy_name_ (1),
    custom_balancing_strategy_name_ (1)
{
  this->pull_handler_.initialize (&this->monitor_map_, this);

  // @note "this->init()" is not called here (in the constructor)
  //       since it may thrown an exception.  Throwing an exception in
  //       a constructor in an emulated exception environment is
  //       problematic since native exception semantics cannot be
  //       reproduced in such a case.  As such, init() must be called
  //       by whatever code instantiates this LoadManager.
}

TAO_LB_LoadManager::~TAO_LB_LoadManager (void)
{
}

void
TAO_LB_LoadManager::push_loads (
    const PortableGroup::Location & the_location,
    const CosLoadBalancing::LoadList & loads
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  if (loads.length () == 0)
    ACE_THROW (CORBA::BAD_PARAM ());

  {
    ACE_GUARD (TAO_SYNCH_MUTEX,
               guard,
               this->load_lock_);

    if (this->load_map_.rebind (the_location, loads) == -1)
      ACE_THROW (CORBA::INTERNAL ());
  }

  // Analyze loads for object groups that have members residing at the
  // given location.
  PortableGroup::ObjectGroups_var groups =
    this->object_group_manager_.groups_at_location (the_location
                                                    ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  const CORBA::ULong len = groups->length ();

  for (CORBA::ULong i = 0; i < len; ++i)
    {
      PortableGroup::ObjectGroup_ptr object_group =
        groups[i].in ();

      ACE_TRY
        {
          PortableGroup::Properties_var properties =
            this->get_properties (object_group
                                  ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          PortableGroup::Value value;
          CosLoadBalancing::Strategy_ptr strategy;

          if ((TAO_PG::get_property_value (
                 this->built_in_balancing_strategy_name_,
                 properties.in (),
                 value)
               || TAO_PG::get_property_value (
                    this->custom_balancing_strategy_name_,
                    properties.in (),
                    value))
              && (value >>= strategy)
              && !CORBA::is_nil (strategy))
            {
              strategy->analyze_loads (object_group,
                                       this->lm_ref_.in ()
                                       ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK;
            }
        }
      ACE_CATCHANY
        {
          // Ignore all exceptions.
        }
      ACE_ENDTRY;
      ACE_CHECK;
    }
}

CosLoadBalancing::LoadList *
TAO_LB_LoadManager::get_loads (const PortableGroup::Location & the_location
                               ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   CosLoadBalancing::LocationNotFound))
{
  CosLoadBalancing::LoadList * tmp;
  ACE_NEW_THROW_EX (tmp,
                    CosLoadBalancing::LoadList,
                    CORBA::NO_MEMORY (
                      CORBA::SystemException::_tao_minor_code (
                        TAO_DEFAULT_MINOR_CODE,
                        ENOMEM),
                      CORBA::COMPLETED_NO));
  ACE_CHECK_RETURN (0);

  CosLoadBalancing::LoadList_var loads = tmp;

  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX,
                    guard,
                    this->load_lock_,
                    0);

  if (this->load_map_.find (the_location, *tmp) == 0)
    return loads._retn ();
  else
    ACE_THROW_RETURN (CosLoadBalancing::LocationNotFound (), 0);
}

void
TAO_LB_LoadManager::enable_alert (const PortableGroup::Location & the_location
                                  ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CosLoadBalancing::LoadAlertNotFound))
{
  ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->load_alert_lock_);

  TAO_LB_LoadAlertMap::ENTRY * entry;
  if (this->load_alert_map_.find (the_location, entry) == 0)
    {
      TAO_LB_LoadAlertInfo & info = entry->int_id_;

      // @note This could be problematic if the LoadAlert object is
      //       registered with more than LoadManager.

      if (info.alerted == 1)
        return;  // No need to set the alert status.  It has already
                 // been set.

      // Duplicate before releasing the LoadAlertMap lock to prevent a
      // race condition from occuring.  The LoadAlertInfo map may be
      // altered prior to invoking an operation on the LoadAlert
      // object.
      CosLoadBalancing::LoadAlert_var load_alert =
        CosLoadBalancing::LoadAlert::_duplicate (info.load_alert.in ());

      // The alert condition will be enabled.
      // @@ What happens if the below call fails?  This variable
      //    should be reset to zero!
      info.alerted = 1;

      {
        // Release the lock prior to making the below remote
        // invocation.
        ACE_Reverse_Lock<TAO_SYNCH_MUTEX> reverse_lock (
          this->load_alert_lock_);
        ACE_GUARD (ACE_Reverse_Lock<TAO_SYNCH_MUTEX>,
                   reverse_guard,
                   reverse_lock);

        // Use AMI to make the following operation "non-blocking,"
        // allowing the caller to continue without being forced to
        // wait for a response.
        //
        // AMI is used to improve member selection times and overall
        // throughput since the LoadAlert object need not be alerted
        // synchronously.  In particular, the load alert can and
        // should be performed in parallel to other tasks, such as
        // member selection.
        load_alert->sendc_enable_alert (this->load_alert_handler_.in ()
                                        ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;
      }
    }
  else
    ACE_THROW (CosLoadBalancing::LoadAlertNotFound ());
}

void
TAO_LB_LoadManager::disable_alert (const PortableGroup::Location & the_location
                                   ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CosLoadBalancing::LoadAlertNotFound))
{
  ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->load_alert_lock_);

  TAO_LB_LoadAlertMap::ENTRY * entry;
  if (this->load_alert_map_.find (the_location, entry) == 0)
    {
      TAO_LB_LoadAlertInfo & info = entry->int_id_;

      // @note This could be problematic if the LoadAlert object is
      //       registered with more than LoadManager.
      if (info.alerted == 0)
        return;  // No need to set the alert status.  It has already
                 // been set.

      // Duplicate before releasing the LoadAlertMap lock to prevent a
      // race condition from occuring.  The LoadAlertInfo map may be
      // altered prior to invoking an operation on the LoadAlert
      // object.
      CosLoadBalancing::LoadAlert_var load_alert =
        CosLoadBalancing::LoadAlert::_duplicate (info.load_alert.in ());

      // The alert condition will be disabled.
      // @@ What happens if the below call fails?  This variable
      //    should be reset to one!
      info.alerted = 0;

      {
        // Release the lock prior to making the below remote
        // invocation.
        ACE_Reverse_Lock<TAO_SYNCH_MUTEX> reverse_lock (
          this->load_alert_lock_);
        ACE_GUARD (ACE_Reverse_Lock<TAO_SYNCH_MUTEX>,
                   reverse_guard,
                   reverse_lock);

        // Use AMI to make the following operation "non-blocking,"
        // allowing the caller to continue without being forced to
        // wait for a response.
        //
        // AMI is used to improve member selection times and overall
        // throughput since the LoadAlert object need not be alerted
        // synchronously.  In particular, the load alert can and
        // should be performed in parallel to other tasks, such as
        // member selection.
        load_alert->sendc_disable_alert (this->load_alert_handler_.in ()
                                         ACE_ENV_ARG_PARAMETER);
        ACE_CHECK;
      }
    }
  else
    ACE_THROW (CosLoadBalancing::LoadAlertNotFound ());
}

void
TAO_LB_LoadManager::register_load_alert (
    const PortableGroup::Location & the_location,
    CosLoadBalancing::LoadAlert_ptr load_alert
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   CosLoadBalancing::LoadAlertAlreadyPresent,
                   CosLoadBalancing::LoadAlertNotAdded))
{
  if (CORBA::is_nil (load_alert))
    ACE_THROW (CORBA::BAD_PARAM ());

  ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->load_alert_lock_);

  TAO_LB_LoadAlertInfo info;
  info.load_alert = CosLoadBalancing::LoadAlert::_duplicate (load_alert);

  int result = this->load_alert_map_.bind (the_location, info);

  if (result == 1)
    {
      ACE_THROW (CosLoadBalancing::LoadAlertAlreadyPresent ());
    }
  else if (result == -1)
    {
      // Problems dude!
      ACE_THROW (CosLoadBalancing::LoadAlertNotAdded ());
    }
}

CosLoadBalancing::LoadAlert_ptr
TAO_LB_LoadManager::get_load_alert (
    const PortableGroup::Location & the_location
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   CosLoadBalancing::LoadAlertNotFound))
{
  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX,
                    guard,
                    this->load_alert_lock_,
                    CosLoadBalancing::LoadAlert::_nil ());

  TAO_LB_LoadAlertMap::ENTRY * entry;
  if (this->load_alert_map_.find (the_location, entry) == 0)
    {
      TAO_LB_LoadAlertInfo & info = entry->int_id_;

      return
        CosLoadBalancing::LoadAlert::_duplicate (info.load_alert.in ());
    }
  else
    {
      ACE_THROW_RETURN (CosLoadBalancing::LoadAlertNotFound (),
                        CosLoadBalancing::LoadAlert::_nil ());
    }
}

void
TAO_LB_LoadManager::remove_load_alert (
    const PortableGroup::Location & the_location
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   CosLoadBalancing::LoadAlertNotFound))
{
  // Disable the "alert" status on the LoadAlert object since it will
  // no longer be associated with the LoadManager.  In particular,
  // requests should be allowed through once again since there will be
  // no way to control the load shedding mechanism once the LoadAlert
  // object is no longer under the control of the LoadManager.
  this->disable_alert (the_location
                       ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->load_alert_lock_);

  if (this->load_alert_map_.unbind (the_location) != 0)
    {
      ACE_THROW (CosLoadBalancing::LoadAlertNotFound ());
    }
}

void
TAO_LB_LoadManager::register_load_monitor (
    const PortableGroup::Location & the_location,
    CosLoadBalancing::LoadMonitor_ptr load_monitor
    ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   CosLoadBalancing::MonitorAlreadyPresent))
{
  if (CORBA::is_nil (load_monitor))
    ACE_THROW (CORBA::BAD_PARAM ());

  const CosLoadBalancing::LoadMonitor_var the_monitor =
    CosLoadBalancing::LoadMonitor::_duplicate (load_monitor);

  ACE_GUARD (TAO_SYNCH_MUTEX,
             guard,
             this->monitor_lock_);

  int result = this->monitor_map_.bind (the_location, the_monitor);

  if (result == 0
      && this->monitor_map_.current_size () == 1)
    {
      // Register the "pull monitoring" event handler only after the
      // first load monitor is registered.  This is an optimization to
      // prevent unnecessary invocation of the "pull monitoring" event
      // handler.
      ACE_Time_Value interval (TAO_LB_PULL_HANDLER_INTERVAL, 0);
      ACE_Time_Value restart (TAO_LB_PULL_HANDLER_RESTART, 0);
      this->timer_id_ = this->reactor_->schedule_timer (&this->pull_handler_,
                                                        0,
                                                        interval,
                                                        restart);

      if (this->timer_id_ == -1)
        {
          if (TAO_debug_level > 0)
            ACE_ERROR ((LM_ERROR,
                        "TAO_LB_LoadManager::register_load_monitor: "
                        "Unable to schedule timer.\n"));