ec_multiple.cpp

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

          break;
        case 'd':
          this->schedule_file_ = get_opt.opt_arg ();
          break;

        case '?':
        default:
          ACE_DEBUG ((LM_DEBUG,
                      "Usage: %s "
                      "[ORB options] "
                      "-l <local_name> "
                      "-r <remote_name> "
                      "-s <global|local|runtime> "
                      "-i <consumer disc.,supplier disc.> "
                      "-x (short circuit EC) "
                      "-h <high priority args> "
                      "-w <low priority args> "
                      "-p <pid file name> "
                      "-d <schedule file name> "
                      "\n",
                      argv[0]));
          return -1;
        }
    }

  if (this->hp_message_count_ < 0
      || this->hp_message_count_ >= Test_ECG::MAX_EVENTS)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "%s: HP event count (%d) is out of range, "
                  "reset to default (%d)\n",
                  argv[0], this->lp_message_count_,
                  160));
      this->hp_message_count_ = 160;
    }

  if (this->lp_message_count_ < 0
      || this->lp_message_count_ >= Test_ECG::MAX_EVENTS)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "%s: LP event count (%d) is out of range, "
                  "reset to default (%d)\n",
                  argv[0], this->lp_message_count_,
                  4));
      this->lp_message_count_ = 4;
    }

  if (this->hp_consumers_ <= 0
      || this->lp_consumers_ < 0
      || this->hp_consumers_ + this->lp_consumers_ >= Test_ECG::MAX_CONSUMERS
      || this->hp_suppliers_ <= 0
      || this->lp_suppliers_ < 0
      || this->hp_suppliers_ + this->lp_suppliers_ >= Test_ECG::MAX_SUPPLIERS)
    {
      ACE_ERROR_RETURN ((LM_DEBUG,
                         "%s: number of consumers (low: %d, high: %d) or "
                         "suppliers (low: %d, high: %d) out of range\n",
                         argv[0],
                         lp_consumers_, hp_consumers_,
                         lp_suppliers_, lp_suppliers_), -1);
    }

  return 0;
}

Test_Supplier::Test_Supplier (Test_ECG *test,
                              void *cookie)
  :  test_ (test),
     cookie_ (cookie),
     consumer_ (this)
{
}

