task_client.cpp

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

  ACE_DEBUG ((LM_DEBUG,
              "high priority jitter (%u samples):\n", number_of_samples));

  ACE_DEBUG ((LM_DEBUG,"Latency stats (time in usec)\n"));
  stats.print_summary (1, 1, stderr);

  return sqrt (jitter / (number_of_samples - 1));
}

// Returns the jitter in usecs.

ACE_timer_t
Client::get_low_priority_jitter (void)
{
  if (this->ts_->thread_count_ == 1)
    return 0;

  ACE_timer_t jitter = 0.0;
  ACE_timer_t average = get_low_priority_latency ();
  u_int number_of_samples = 0;

  // Compute the standard deviation, i.e., jitter, from the values
  // stored in the global_jitter_array_.

  ACE_Stats stats;

  // We first compute the sum of the squares of the differences each
  // latency has from the average.

  for (u_int j = 1;
       j < this->ts_->thread_count_;
       j++)
    {
      ACE_DEBUG ((LM_DEBUG, "count: %u\n", ts_->count_[j]));

      JITTER_ARRAY_ITERATOR iterator =
        this->ts_->global_jitter_array_ [j]->begin ();

      ACE_timer_t number_of_calls =
        this->ts_->count_ [j] / this->ts_->granularity_;

      ACE_timer_t *latency = 0;

      u_int i = 0;

      for (iterator.first ();
           i < number_of_calls && iterator.next (latency) == 1;
           iterator.advance ())
        {
          ++number_of_samples;
          ACE_timer_t difference = *latency - average;
          jitter += difference * difference;
          stats.sample (ACE_round (*latency));
        }
    }

  ACE_DEBUG ((LM_DEBUG,
              "low priority jitter (%u samples):\n", number_of_samples));
  ACE_DEBUG ((LM_DEBUG,"Latency stats (time in usec)\n"));
  stats.print_summary (1, 1, stderr);

  // Return the square root of the sum of the differences computed
  // above, i.e. jitter.
  return sqrt (jitter / (number_of_samples - 1));
}

ACE_timer_t
Client::get_jitter (u_int id)
{
  ACE_timer_t jitter = 0.0;
  ACE_timer_t average = get_latency (id);
  u_int number_of_samples = 0;

  // Compute the standard deviation, i.e., jitter, from the values
  // stored in the global_jitter_array_.

  ACE_Stats stats;

  // We first compute the sum of the squares of the differences each
  // latency has from the average.

  JITTER_ARRAY_ITERATOR iterator =
    this->ts_->global_jitter_array_ [id]->begin ();

  ACE_timer_t number_of_calls =
    this->ts_->count_[id] / this->ts_->granularity_;

  ACE_timer_t *latency = 0;

  u_int i = 0;

  for (iterator.first ();
       i < number_of_calls && iterator.next (latency) == 1;
       i ++,iterator.advance ())
    {
      ++number_of_samples;
      ACE_timer_t difference = *latency - average;
      jitter += difference * difference;
      stats.sample (ACE_round (*latency));
    }

  ACE_DEBUG ((LM_DEBUG,
             "jitter for thread id %u:\n", id));
  ACE_DEBUG ((LM_DEBUG,"Latency stats (time in usec)\n"));
  stats.print_summary (1, 1, stderr);

  // Return the square root of the sum of the differences computed
  // above, i.e. jitter.
  return sqrt (jitter / (number_of_samples - 1));
}

void
Client::find_frequency (void)
{
    if (this->ts_->thread_per_rate_ == 0)
      {
        if (this->id_ == 0)
          {
            ACE_DEBUG ((LM_DEBUG,
                        "(%t) I'm the high priority client, my id is %d.\n",
                        this->id_));
            this->frequency_ = CB_HIGH_PRIORITY_RATE;
          }
        else
          {
            ACE_DEBUG ((LM_DEBUG,
                        "(%t) I'm a low priority client, my id is %d.\n",
                        this->id_));
            this->frequency_ = CB_LOW_PRIORITY_RATE;
          }
      }
    else
      switch (this->id_)
        {
        case CB_20HZ_CONSUMER:
          this->frequency_ = CB_20HZ_CONSUMER_RATE;
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) I'm a %u Hz frequency client, "
                      "my id is %u.\n",
                      CB_20HZ_CONSUMER_RATE,
                      this->id_));
          break;
        case CB_10HZ_CONSUMER:
          this->frequency_ = CB_10HZ_CONSUMER_RATE;
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) I'm a %u Hz frequency client, "
                      "my id is %u.\n",
                      CB_10HZ_CONSUMER_RATE,
                      this->id_));
          break;
        case CB_5HZ_CONSUMER:
          this->frequency_ = CB_5HZ_CONSUMER_RATE;
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) I'm a %u Hz frequency client, "
                      "my id is %u.\n",
                      CB_5HZ_CONSUMER_RATE,
                      this->id_));
          break;
        case CB_1HZ_CONSUMER:
          this->frequency_ = CB_1HZ_CONSUMER_RATE;
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) I'm a %u Hz frequency client, "
                      "my id is %u.\n",
                      CB_1HZ_CONSUMER_RATE,
                      this->id_));
          break;
        default:
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) Invalid Thread ID!!!!\n",
                      this->id_));
        }
}

