client.cpp

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

int
Client_i::activate_util_thread (void)
{
  ACE_NEW_RETURN (this->util_thread_,
                  Util_Thread (this->ts_,
                               &this->util_thread_manager_),
                  -1);

  // Time the utilization thread' "computation" to get %IdleCPU at the
  // end of the test.
  this->calc_util_time ();

  if (this->ts_->use_utilization_test_ == 1)
    // Activate the utilization thread only if specified.  See
    // description of this variable in header file.
    {
      this->low_priority_ =
        ACE_Sched_Params::priority_min (ACE_SCHED_FIFO,
                                        ACE_SCOPE_THREAD);
      ACE_DEBUG ((LM_DEBUG,
                  "Creating utilization thread with priority of %d\n",
                  this->low_priority_));

      // Activate the Utilization thread.  It will wait until all
      // threads have finished binding.
      this->util_thread_->activate (
        GLOBALS::instance ()->thr_create_flags,
        1,
        0,
        this->low_priority_);
    }
  else
    this->util_thread_->close ();
  return 0;
}

void
Client_i:: print_context_stats (void)
{
  if (this->ts_->context_switch_test_ == 1)
    {
#if defined (ACE_HAS_PRUSAGE_T)
      this->timer_for_context_switch.stop ();
      this->timer_for_context_switch.get_rusage (this->usage);
      // Add up the voluntary context switches & involuntary context
      // switches.
      this->context_switch_ =
        this->usage.pr_vctx + this->usage.pr_ictx - this->context_switch_;
      ACE_DEBUG ((LM_DEBUG,
                  "Voluntary context switches = %d, Involuntary context switches = %d\n",
                  this->usage.pr_vctx,
                  this->usage.pr_ictx));
#elif defined (ACE_HAS_GETRUSAGE) && !defined (ACE_WIN32)
      this->timer_for_context_switch.stop ();
      this->timer_for_context_switch.get_rusage (this->usage);
      // Add up the voluntary context switches & involuntary context
      // switches.
      this->context_switch_ =
        this->usage.ru_nvcsw + this->usage.ru_nivcsw - this->context_switch_;
      ACE_DEBUG ((LM_DEBUG,
                  "Voluntary context switches=%d, Involuntary context switches=%d\n",
                  this->usage.ru_nvcsw,
                  this->usage.ru_nivcsw));
#elif defined (VXWORKS) /* ACE_HAS_GETRUSAGE */
      taskSwitchHookDelete ((FUNCPTR) &switchHook);
      ACE_DEBUG ((LM_DEBUG,
                  "Context switches=%d\n",
                  ctx));
#endif /* ACE_HAS_PRUSAGE_T */
    }
}

void
Client_i::print_latency_stats (void)
{
  // If running the utilization test, don't report latency nor jitter.
  if (this->ts_->use_utilization_test_ == 0)
    {
#if defined (VXWORKS)
      ACE_DEBUG ((LM_DEBUG,
                  "Test done.\n"
                  "High priority client latency : %f usec, jitter: %f usec\n"
                  "Low priority client latency : %f usec, jitter: %f usec\n",
                  this->high_priority_client_->get_high_priority_latency (),
                  this->high_priority_client_->get_high_priority_jitter (),
                  this->low_priority_client_[0]->get_low_priority_latency (),
                  this->low_priority_client_[0]->get_low_priority_jitter ()));
      // Output the latency values to a file, tab separated, to import
      // it to Excel to calculate jitter, in the mean time we come up
      // with the sqrt() function.
      output_latency ();
#elif defined (CHORUS_MVME)
      ACE_DEBUG ((LM_DEBUG,
                  "Test done.\n"
                  "High priority client latency : %u usec\n"
                  "Low priority client latency : %u usec\n",
                  this->high_priority_client_->get_high_priority_latency (),
                  this->low_priority_client_[0]->get_low_priority_latency () ));
      // Output the latency values to a file, tab separated, to import
      // it to Excel to calculate jitter, in the mean time we come up
      // with the sqrt() function.
      output_latency ();
#else /* !CHORUS_MVME */
      ACE_DEBUG ((LM_DEBUG, "Test done.\n"
                  "High priority client latency : %f usec, jitter: %f usec\n"
                  "Low priority client latency : %f usec, jitter: %f usec\n",
                  this->high_priority_client_->get_high_priority_latency (),
                  this->high_priority_client_->get_high_priority_jitter (),
                  this->low_priority_client_[0]->get_low_priority_latency (),
                  this->low_priority_client_[0]->get_low_priority_jitter ()));
      // output_latency ();
#endif /* !VXWORKS && !CHORUS_MVME */
    }
}

void
Client_i::print_util_stats (void)
{
 if (this->ts_->use_utilization_test_ == 1)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "(%t) Scavenger task performed \t%u computations\n"
                  "(%t) CLIENT task performed \t\t%u %s calls as requested\n\n"
                  "(%t) Utilization test time is \t\t%f microseconds\n\t%s\n",
                  this->util_thread_->get_number_of_computations (),
                  this->ts_->loop_count_,
                  this->ts_->remote_invocations_ == 1  ?  "CORBA"  :  "local",
                  this->ts_->util_test_time_,
                  this->ts_->remote_invocations_ == 1 ?
                    "NOW run the same test again, adding the \"-l\" option.  See README file for explanation." :
                    " "
                  ));
    }
}

