client.cpp

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

// client.cpp,v 1.26 2003/10/28 18:34:34 bala Exp

#include "ace/Get_Opt.h"
#include "ace/High_Res_Timer.h"
#include "ace/Stats.h"
#include "ace/Sample_History.h"
#include "ace/Read_Buffer.h"
#include "ace/Array_Base.h"
#include "ace/Task.h"
#include "tao/ORB_Core.h"
#include "tao/debug.h"
#include "tao/RTCORBA/RTCORBA.h"
#include "tao/RTCORBA/Priority_Mapping_Manager.h"
#include "testC.h"
#include "tests/RTCORBA/common_args.cpp"
#include "tests/RTCORBA/check_supported_priorities.cpp"
#include "ace/Event.h"

ACE_RCSID(Thread_Pool, client, "client.cpp,v 1.26 2003/10/28 18:34:34 bala Exp")

enum Priority_Setting
{
  AT_THREAD_CREATION = 0,
  AFTER_THREAD_CREATION = 1
};

static const char *ior = "file://ior";
static const char *rates_file = "rates";
static const char *invocation_priorities_file = "empty-file";
static int shutdown_server = 0;
static int do_dump_history = 0;
static ACE_UINT32 gsf = 0;
static CORBA::ULong continuous_workers = 0;
static int done = 0;
static CORBA::ULong time_for_test = 10;
static CORBA::ULong work = 10;
static CORBA::ULong max_throughput_timeout = 5;
static CORBA::ULong continuous_worker_priority = 0;
static int set_priority = 1;
static Priority_Setting priority_setting = AFTER_THREAD_CREATION;
static int individual_continuous_worker_stats = 0;
static int print_missed_invocations = 0;
static ACE_hrtime_t test_start;
static CORBA::ULong prime_number = 9619;
static int count_missed_end_deadlines = 0;

struct Synchronizers
{
  Synchronizers (void)
    : worker_lock_ (),
      workers_ (1),
      workers_ready_ (0),
      number_of_workers_ (0)
  {
  }

  ACE_SYNCH_MUTEX worker_lock_;
  ACE_Event workers_;
  CORBA::ULong workers_ready_;
  CORBA::ULong number_of_workers_;
};

int
parse_args (int argc, char *argv[])
{
  ACE_Get_Opt get_opts (argc, argv,
                        "c:e:g:hi:k:m:p:q:r:t:u:v:w:x:y:z:" //client options
                        "b:f:hl:n:o:s:" // server options
                        );
  int c;

  while ((c = get_opts ()) != -1)
    switch (c)
      {
      case 'c':
        continuous_workers =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'e':
        count_missed_end_deadlines =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'g':
        do_dump_history =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'i':
        individual_continuous_worker_stats =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'k':
        ior =
          get_opts.opt_arg ();
        break;

      case 'm':
        print_missed_invocations =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'p':
        invocation_priorities_file =
          get_opts.opt_arg ();
        break;

      case 'q':
        prime_number =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'r':
        rates_file =
          get_opts.opt_arg ();
        break;

      case 't':
        time_for_test =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'u':
        continuous_worker_priority =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'v':
        priority_setting =
          Priority_Setting (ACE_OS::atoi (get_opts.opt_arg ()));
        break;

      case 'w':
        work =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'x':
        shutdown_server =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'y':
        set_priority =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'z':
        max_throughput_timeout =
          ACE_OS::atoi (get_opts.opt_arg ());
        break;

      case 'b':
      case 'f':
      case 'l':
      case 'n':
      case 'o':
      case 's':
        // server options: ignored.
        break;

      case 'h':
      case '?':
      default:
        ACE_ERROR_RETURN ((LM_ERROR,
                           "usage:  %s\n"
                           "\t-c <number of continuous workers> (defaults to %d)\n"
                           "\t-g <count missed end deadlines> (defaults to %d)\n"
                           "\t-g <show history> (defaults to %d)\n"
                           "\t-h <help: shows options menu>\n"
                           "\t-i <print stats of individual continuous workers> (defaults to %d)\n"
                           "\t-k <ior> (defaults to %s)\n"
                           "\t-m <print missed invocations for paced workers> (defaults to %d)\n"
                           "\t-p <invocation priorities file> (defaults to %s)\n"
                           "\t-q <prime number> (defaults to %d)\n"
                           "\t-r <rates file> (defaults to %s)\n"
                           "\t-t <time for test> (defaults to %d)\n"
                           "\t-u <continuous worker priority> (defaults to %d)\n"
                           "\t-v <priority setting: AT_THREAD_CREATION = 0, AFTER_THREAD_CREATION = 1> (defaults to %s)\n"
                           "\t-w <work> (defaults to %d)\n"
                           "\t-x <shutdown server> (defaults to %d)\n"
                           "\t-y <set invocation priorities> (defaults to %d)\n"
                           "\t-z <timeout for max throughput measurement> (defaults to %d)\n"
                           "\n",
                           argv [0],
                           continuous_workers,
                           count_missed_end_deadlines,
                           do_dump_history,
                           individual_continuous_worker_stats,
                           ior,
                           print_missed_invocations,
                           invocation_priorities_file,
                           prime_number,
                           rates_file,
                           time_for_test,
                           continuous_worker_priority,
                           priority_setting == 0 ? "AT_THREAD_CREATION" : "AFTER_THREAD_CREATION",
                           work,
                           shutdown_server,
                           set_priority,
                           max_throughput_timeout),
                          -1);
      }

  return 0;
}

