ec_mcast.cpp

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

// EC_Mcast.cpp,v 1.50 2003/11/24 05:23:17 bala Exp

#include "EC_Mcast.h"

#include "orbsvcs/Event_Utilities.h"
#include "orbsvcs/Event_Service_Constants.h"
#include "orbsvcs/Time_Utilities.h"

#include "orbsvcs/Event/EC_Event_Channel.h"
#include "orbsvcs/Event/EC_Default_Factory.h"

#include "tao/ORB_Core.h"

#include "ace/Get_Opt.h"
#include "ace/Auto_Ptr.h"
#include "ace/Sched_Params.h"
#include "ace/Read_Buffer.h"
#include "ace/OS_NS_sys_time.h"

#if !defined (__ACE_INLINE__)
#include "EC_Mcast.i"
#endif /* __ACE_INLINE__ */

ACE_RCSID(EC_Mcast, EC_Mcast, "EC_Mcast.cpp,v 1.50 2003/11/24 05:23:17 bala Exp")

#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
template class ACE_PushConsumer_Adapter<ECM_Supplier>;
#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
#pragma instantiate ACE_PushConsumer_Adapter<ECM_Supplier>
#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */

ECM_Driver::ECM_Driver (void)
  : event_period_ (250000),
    event_count_ (100),
    config_filename_ (0),
    pid_filename_ (0),
    local_federations_count_ (0),
    all_federations_count_ (0)
{
}