void
Client_i::print_priority_inversion_stats (void)
{
  this->print_context_stats ();
  this->print_latency_stats ();
  this->print_util_stats ();
}

int
Client_i::start_servant (void)
{
  char high_thread_args[BUFSIZ];

  ACE_OS::sprintf (high_thread_args,
                   "-ORBSndSock 32768 "
                   "-ORBRcvSock 32768 ");

  Cubit_Task *high_priority_task;

  ACE_NEW_RETURN (high_priority_task,
                  Cubit_Task ((const char *) high_thread_args,
                              (const char *) "internet",
                              (u_int) 1,
                              &this->server_thread_manager_,
                              (u_int) 0), // task id 0.
                  -1);

  this->high_priority_ =
    this->priority_.get_high_priority ();

  ACE_DEBUG ((LM_DEBUG,
              "Creating servant 0 with high priority %d\n",
              this->high_priority_));

  // Make the high priority task an active object.
   if (high_priority_task->activate (
       GLOBALS::instance ()->thr_create_flags,
       1,
       0,
       this->high_priority_) == -1)
     ACE_ERROR ((LM_ERROR,
                 "(%P|%t) %p\n",
                 "\thigh_priority_task->activate failed"));

   ACE_MT (ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ready_mon, GLOBALS::instance ()->ready_mtx_,-1));

   // Wait on the condition variable till the high priority cubit_task
   // has finished argument processing.
   while (!GLOBALS::instance ()->ready_)
     GLOBALS::instance ()->ready_cnd_.wait ();
   // wait on the barrier till the servant writes its ior.
   GLOBALS::instance ()->barrier_->wait ();

   this->ts_->one_ior_ =
     high_priority_task->get_servant_ior (0);

   return 0;
}

int
Client_i::do_priority_inversion_test (void)
{
  this->timer_.start ();
#if defined (VXWORKS)
  ctx = 0;
  ACE_NEW_RETURN (this->task_id_,
                  char[TASK_ID_LEN],
                  -1);
#endif /* VXWORKS */
  ACE_DEBUG ((LM_DEBUG,
              "(%P|%t) <<<<<<< starting test on %D\n"));
  GLOBALS::instance ()->num_of_objs = 1;

  for (int j = 0; j < this->argc_; j++)
    if (ACE_OS::strcmp (this->argv_[j], "-u") == 0)
      {
        this->start_servant ();
        break;
      }
  // Create and activate the high priority client.
  int result = this->activate_high_client ();
  if (result < 0)
    return result;

  ACE_MT (ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ready_mon, this->ts_->ready_mtx_,-1));

  // Wait on the condition variable until the high priority thread
  // wakes us up.
  while (!this->ts_->ready_)
    this->ts_->ready_cnd_.wait ();

  result = this->activate_low_client ();

  if (result < 0)
    return result;

  // Activate the utilization thread if necessary.
  result = this->activate_util_thread ();
  if (result < 0)
    return result;

  // Wait for all the client threads to be initialized before going
  // any further.
  this->ts_->barrier_->wait ();

  STOP_QUANTIFY;
  CLEAR_QUANTIFY;

  // Collect the context switch data.
  this->get_context_switches ();

  // Wait for all the client threads to exit (except the utilization
  // thread).
  this->client_thread_manager_.wait ();

  STOP_QUANTIFY;
  ACE_DEBUG ((LM_DEBUG,"(%P|%t) >>>>>>> ending test on %D\n"));

  this->timer_.stop ();
  this->timer_.elapsed_time (this->delta_);

  // Signal the utilization thread to finish with its work..  only if
  // utilization test was specified.  See description of this variable
  // in header file.
  if (this->ts_->use_utilization_test_ == 1)
    {
      this->util_thread_->done_ = 1;

      // This will wait for the utilization thread to finish.
      this->util_thread_manager_.wait ();
    }

  ACE_DEBUG ((LM_DEBUG,
              "-------------------------- Stats -------------------------------\n"));

  this->print_priority_inversion_stats ();

  return 0;
}