double
to_seconds (ACE_UINT64 hrtime,
            ACE_UINT32 sf)
{
  double seconds =
#if defined ACE_LACKS_LONGLONG_T
    hrtime / sf;
#else  /* ! ACE_LACKS_LONGLONG_T */
  ACE_static_cast (double,
                   ACE_UINT64_DBLCAST_ADAPTER (hrtime / sf));
#endif /* ! ACE_LACKS_LONGLONG_T */
  seconds /= ACE_HR_SCALE_CONVERSION;

  return seconds;
}

ACE_UINT64
to_hrtime (double seconds,
           ACE_UINT32 sf)
{
  return ACE_UINT64 (seconds * sf * ACE_HR_SCALE_CONVERSION);
}

int
start_synchronization (test_ptr test,
                       Synchronizers &synchronizers)
{
  CORBA::ULong synchronization_iterations = 1;
  ACE_TRY_NEW_ENV
    {
      for (CORBA::ULong i = 0;
           i < synchronization_iterations;
           ++i)
        {
          test->method (work,
                        prime_number
                        ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
                           "Exception caught:");
      return -1;
    }
  ACE_ENDTRY;

  {
    ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
                      mon,
                      synchronizers.worker_lock_,
                      -1);

    if (synchronizers.workers_ready_ == 0)
      {
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "\n"));
      }

    ++synchronizers.workers_ready_;

    if (TAO_debug_level > 0)
      ACE_DEBUG ((LM_DEBUG,
                  "%d worker ready\n",
                  synchronizers.workers_ready_));

    if (synchronizers.workers_ready_ ==
        synchronizers.number_of_workers_)
      {
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "\n"));

        test_start =
          ACE_OS::gethrtime ();

        synchronizers.workers_.signal ();

        return 0;
      }
  }

  synchronizers.workers_.wait ();

  return 0;
}

int
end_synchronization (Synchronizers &synchronizers)
{
  {
    ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
                      mon,
                      synchronizers.worker_lock_,
                      -1);

    if (synchronizers.workers_ready_ ==
        synchronizers.number_of_workers_)
      {
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "\n"));

        synchronizers.workers_.reset ();
      }

    if (TAO_debug_level > 0)
      ACE_DEBUG ((LM_DEBUG,
                  "%d worker completed\n",
                  synchronizers.workers_ready_));

    --synchronizers.workers_ready_;

    if (synchronizers.workers_ready_ == 0)
      {
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,
                      "\n"));

        synchronizers.workers_.signal ();

        return 0;
      }
  }

  synchronizers.workers_.wait ();

  return 0;
}

int
max_throughput (test_ptr test,
                RTCORBA::Current_ptr current,
                RTCORBA::PriorityMapping &priority_mapping,
                CORBA::ULong &max_rate)
{
  CORBA::ULong calls_made = 0;
  CORBA::Short CORBA_priority = 0;
  CORBA::Short native_priority = 0;

  ACE_TRY_NEW_ENV
    {
      CORBA_priority =
        current->the_priority (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;

      CORBA::Boolean result =
        priority_mapping.to_native (CORBA_priority,
                                    native_priority);
      if (!result)
        ACE_ERROR_RETURN ((LM_ERROR,
                           "Error in converting CORBA priority %d to native priority\n",
                           CORBA_priority),
                          -1);

      ACE_hrtime_t start =
        ACE_OS::gethrtime ();

      ACE_hrtime_t end =
        start +
        to_hrtime (max_throughput_timeout, gsf);

      for (;;)
        {
          ACE_hrtime_t now =
            ACE_OS::gethrtime ();

          if (now > end)
            break;

          test->method (work,
                        prime_number
                        ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;

          ++calls_made;
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
                           "Exception caught:");
      return -1;
    }
  ACE_ENDTRY;

  max_rate =
    calls_made / max_throughput_timeout;

  ACE_DEBUG ((LM_DEBUG,
              "\nPriority = %d/%d; Max rate calculations => %d calls in %d seconds; Max rate = %.2f\n",
              CORBA_priority,
              native_priority,
              calls_made,
              max_throughput_timeout,
              calls_made / (double) max_throughput_timeout));