CORBA::ORB_ptr
Client::init_orb (ACE_ENV_SINGLE_ARG_DECL)
{
  ACE_DEBUG ((LM_DEBUG,
              "I'm thread %t\n"));


  // Convert the argv vector into a string.
  ACE_ARGV tmp_args (this->argv_);
  char tmp_buf[BUFSIZ];

  ACE_OS::strcpy (tmp_buf,
                  tmp_args.buf ());
  // Add the argument.
  ACE_OS::strcat (tmp_buf,
                  " -ORBRcvSock 32768 "
                  " -ORBSndSock 32768 ");

  ACE_DEBUG ((LM_DEBUG,
              tmp_buf));

  // Convert back to argv vector style.
  ACE_ARGV tmp_args2 (tmp_buf);
  int argc = tmp_args2.argc ();
  char **argv = tmp_args2.argv ();

  char orbid[64];
  ACE_OS::sprintf (orbid, "orb:%d", this->id_);
  CORBA::ORB_var orb = CORBA::ORB_init (argc,
                                        argv,
                                        orbid
                                        ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (CORBA::ORB::_nil ());

  if (this->id_ == 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "parsing the arguments\n"));

      int result = this->ts_->parse_args (argc,
                                          argv);
      if (result != 0)
        return CORBA::ORB::_nil ();

      ACE_DEBUG ((LM_DEBUG,
                  "(%t)Arguments parsed successfully\n"));

      ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ready_mon,
                        this->ts_->ready_mtx_,
                        CORBA::ORB::_nil ());
      this->ts_->ready_ = 1;
      this->ts_->ready_cnd_.broadcast ();
      ready_mon.release ();
    }

  ACE_DEBUG ((LM_DEBUG,
              "(%t) ORB_init success\n"));
  return orb._retn ();
}

int
Client::get_cubit (CORBA::ORB_ptr orb ACE_ENV_ARG_DECL)
{
  char *my_ior =
    this->ts_->use_utilization_test_ == 1
    ? this->ts_->one_ior_
    : this->ts_->iors_[this->id_];

  // If we are running the "1 to n" test make sure all low
  // priority clients use only 1 low priority servant.
  if (this->id_ > 0
      && this->ts_->one_to_n_test_ == 1)
    my_ior = this->ts_->iors_[1];

  if (my_ior == 0)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Must specify valid ior filename with -f option\n"),
                      -1);

  CORBA::Object_var objref =
    orb->string_to_object (my_ior
                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  if (CORBA::is_nil (objref.in ()))
    ACE_ERROR_RETURN ((LM_ERROR,
                       " (%t) string_to_object Failed!\n"),
                      -1);

  // Narrow the CORBA::Object reference to the stub object,
  // checking the type along the way using _is_a.
  this->cubit_ = Cubit::_narrow (objref.in ()
                                 ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  if (CORBA::is_nil (this->cubit_))
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Create cubit failed\n"),
                      -1);

  ACE_DEBUG ((LM_DEBUG,
              "(%t) Binding succeeded\n"));

  CORBA::String_var str =
    orb->object_to_string (this->cubit_
                           ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  ACE_DEBUG ((LM_DEBUG,
              "(%t) CUBIT OBJECT connected to <%s>\n",
              str.in ()));

  return 0;
}

int
Client::svc (void)
{
  ACE_DECLARE_NEW_CORBA_ENV;
  ACE_TRY
    {
      // Initialize the ORB.
      CORBA::ORB_var orb = this->init_orb (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      // Find the frequency of CORBA requests based on thread id.
      this->find_frequency ();

      // Get the cubit object from the file.
      int r = this->get_cubit (orb.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      if (r != 0)
        return r;

      ACE_DEBUG ((LM_DEBUG,
                  "(%t) Waiting for other threads to "
                  "finish binding..\n"));

      // Wait for all the client threads to be initialized before going
      // any further.
      this->ts_->barrier_->wait ();
      ACE_DEBUG ((LM_DEBUG,
                  "(%t; %D) Everyone's done, here I go!!\n"));
      if (this->ts_->oneway_ == 1)
        ACE_DEBUG ((LM_DEBUG,
                    "(%t) **** USING ONEWAY CALLS ****\n"));

      // Perform the tests.
      int result = this->run_tests ();
      if (result != 0)
        return result;

      // release the semaphore
      if (this->ts_->thread_per_rate_ == 1
          && this->id_ == this->ts_->thread_count_ - 1)
        this->ts_->semaphore_->release (this->ts_->thread_count_ - 1);
      else
        this->ts_->semaphore_->release ();

      // shutdown the server if necessary.
      if (this->ts_->shutdown_)
        {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) CALLING SHUTDOWN() ON THE SERVANT\n"));
          this->cubit_->shutdown (ACE_ENV_SINGLE_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }

      CORBA::release (this->cubit_);
      this->cubit_ = 0;

      orb->destroy (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "Task_Client::svc()");
    }
  ACE_ENDTRY;

  // To avoid a memPartFree on VxWorks.  It will leak memory, though.
  ACE_THR_FUNC_RETURN status = 0;

  if (thr_mgr ())
    thr_mgr ()->exit (status, 1);
  else
    ACE_OS::thr_exit (status);

  return 0;
}

int
Client::cube_octet (void)
{
  ACE_DECLARE_NEW_CORBA_ENV;
  ACE_TRY
    {
      this->call_count_++;
      // Cube an octet.
      CORBA::Octet arg_octet = func (this->num_);
      CORBA::Octet ret_octet = 0;

      START_QUANTIFY;

      if (this->ts_->use_utilization_test_ == 1 && this->ts_->remote_invocations_ == 0)
        ret_octet = this->cubit_impl_.cube_octet (arg_octet
                                                  ACE_ENV_ARG_PARAMETER);
      else
        ret_octet = this->cubit_->cube_octet (arg_octet