void
Test_Supplier::open (const char* name,
                     int event_a,
                     int event_b,
                     int message_count,
                     const RtecScheduler::Period_t& rate,
                     RtecEventChannelAdmin::EventChannel_ptr ec
                     ACE_ENV_ARG_DECL)
{
  this->event_a_ = event_a;
  this->event_b_ = event_b;
  this->message_count_ = message_count;

  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  RtecScheduler::handle_t rt_info =
    server->create (name ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // The execution times are set to reasonable values, but
  // actually they are changed on the real execution, i.e. we
  // lie to the scheduler to obtain right priorities; but we
  // don't care if the set is schedulable.
  ACE_Time_Value tv (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  ACE_DEBUG ((LM_DEBUG, "register supplier \"%s\"\n", name));
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               rate,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               1,
                   RtecScheduler::OPERATION
               ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  this->supplier_id_ = ACE::crc32 (name);
  ACE_DEBUG ((LM_DEBUG, "ID for <%s> is %04.4x\n", name,
              this->supplier_id_));

  ACE_SupplierQOS_Factory qos;
  qos.insert (this->supplier_id_,
              this->event_a_,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              this->event_b_,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              ACE_ES_EVENT_SHUTDOWN,
              rt_info, 1);

  RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
    ec->for_suppliers (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->consumer_proxy_ =
    supplier_admin->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  RtecEventComm::PushSupplier_var objref = this->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->consumer_proxy_->connect_push_supplier (objref.in (),
                                                qos.get_SupplierQOS ()
                                                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
Test_Supplier::close (ACE_ENV_SINGLE_ARG_DECL)
{
  if (CORBA::is_nil (this->consumer_proxy_.in ()))
    return;

  RtecEventChannelAdmin::ProxyPushConsumer_var proxy =
    this->consumer_proxy_._retn ();
  proxy->disconnect_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
}

void
Test_Supplier::activate (const char* name,
                         const RtecScheduler::Period_t& rate,
                         RtecEventChannelAdmin::EventChannel_ptr ec
                         ACE_ENV_ARG_DECL)
{
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  const int bufsize = 512;
  char buf[bufsize];
  ACE_OS::strcpy (buf, "consumer_");
  ACE_OS::strcat (buf, name);
  RtecScheduler::handle_t rt_info =
    server->create (buf ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // The execution times are set to reasonable values, but
  // actually they are changed on the real execution, i.e. we
  // lie to the scheduler to obtain right priorities; but we
  // don't care if the set is schedulable.
  ACE_Time_Value tv (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  ACE_DEBUG ((LM_DEBUG, "activate \"%s\"\n", buf));
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               rate,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               1,
               RtecScheduler::OPERATION
               ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // Also connect our consumer for timeout events from the EC.
  int interval = rate / 10;
  ACE_Time_Value tv_timeout (interval / ACE_ONE_SECOND_IN_USECS,
                             interval % ACE_ONE_SECOND_IN_USECS);
  TimeBase::TimeT timeout;
  ORBSVCS_Time::Time_Value_to_TimeT (timeout, tv_timeout);

  ACE_ConsumerQOS_Factory consumer_qos;
  consumer_qos.start_disjunction_group ();
  consumer_qos.insert_time (ACE_ES_EVENT_INTERVAL_TIMEOUT,
                            timeout,
                            rt_info);

  // = Connect as a consumer.
  RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
    ec->for_consumers (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->supplier_proxy_ =
    consumer_admin->obtain_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  RtecEventComm::PushConsumer_var cref =
    this->consumer_._this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->supplier_proxy_->connect_push_consumer (
       cref.in (),
       consumer_qos.get_ConsumerQOS ()
       ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
Test_Supplier::push (const RtecEventComm::EventSet& events
                     ACE_ENV_ARG_DECL)
{
#if 0
   const int bufsize = 128;
   char buf[bufsize];
   ACE_OS::sprintf (buf, "Supplier %d receives event in thread: ",
                    this->supplier_id_);
   print_priority_info (buf);
#endif

  if (events.length () == 0 || this->message_count_ < 0)
    {
      // ACE_DEBUG ((LM_DEBUG, "no events\n"));
      return;
    }

  RtecEventComm::EventSet sent (events.length ());
  sent.length (events.length ());

  for (u_int i = 0; i < events.length (); ++i)
    {
      const RtecEventComm::Event& e = events[i];
      if (e.header.type != ACE_ES_EVENT_INTERVAL_TIMEOUT)
        continue;

      // ACE_DEBUG ((LM_DEBUG, "Test_Supplier - timeout (%t)\n"));

      RtecEventComm::Event& s = sent[i];
      s.header.source = this->supplier_id_;
      s.header.ttl = 1;

      ACE_hrtime_t t = ACE_OS::gethrtime ();
      ORBSVCS_Time::hrtime_to_TimeT (s.header.creation_time, t);

      this->message_count_--;

      if (this->message_count_ < 0)
        {
          //this->supplier_proxy_->disconnect_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
          //if (ACE_ENV_SINGLE_ARG_PARAMETER.exception () != 0) return;
          this->test_->shutdown_supplier (this->cookie_,
                                          this->consumer_proxy_.in ()
                                          ACE_ENV_ARG_PARAMETER);
        }
      if (this->message_count_ % 2 == 0)
        {
          // Generate an A event...
          s.header.type = this->event_a_;
        }
      else
        {
          s.header.type = this->event_b_;
        }
    }
  this->test_->push_supplier (this->cookie_,
                              this->consumer_proxy_.in (),
                              sent
                              ACE_ENV_ARG_PARAMETER);
}

void
Test_Supplier::disconnect_push_supplier (ACE_ENV_SINGLE_ARG_DECL)
    ACE_THROW_SPEC ((CORBA::SystemException))
{
  if (CORBA::is_nil (this->supplier_proxy_.in ()))
    return;

  this->supplier_proxy_->disconnect_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
}

void
Test_Supplier::disconnect_push_consumer (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
{
}

int Test_Supplier::supplier_id (void) const
{
  return this->supplier_id_;
}

Test_Consumer::Test_Consumer (Test_ECG *test,
                              void *cookie)
  : test_ (test),
    cookie_ (cookie)
{
}

void
Test_Consumer::open (const char* name,
                     int event_a, int event_b,
                     RtecEventChannelAdmin::EventChannel_ptr ec
                     ACE_ENV_ARG_DECL)
{
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  RtecScheduler::handle_t rt_info =
    server->create (name ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // The worst case execution time is far less than 2
  // milliseconds, but that is a safe estimate....
  ACE_Time_Value tv (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  ACE_DEBUG ((LM_DEBUG, "register consumer \"%s\"\n", name));
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               0,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               0,
               RtecScheduler::OPERATION
               ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  ACE_ConsumerQOS_Factory qos;
  qos.start_disjunction_group ();
  qos.insert_type (ACE_ES_EVENT_SHUTDOWN, rt_info);
  qos.insert_type (event_a, rt_info);
  qos.insert_type (event_b, rt_info);

  // = Connect as a consumer.
  RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
    ec->for_consumers (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->supplier_proxy_ =
    consumer_admin->obtain_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  RtecEventComm::PushConsumer_var objref = this->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  this->supplier_proxy_->connect_push_consumer (objref.in (),
                                                qos.get_ConsumerQOS ()
                                                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
Test_Consumer::close (ACE_ENV_SINGLE_ARG_DECL)
{
  if (CORBA::is_nil (this->supplier_proxy_.in ()))
    return;

  RtecEventChannelAdmin::ProxyPushSupplier_var proxy =
    this->supplier_proxy_._retn ();
  proxy->disconnect_push_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
}

void
Test_Consumer::push (const RtecEventComm::EventSet& events
                     ACE_ENV_ARG_DECL)
    ACE_THROW_SPEC ((CORBA::SystemException))
{
  ACE_hrtime_t arrival = ACE_OS::gethrtime ();
  this->test_->push_consumer (this->cookie_, arrival, events ACE_ENV_ARG_PARAMETER);
}

void
Test_Consumer::disconnect_push_consumer (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
    ACE_THROW_SPEC ((CORBA::SystemException))
{
}

int
main (int argc, char *argv [])
{
  Test_ECG *test;

  // Dynamically allocate the Test_ECG instance so that we don't have
  // to worry about running out of stack space if it's large.
  ACE_NEW_RETURN (test,
                  Test_ECG,
                  -1);

  const int status = test->run (argc, argv);

  delete test;
  return status;
}