int
ECM_Driver::run (int argc, char* argv[])
{
  ACE_DECLARE_NEW_CORBA_ENV;
  ACE_TRY
    {
      this->orb_ =
        CORBA::ORB_init (argc, argv, "" ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CORBA::Object_var poa_object =
        this->orb_->resolve_initial_references("RootPOA"
                                               ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (this->parse_args (argc, argv))
        return 1;

     if (this->parse_config_file ())
       return 1;

      ACE_DEBUG ((LM_DEBUG,
                  "Execution parameters:\n"
                  "  event period = <%d> (usecs)\n"
                  "  event count = <%d>\n"
                  "  config file name = <%s>\n"
                  "  pid file name = <%s>\n",

                  this->event_period_,
                  this->event_count_,

                  this->config_filename_?this->config_filename_:"nil",
                  this->pid_filename_?this->pid_filename_:"nil") );

      int i;
      for (i = 0; i < this->local_federations_count_; ++i)
        {
          ACE_DEBUG ((LM_DEBUG,
                      "    name = <%s>\n"
                      "    port = <%d>\n"
                      "    supplier types:\n",
                      this->local_federations_[i]->name ()?this->local_federations_[i]->name ():"nil",
                      this->local_federations_[i]->mcast_port ()));
          int j;
          for (j = 0;
               j < this->local_federations_[i]->supplier_types ();
               ++j)
            {

              ACE_DEBUG ((LM_DEBUG,
                          "    name = <%s>\n"
                          "    ipadd = <%x>\n",
                          this->local_federations_[i]->supplier_name (j),
                          this->local_federations_[i]->supplier_ipaddr (j)));
            }
          ACE_DEBUG ((LM_DEBUG,
                      "    consumer types:\n"));
          for (j = 0;
               j < this->local_federations_[i]->consumer_types ();
               ++j)
            {
              ACE_DEBUG ((LM_DEBUG,
                          "    name = <%s>\n"
                          "    ipadd = <%x>\n",
                          this->local_federations_[i]->consumer_name (j),
                          this->local_federations_[i]->consumer_ipaddr (j)));
            }
        }

      if (this->pid_filename_ != 0)
        {
          FILE* pid = ACE_OS::fopen (this->pid_filename_, "w");
          if (pid != 0)
            {
              ACE_OS::fprintf (pid, "%ld\n",
                               ACE_static_cast (long, ACE_OS::getpid ()));
              ACE_OS::fclose (pid);
            }
        }

      TAO_EC_Event_Channel_Attributes attr (root_poa.in (),
                                            root_poa.in ());
      TAO_EC_Event_Channel ec_impl (attr);

      // Register Event_Service with the Naming Service.
      RtecEventChannelAdmin::EventChannel_var ec =
        ec_impl._this (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CORBA::String_var str =
        this->orb_->object_to_string (ec.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "The (local) EC IOR is <%s>\n", str.in ()));

      poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ec_impl.activate (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: local EC objref ready\n"));

      this->open_federations (ec.in ()
                              ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: open_federations done\n"));

      this->open_senders (ec.in ()
                          ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: open_senders done\n"));

      this->open_receivers (ec.in ()
                            ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: open_receivers done\n"));

      this->activate_federations (ec.in ()
                                  ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: activate_federations done\n"));

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: running the test\n"));
      while (this->federations_running_ > 0)
        this->orb_->perform_work ();

      this->dump_results ();

      this->close_receivers (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
      this->close_senders (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      this->close_federations (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      ACE_DEBUG ((LM_DEBUG, "EC_Mcast: shutdown the EC\n"));

      ec_impl.shutdown (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

    }
  ACE_CATCH (CORBA::SystemException, sys_ex)
    {
      ACE_PRINT_EXCEPTION (sys_ex, "SYS_EX");
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "NON SYS EX");
    }
  ACE_ENDTRY;
  return 0;
}

void
ECM_Driver::federation_has_shutdown (ECM_Local_Federation *federation
                                     ACE_ENV_ARG_DECL_NOT_USED)
{
  ACE_DEBUG ((LM_DEBUG, "Federation <%s> shuting down\n",
              federation->name ()));
  if (this->federations_running_ > 0)
    this->federations_running_--;
}

void
ECM_Driver::open_federations (RtecEventChannelAdmin::EventChannel_ptr ec
                              ACE_ENV_ARG_DECL)
{
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->open (this->event_count_,
                                         ec ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::activate_federations (RtecEventChannelAdmin::EventChannel_ptr ec
                                  ACE_ENV_ARG_DECL)
{
  this->federations_running_ = this->local_federations_count_;
  RtecEventComm::Time interval = this->event_period_;
  interval *= 10;
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->activate (ec,
                                             interval
                                             ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::close_federations (ACE_ENV_SINGLE_ARG_DECL)
{
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->close (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::open_senders (RtecEventChannelAdmin::EventChannel_ptr ec
                          ACE_ENV_ARG_DECL)
{
  if (this->endpoint_.dgram ().open (ACE_Addr::sap_any) == -1)
    {
      // @@ TODO throw an application specific exception.
      ACE_THROW (CORBA::COMM_FAILURE ());
    }

  ACE_INET_Addr ignore_from;
  this->endpoint_.dgram ().get_local_addr (ignore_from);
  ACE_DEBUG ((LM_DEBUG, "ECM_Driver::open_senders - "
              "local endpoint = (%u:%d)\n",
              ignore_from.get_ip_address (),
              ignore_from.get_port_number ()));
  for (int i = 0; i < this->all_federations_count_; ++i)
    {
      this->all_federations_[i]->open (&this->endpoint_,
                                       ec
                                       ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::close_senders (ACE_ENV_SINGLE_ARG_DECL)
{
  for (int i = 0; i < this->all_federations_count_; ++i)
    {
      this->all_federations_[i]->close (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
    }
  this->endpoint_.dgram ().close ();
}

void
ECM_Driver::open_receivers (RtecEventChannelAdmin::EventChannel_ptr ec
                            ACE_ENV_ARG_DECL)
{
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->open_receiver (ec,
                                                  &this->endpoint_
                                                  ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::close_receivers (ACE_ENV_SINGLE_ARG_DECL)
{
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->close_receiver (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;
    }
}

void
ECM_Driver::dump_results (void)
{
  for (int i = 0; i < this->local_federations_count_; ++i)
    {
      this->local_federations_[i]->dump_results ();
    }
}


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

int
ECM_Driver::parse_args (int argc, char *argv [])
{
  ACE_Get_Opt get_opt (argc, argv, "l:p:c:n:t:f:");
  int opt;

  while ((opt = get_opt ()) != EOF)
    {
      switch (opt)
        {
        case 'p':
          this->pid_filename_ = get_opt.opt_arg ();
          break;

        case 'c':
          this->config_filename_ = get_opt.opt_arg ();
          break;

        case 't':
          this->event_period_ = ACE_OS::atoi (get_opt.opt_arg ());
          break;

        case 'n':
          this->event_count_ = ACE_OS::atoi (get_opt.opt_arg ());
          break;

        case 'f':
          {
            char* aux;
            int i = 0;
            for (char* arg = ACE_OS::strtok_r (get_opt.opt_arg (), ",", &aux);
                 arg != 0 && i < ECM_Driver::MAX_LOCAL_FEDERATIONS;
                 arg = ACE_OS::strtok_r (0, ",", &aux), ++i)
              {
                this->local_names_[i] = arg;
              }
            this->local_federations_count_ = i;
          }
          break;

        case '?':
        default:
          ACE_DEBUG ((LM_DEBUG,
                      "Usage: %s "
                      "[ORB options] "
                      "-n <event_count> "
                      "-t <event_period> "
                      "-l <localname> "
                      "-p <pid file name> "
                      "-c <config file name> "
                      "-f federation,federation,... "
                      "\n",