int
Client_i::do_thread_per_rate_test (void)
{
  Client CB_20Hz_client (&this->client_thread_manager_,
                         this->ts_,
                         this->argc_,
                         this->argv_,
                         CB_20HZ_CONSUMER);
  Client CB_10Hz_client (&this->client_thread_manager_,
                         this->ts_,
                         this->argc_,
                         this->argv_,
                         CB_10HZ_CONSUMER);
  Client CB_5Hz_client (&this->client_thread_manager_,
                        this->ts_,
                        this->argc_,
                        this->argv_,
                        CB_5HZ_CONSUMER);
  Client CB_1Hz_client (&this->client_thread_manager_,
                        this->ts_,
                        this->argc_,
                        this->argv_,
                        CB_1HZ_CONSUMER);
  ACE_Sched_Priority priority;

  priority = this->priority_.get_high_priority ();
  ACE_DEBUG ((LM_DEBUG,
              "Creating 20 Hz client with priority %d\n",
              priority));

  if (CB_20Hz_client.activate (
       GLOBALS::instance ()->thr_create_flags,
       1,
       1,
       priority) == -1)
    ACE_ERROR ((LM_ERROR,
                "(%P|%t) errno = %p: activate failed\n"));
  // The high priority thread is parsing the arguments, so wait on the
  // condition variable until it wakes us up.

  ACE_DEBUG ((LM_DEBUG,
              "(%t) Waiting for argument parsing\n"));

  ACE_MT (ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ready_mon, this->ts_->ready_mtx_,-1));

  // Wait on the condition variable until the high priority thread
  // wakes us up.
  while (!this->ts_->ready_)
    this->ts_->ready_cnd_.wait ();

  ACE_DEBUG ((LM_DEBUG,
              "(%t) Argument parsing waiting done\n"));

  priority = ACE_Sched_Params::previous_priority (ACE_SCHED_FIFO,
                                                  priority,
                                                  ACE_SCOPE_THREAD);
  ACE_DEBUG ((LM_DEBUG,
              "Creating 10 Hz client with priority %d\n",
              priority));

  if (CB_10Hz_client.activate (
       GLOBALS::instance ()->thr_create_flags,
       1,
       1,
       priority) == -1)
    ACE_ERROR ((LM_ERROR,
                "(%P|%t) errno = %p: activate failed\n"));

  priority = ACE_Sched_Params::previous_priority (ACE_SCHED_FIFO,
                                                  priority,
                                                  ACE_SCOPE_THREAD);
  ACE_DEBUG ((LM_DEBUG,
              "Creating 5 Hz client with priority %d\n",
              priority));

  if (CB_5Hz_client.activate (
       GLOBALS::instance ()->thr_create_flags,
       1,
       1,
       priority) == -1)
    ACE_ERROR ((LM_ERROR,
                "(%P|%t) errno = %p: activate failed\n"));

  priority = ACE_Sched_Params::previous_priority (ACE_SCHED_FIFO,
                                                  priority,
                                                  ACE_SCOPE_THREAD);
  ACE_DEBUG ((LM_DEBUG,
              "Creating 1 Hz client with priority %d\n",
              priority));

  if (CB_1Hz_client.activate (
       GLOBALS::instance ()->thr_create_flags,
       1,
       1,
       priority) == -1)
    ACE_ERROR ((LM_ERROR,
                "(%P|%t) errno = %p: activate failed\n"));

  // Wait for all the threads to exit.
  this->client_thread_manager_.wait ();

  ACE_DEBUG ((LM_DEBUG,
              "Test done.\n"
              "20Hz client latency : %A usec, jitter: %A usec\n"
              "10Hz client latency : %A usec, jitter: %A usec\n"
              "5Hz client latency : %A usec, jitter: %A usec\n"
              "1Hz client latency : %A usec, jitter: %A usec\n",
              CB_20Hz_client.get_latency (0),
              CB_20Hz_client.get_jitter (0),
              CB_10Hz_client.get_latency (1),
              CB_10Hz_client.get_jitter (1),
              CB_5Hz_client.get_latency (2),
              CB_5Hz_client.get_jitter (2),
              CB_1Hz_client.get_latency (3),
              CB_1Hz_client.get_jitter (3) ));
  return 0;
}

// This is the main routine of the client, where we create a high
// priority and a low priority client. we then activate the clients
// with the appropriate priority threads, and wait for them to
// finish. After they aer done, we compute the latency and jitter
// metrics and print them.

int
main (int argc, char *argv[])
{
  ACE_Log_Msg::instance()->clr_flags (ACE_Log_Msg::LOGGER);

  Client_i client;

  int result = client.init (argc,argv);

  if (result < 0)
    return result;

  // Run the tests.
  client.run ();

#if defined (CHORUS_MVME)
  int pTime;
  if (pccTimer (PCC2_TIMER1_STOP,
                &pTime) != K_OK)
    ACE_DEBUG ((LM_DEBUG,
                "pccTimer has a pending bench mark\n"));
#endif /* CHORUS_MVME */
  return 0;
}

#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
template class ACE_Singleton<Globals,ACE_Null_Mutex>;
template class ACE_Unbounded_Queue<ACE_timer_t>;
template class ACE_Unbounded_Queue_Iterator<ACE_timer_t>;
template class ACE_Node<ACE_timer_t>;
#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
#pragma instantiate ACE_Singleton<Globals,ACE_Null_Mutex>
#pragma instantiate ACE_Unbounded_Queue<ACE_timer_t>
#pragma instantiate ACE_Unbounded_Queue_Iterator<ACE_timer_t>
#pragma instantiate ACE_Node<ACE_timer_t>
#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */