client.cpp

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

// client.cpp,v 1.112 2003/11/04 05:21:33 dhinton Exp

#include "ace/config-all.h"
#if defined (VXWORKS)
# undef ACE_MAIN
# define ACE_MAIN client
#endif /* VXWORKS */

#include "ace/Sched_Params.h"
#include "tao/Strategies/advanced_resource.h"
#include "server.h"
#include "client.h"
#include "Globals.h"
#include "ace/Barrier.h"

#if defined (ACE_HAS_QUANTIFY)
# include "quantify.h"
#endif /* ACE_HAS_QUANTIFY */

ACE_RCSID(MT_Cubit, client, "client.cpp,v 1.112 2003/11/04 05:21:33 dhinton Exp")

#if defined (VXWORKS)
u_int ctx = 0;
u_int ct = 0;

typedef struct
{
  char name[15];
  WIND_TCB *tcb;
#if (CPU_FAMILY == PPC)
  INSTR pc;
#else
  INSTR *pc;
#endif
} task_info;

const u_int SWITCHES = 25000;
task_info tInfo[SWITCHES];

extern "C"
int
switchHook (WIND_TCB *pOldTcb,    // pointer to old task's WIND_TCB.
            WIND_TCB *pNewTcb)   // pointer to new task's WIND_TCB.
{
  ACE_UNUSED_ARG (pOldTcb);

  // We create the client threads with names starting with "@".
  if (pNewTcb->name[0] == '@')
    ctx++;

  if (ct < SWITCHES)
    {
      ACE_OS::strncpy (tInfo[ct].name,
                       pNewTcb->name,
                       TASKNAME_LEN);
      tInfo[ct].tcb = pNewTcb;
      tInfo[ct].pc  = pNewTcb->regs.pc;
      ct++;
    }

  return 0;
}
#endif /* VXWORKS */

// Constructor.

Client_i::Client_i (void)
  : high_priority_client_ (0),
    low_priority_client_ (0),
    util_thread_ (0),
    ts_ (0),
    num_low_priority_ (0),
    num_priorities_ (0),
    grain_ (0),
    counter_ (0),
    task_id_ (0),
    argc_ (0),
    argv_ (0),
    context_switch_ (0)
{
}

// Destructor.

Client_i::~Client_i (void)
{
  delete this->high_priority_client_;
  if (this->low_priority_client_ != 0)
    // Delete the low priority task array.
    for (u_int i = this->num_low_priority_;
         i > 0;
         i--)
      delete this->low_priority_client_[i - 1];
  delete [] this->low_priority_client_;
  delete this->util_thread_;
  delete this->ts_;
}

int
Client_i::init (int argc, char *argv[])
{
  this->argc_ = argc;
  this->argv_ = argv;

  int result;
  result = GLOBALS::instance ()->sched_fifo_init ();
  if (result == -1)
    return result;
  VX_VME_INIT;
  FORCE_ARGV (this->argc_,this->argv_);
  // Make sure we've got plenty of socket handles.  This call will use
  // the default maximum.
  ACE::set_handle_limit ();

  ACE_NEW_RETURN (this->ts_,
                  Task_State,
                  -1);

  // Preliminary argument processing.
  for (int i=0;
       i< this->argc_;
       i++)
    {
      if (ACE_OS::strcmp (this->argv_[i],"-r") == 0)
        this->ts_->thread_per_rate_ = 1;
      else if (ACE_OS::strcmp (this->argv_[i],"-t") == 0
               && (i - 1 < this->argc_))
        this->ts_->thread_count_ =
          ACE_OS::atoi (this->argv_[i+1]);
    }
  PCCTIMER_INIT;
  return 0;
}

void
Client_i::run (void)
{
  if (this->ts_->thread_per_rate_ == 0)
    {
      this->do_priority_inversion_test ();

      if (this->ts_->use_utilization_test_ == 1)
        {
          // Exit.  Otherwise, the process just waits forever.
          ACE_OS::exit ();
        }
    }
  else
    this->do_thread_per_rate_test ();
}

#if defined (VXWORKS)
void
Client_i::output_taskinfo (void)
{
  FILE *file_handle = ACE_OS::fopen ("taskinfo.txt", "w");

  if (file_handle == 0)
    ACE_ERROR ((LM_ERROR,
                "%p\n",
                "open"));

  ACE_DEBUG ((LM_DEBUG,
              "--->Output file for taskinfo data is \"taskinfo.txt\"\n"));

  // This loop visits each client.  thread_count_ is the number of
  // clients.

  for (u_int j = 0; j < SWITCHES; j ++)
    ACE_OS::fprintf(file_handle,
                    "\tname= %s\ttcb= %p\tpc= %p\n",
                    tInfo[j].name,
                    tInfo[j].tcb,
                    tInfo[j].pc);

  ACE_OS::fclose (file_handle);
}
#endif /* VXWORKS */

void
Client_i::get_context_switches (void)
{
#if (defined (ACE_HAS_PRUSAGE_T) || defined (ACE_HAS_GETRUSAGE)) && !defined (ACE_WIN32)

  if (this->ts_->context_switch_test_ == 1)
    {
      this->timer_for_context_switch.start ();
      this->timer_for_context_switch.get_rusage (this->usage);
# if defined (ACE_HAS_PRUSAGE_T)
      this->context_switch_ = this->usage.pr_vctx + this->usage.pr_ictx;
# else  /* ACE_HAS_PRUSAGE_T */
      this->context_switch_ = this->usage.ru_nvcsw + this->usage.ru_nivcsw;
# endif /* ACE_HAS_GETRUSAGE */
    }
#endif /* ACE_HAS_PRUSAGE_T || ACE_HAS_GETRUSAGE */

#if defined (VXWORKS)
  if (this->ts_->context_switch_test_ == 1)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "Adding the context switch hook!\n"));
      taskSwitchHookAdd ((FUNCPTR) &switchHook);
    }
#endif /* VXWORKS */
}

void
Client_i::output_latency (void)
{
  FILE *latency_file_handle = 0;
  char latency_file[BUFSIZ];
  char buffer[BUFSIZ];

  ACE_OS::sprintf (latency_file,
                   "c%d",
                   this->ts_->thread_count_);

  ACE_DEBUG ((LM_DEBUG,
              "--->Output file for latency data is \"%s\"\n",
              latency_file));

  latency_file_handle = ACE_OS::fopen (latency_file, "w");

  // This loop visits each client.  thread_count_ is the number of
  // clients.
  for (u_int j = 0;
       j < this->ts_->thread_count_;
       j++)
    {
      ACE_OS::sprintf(buffer,
                      "%s #%d",
                      j == 0
                      ? "High Priority"
                      : "Low Priority",
                      j);
      // This loop visits each request latency from a client.
      JITTER_ARRAY_ITERATOR iterator =
        this->ts_->global_jitter_array_ [j]->begin ();

      u_int i = 0;
      ACE_timer_t *latency = 0;

      for (iterator.first ();
           (i < (j == 0
                 ? this->ts_->high_priority_loop_count_
                 : this->ts_->loop_count_) / this->ts_->granularity_) &&
             (iterator.next (latency));
           i++,iterator.advance ())
        {
          ACE_OS::sprintf (buffer + ACE_OS::strlen (buffer),
#if defined (CHORUS_MVME)
                          "\t%u\n",
#else
                          "\t%f\n",
#endif /* !CHORUS_MVME */
                           *latency);
          ACE_OS::fputs (buffer,
                         latency_file_handle);
          buffer[0] = 0;
        }
    }

  ACE_OS::fclose (latency_file_handle);
}

// Mechanism to distribute the available priorities among the threads
// when there are not enough different priorities for all threads.

void
Client_i::init_low_priority (void)
{
  ACE_Sched_Priority prev_priority = this->high_priority_;
  if (this->ts_->use_multiple_priority_ == 1)
    this->low_priority_ =
      this->priority_.get_low_priority (this->num_low_priority_,
                                        prev_priority,
                                        1);
  else
    this->low_priority_ =
      this->priority_.get_low_priority (this->num_low_priority_,
                                        prev_priority,
                                        0);
  this->num_priorities_ =
    this->priority_.number_of_priorities ();
  this->grain_ = this->priority_.grain ();
  this->counter_ = 0;
}

void
Client_i::calc_util_time (void)
{
  MT_Cubit_Timer timer (ACE_ONE_SECOND_IN_MSECS);
  // Time the utilization thread' "computation" to get % IdleCPU at the
  // end of the test.

  // Execute one computation.
  timer.start ();
#if defined (CHORUS_MVME)
  this->util_thread_->computation ();
  timer.stop ();
  this->util_task_duration_ = timer.get_elapsed ();
#else
  for (u_int i = 0;
       i < NUM_UTIL_COMPUTATIONS;
       i++)
    this->util_thread_->computation ();

  timer.stop ();
  this->util_task_duration_ = timer.get_elapsed () / NUM_UTIL_COMPUTATIONS;
#endif /* !CHORUS_MVME */
}

int
Client_i::activate_high_client (void)
{
  ACE_NEW_RETURN (this->high_priority_client_,
                  Client (&this->client_thread_manager_,
                          this->ts_,
                          this->argc_,
                          this->argv_,
                          0),
                  -1);

#if defined (VXWORKS)
  // Set a task_id string starting with "@", so we are able to
  // accurately count the number of context switches.
  ACE_OS::strcpy (this->task_id_,
                  "@High");
#endif /* VXWORKS */

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

  ACE_DEBUG ((LM_DEBUG,
              "Creating 1 client with high priority of %d\n",
              this->high_priority_));
  if (this->high_priority_client_->activate (
      GLOBALS::instance ()->thr_create_flags,
      1,
      0,
      this->high_priority_,
      -1,
      0,
      0,
      0,
      0,
      (ACE_thread_t *) &this->task_id_) == -1)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "%p; priority is %d\n",
                       "activate failed",
                       this->high_priority_),
                      -1);
  return 0;
}

int
Client_i::activate_low_client (void)
{
  ACE_NEW_RETURN (this->low_priority_client_,
                  Client *[this->ts_->thread_count_],
                  -1);

  // Hack to make sure we have something in this pointer, when
  // thread_count == 1.
  this->low_priority_client_[0] =
    this->high_priority_client_;
  this->num_low_priority_ =
    this->ts_->thread_count_ - 1;

  // Set the priority for the low priority threads.
  this->init_low_priority ();

  ACE_DEBUG ((LM_DEBUG,
              "Creating %d clients at priority %d\n",
              this->ts_->thread_count_ - 1,
              this->low_priority_));

  for (u_int i = this->num_low_priority_;
       i > 0;
       i--)
    {
      ACE_NEW_RETURN (this->low_priority_client_ [i - 1],
                      Client (&this->client_thread_manager_,
                              this->ts_,
                              this->argc_,
                              this->argv_,
                              i),
                      -1);
#if defined (VXWORKS)
      // Pace the connection establishment on VxWorks.
      const ACE_Time_Value delay (0L, 500000L);
      ACE_OS::sleep (delay);

      // Set a task_id string startiing with "@", so we are able to
      // accurately count the number of context switches on VXWORKS
      sprintf (this->task_id_,
               "@Low%u",
               i);
#endif /* VXWORKS */
      ACE_DEBUG ((LM_DEBUG,
                  "Creating client with thread ID %d and priority %d\n",
                  i,
                  this->low_priority_));
      // The first thread starts at the lowest priority of all the low
      // priority clients.
      if (this->low_priority_client_[i - 1]->activate (
          GLOBALS::instance ()->thr_create_flags,
          1,
          0,
          this->low_priority_, // These are constructor defaults.
          -1,                  // int grp_id = -1,
          0,                   // ACE_Task_Base *task = 0,
          0,                   // ACE_hthread_t thread_handles[] = 0,
          0,                   // void *stack[] = 0,
          0,                   // size_t stack_size[] = 0,
          (ACE_thread_t *) &this->task_id_) == -1)
        ACE_ERROR ((LM_ERROR,
                    "%p; priority is %d\n",
                    "activate failed",
                    this->low_priority_));

      if (this->ts_->use_multiple_priority_ == 1)
        {
          this->counter_ = (this->counter_ + 1) % this->grain_;

          if (this->counter_ == 0
              // Just so when we distribute the priorities among the
              // threads, we make sure we don't go overboard.
              && this->num_priorities_ * this->grain_ > this->num_low_priority_ - (i - 1))
            // Get the next higher priority.
            this->low_priority_ = ACE_Sched_Params::next_priority
              (ACE_SCHED_FIFO, this->low_priority_, ACE_SCOPE_THREAD);
        }
    } /* end of for () */

  return 0;
}