TAO/performance-tests/RTCorba/Oneways/Reliable/client.cpp

   1 #include "testC.h"
   2
   3 #include "tao/Messaging/Messaging.h"
   4 #include "tao/AnyTypeCode/TAOA.h"
   5 #include "tao/AnyTypeCode/Any.h"
   6 #include "tao/TAOC.h"
   7 #include "ace/Get_Opt.h"
   8 #include "ace/Sched_Params.h"
   9 #include "ace/Stats.h"
  10 #include "ace/Throughput_Stats.h"
  11 #include "ace/High_Res_Timer.h"
  12 #include "ace/OS_NS_errno.h"
  13 #include "ace/OS_NS_string.h"
  14
  15 #define USING_TIMERS
  16 //#define USING_QUANTIFY
  17
  18 #if defined (USING_QUANTIFY)
  19
  20 #if defined (ACE_WIN32)
  21
  22 #include "pure.h"
  23
  24 #else /* !ACE_WIN32 */
  25
  26 #include "quantify.h"
  27
  28 inline int QuantifyClearData ()
  29 {
  30   return quantify_clear_data ();
  31 }
  32
  33 inline int QuantifyStartRecordingData ()
  34 {
  35   return quantify_start_recording_data ();
  36 }
  37
  38 inline int QuantifyStopRecordingData ()
  39 {
  40   return quantify_stop_recording_data ();
  41 }
  42
  43 #endif /* ACE_WIN32 */
  44
  45 #endif /* USING_QUANTIFY */
  46
  47 // Default IOR.
  48 static const ACE_TCHAR *ior = ACE_TEXT("file://test.ior");
  49
  50 // Levels at which syncscope policy can be set.
  51 enum LEVEL {ORB_LEVEL, THREAD_LEVEL, OBJECT_LEVEL};
  52
  53 // Default is OBJECT level.
  54 static LEVEL level = OBJECT_LEVEL;
  55
  56 // Default iterations.
  57 static CORBA::ULong iterations = 100;
  58
  59 // Default amount of work.
  60 static CORBA::ULong work = 0;
  61
  62 // Benchmark payload based operations?
  63 static int payload_test = 0;
  64
  65 // Default payload size.
  66 static CORBA::ULong payload_size = 0;
  67
  68 // Default number of invocations to buffer before flushing.
  69 static CORBA::ULong buffering_queue_size = iterations / 2;
  70
  71 // Benchmark the twoway operation?
  72 static int test_twoway = 0;
  73
  74 // Shut down server after test?
  75 static int shutdown_server = 0;
  76
  77 // Default SyncScope value.
  78 static Messaging::SyncScope sync_scope = Messaging::SYNC_WITH_TRANSPORT;
  79
  80 // Global scale factor.
  81 static ACE_High_Res_Timer::global_scale_factor_type gsf = 0;
  82
  83 static void
  84 print_params ()
  85 {
  86   if (test_twoway)
  87     {
  88       ACE_DEBUG ((LM_DEBUG,
  89                   "\nTesting twoway requests\n"));
  90     }
  91   else
  92     {
  93       const char *one_way_style = 0;
  94       if (sync_scope == Messaging::SYNC_NONE)
  95         one_way_style = "SYNC_NONE";
  96       else if (sync_scope == Messaging::SYNC_WITH_TRANSPORT)
  97         one_way_style = "SYNC_WITH_TRANSPORT";
  98       else if (sync_scope == Messaging::SYNC_WITH_SERVER)
  99         one_way_style = "SYNC_WITH_SERVER";
 100       else if (sync_scope == Messaging::SYNC_WITH_TARGET)
 101         one_way_style = "SYNC_WITH_TARGET";
 102
 103       const char *payload_style = 0;
 104       if (payload_test)
 105         payload_style = "Payload based";
 106       else
 107         payload_style = "Work based";
 108
 109       ACE_DEBUG ((LM_DEBUG,
 110                   "\nTesting oneway requests: %C : %C\n",
 111                   one_way_style,
 112                   payload_style));
 113
 114       if (sync_scope == Messaging::SYNC_NONE)
 115         {
 116           ACE_DEBUG ((LM_DEBUG,
 117                       "Request queue limited to %d messages\n",
 118                       buffering_queue_size));
 119         }
 120
 121       if (payload_test)
 122         {
 123           ACE_DEBUG ((LM_DEBUG,
 124                       "Payload size %d bytes\n",
 125                       payload_size));
 126         }
 127     }
 128
 129   ACE_DEBUG ((LM_DEBUG,
 130               "%d invocations\n",
 131               iterations));
 132 }
 133
 134 static void
 135 twoway_work_test (Test_ptr server)
 136 {
 137 #if defined (USING_TIMERS)
 138   ACE_Throughput_Stats latency;
 139   ACE_hrtime_t base = ACE_OS::gethrtime ();
 140 #endif /* USING_TIMERS */
 141
 142 #if defined (USING_QUANTIFY)
 143   // Reset Quantify data recording; whatever happened in the past is
 144   // not relevant to this test.
 145   QuantifyClearData ();
 146   QuantifyStartRecordingData ();
 147 #endif /* USING_QUANTIFY */
 148
 149   for (CORBA::ULong i = 0; i != iterations; ++i)
 150     {
 151 #if defined (USING_TIMERS)
 152       ACE_hrtime_t latency_base = ACE_OS::gethrtime ();
 153 #endif /* USING_TIMERS */
 154
 155       server->twoway_work_test (work);
 156
 157 #if defined (USING_TIMERS)
 158       ACE_hrtime_t now = ACE_OS::gethrtime ();
 159
 160       latency.sample (now - base,
 161                       now - latency_base);
 162 #endif /* USING_TIMERS */
 163     }
 164
 165 #if defined (USING_QUANTIFY)
 166   // Stop recording data here; whatever happens after this in the test
 167   // is not relevant to this test.
 168   QuantifyStopRecordingData ();
 169 #endif /* USING_QUANTIFY */
 170
 171 #if defined (USING_TIMERS)
 172   latency.dump_results (ACE_TEXT("Twoway"), gsf);
 173 #endif /* USING_TIMERS */
 174 }
 175
 176 static void
 177 oneway_work_test (Test_ptr server)
 178 {
 179 #if defined (USING_TIMERS)
 180   ACE_Throughput_Stats latency;
 181   ACE_hrtime_t base = ACE_OS::gethrtime ();
 182 #endif /* USING_TIMERS */
 183
 184 #if defined (USING_QUANTIFY)
 185   // Reset Quantify data recording; whatever happened in the past is
 186   // not relevant to this test.
 187   QuantifyClearData ();
 188   QuantifyStartRecordingData ();
 189 #endif /* USING_QUANTIFY */
 190
 191   for (CORBA::ULong i = 0; i != iterations; ++i)
 192     {
 193 #if defined (USING_TIMERS)
 194       ACE_hrtime_t latency_base = ACE_OS::gethrtime ();
 195 #endif /* USING_TIMERS */
 196
 197       server->oneway_work_test (work);
 198
 199 #if defined (USING_TIMERS)
 200       ACE_hrtime_t now = ACE_OS::gethrtime ();
 201
 202       latency.sample (now - base,
 203                       now - latency_base);
 204 #endif /* USING_TIMERS */
 205     }
 206
 207 #if defined (USING_QUANTIFY)
 208   // Stop recording data here; whatever happens after this in the test
 209   // is not relevant to this test.
 210   QuantifyStopRecordingData ();
 211 #endif /* USING_QUANTIFY */
 212
 213 #if defined (USING_TIMERS)
 214   latency.dump_results (ACE_TEXT("Oneway (work based)"), gsf);
 215 #endif /* USING_TIMERS */
 216 }
 217
 218 static void
 219 oneway_payload_test (Test_ptr server)
 220 {
 221 #if defined (USING_TIMERS)
 222   ACE_Throughput_Stats latency;
 223   ACE_hrtime_t base = ACE_OS::gethrtime ();
 224 #endif /* USING_TIMERS */
 225
 226 #if defined (USING_QUANTIFY)
 227   // Reset Quantify data recording; whatever happened in the past is
 228   // not relevant to this test.
 229   QuantifyClearData ();
 230   QuantifyStartRecordingData ();
 231 #endif /* USING_QUANTIFY */
 232
 233   Test::data the_data (payload_size);
 234   the_data.length (payload_size);
 235
 236   for (CORBA::ULong i = 0; i != iterations; ++i)
 237     {
 238 #if defined (USING_TIMERS)
 239       ACE_hrtime_t latency_base = ACE_OS::gethrtime ();
 240 #endif /* USING_TIMERS */
 241
 242       server->oneway_payload_test (the_data);
 243
 244 #if defined (USING_TIMERS)
 245       ACE_hrtime_t now = ACE_OS::gethrtime ();
 246
 247       latency.sample (now - base,
 248                       now - latency_base);
 249 #endif /* USING_TIMERS */
 250     }
 251
 252 #if defined (USING_QUANTIFY)
 253   // Stop recording data here; whatever happens after this in the test
 254   // is not relevant to this test.
 255   QuantifyStopRecordingData ();
 256 #endif /* USING_QUANTIFY */
 257
 258 #if defined (USING_TIMERS)
 259   latency.dump_results (ACE_TEXT("Oneway (payload based)"), gsf);
 260 #endif /* USING_TIMERS */
 261 }
 262
 263 static int
 264 parse_args (int argc, ACE_TCHAR *argv[])
 265 {
 266   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("ps:k:i:t:l:m:w:x"));
 267   int error = 0;
 268   int c;
 269
 270   while ((c = get_opts ()) != -1)
 271     switch (c)
 272       {
 273       case 's':
 274         payload_size = ACE_OS::atoi (get_opts.opt_arg ());
 275         break;
 276
 277       case 'p':
 278         payload_test = 1;
 279         break;
 280
 281       case 'i':
 282         iterations = ACE_OS::atoi (get_opts.opt_arg ());
 283         break;
 284
 285       case 'k':
 286         ior = get_opts.opt_arg ();
 287         break;
 288
 289       case 't':
 290         {
 291           ACE_TCHAR *tmp = get_opts.opt_arg ();
 292
 293           if (!ACE_OS::strcmp (tmp, ACE_TEXT("none")))
 294             sync_scope = Messaging::SYNC_NONE;
 295           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("transport")))
 296             sync_scope = Messaging::SYNC_WITH_TRANSPORT;
 297           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("server")))
 298             sync_scope = Messaging::SYNC_WITH_SERVER;
 299           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("target")))
 300             sync_scope = Messaging::SYNC_WITH_TARGET;
 301           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("twoway")))
 302             test_twoway = 1;
 303           else
 304             error = 1;
 305           break;
 306         }
 307
 308       case 'l':
 309         {
 310           ACE_TCHAR *tmp = get_opts.opt_arg ();
 311
 312           if (!ACE_OS::strcmp (tmp, ACE_TEXT("orb")))
 313             level = ORB_LEVEL;
 314           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("thread")))
 315             level = THREAD_LEVEL;
 316           else if (!ACE_OS::strcmp (tmp, ACE_TEXT("object")))
 317             level = OBJECT_LEVEL;
 318           else
 319             error = 1;
 320           break;
 321         }
 322
 323       case 'm':
 324         buffering_queue_size = ACE_OS::atoi (get_opts.opt_arg ());
 325         break;
 326
 327       case 'w':
 328         work = ACE_OS::atoi (get_opts.opt_arg ());
 329         break;
 330
 331       case 'x':
 332         shutdown_server = 1;
 333         break;
 334
 335       case '?':
 336       default:
 337         error = 1;
 338         break;
 339       }
 340
 341   if (error)
 342     ACE_ERROR_RETURN ((LM_ERROR,
 343                        "usage:  %s "
 344                        "-k <ior> "
 345                        "-s <payload size> "
 346                        "-p <payload based test> "
 347                        "-i <# of iterations> "
 348                        "-t <none|transport|server|target|twoway> "
 349                        "-l <orb|thread|object> "
 350                        "-m <message count> "
 351                        "-w <# of server loops> "
 352                        "-x shutdown server "
 353                        "\n",
 354                        argv [0]),
 355                       -1);
 356
 357   // Indicates successful parsing of the command line
 358   return 0;
 359 }
 360
 361 int
 362 set_rt_mode ()
 363 {
 364   int policy = ACE_SCHED_FIFO;
 365   int priority =
 366     (ACE_Sched_Params::priority_min (policy)
 367      + ACE_Sched_Params::priority_max (policy)) / 2;
 368
 369   // Enable FIFO scheduling
 370   int result =
 371     ACE_OS::sched_params (ACE_Sched_Params (policy,
 372                                             priority,
 373                                             ACE_SCOPE_PROCESS));
 374   if (result != 0)
 375     {
 376       if (ACE_OS::last_error () == EPERM)
 377         {
 378           ACE_DEBUG ((LM_DEBUG,
 379                       "client (%P|%t): user is not superuser, "
 380                       "test runs in time-shared class\n"));
 381         }
 382       else
 383         ACE_ERROR ((LM_ERROR,
 384                     "client (%P|%t): sched_params failed %p\n",
 385                     "set_rt_mode"));
 386     }
 387
 388   // Get our thread handle.
 389   ACE_hthread_t self;
 390   ACE_OS::thr_self (self);
 391
 392   // Set our thread priority.
 393   if (ACE_OS::thr_setprio (self, priority) != 0)
 394     ACE_ERROR ((LM_ERROR,
 395                 "server (%P|%t):thr_setprio failed %p\n",
 396                 "set_rt_mode"));
 397
 398   // Do a sanity check.
 399   if (ACE_OS::thr_getprio (self, priority) == 0)
 400     ACE_DEBUG ((LM_DEBUG,
 401                 "client (%P|%t): thread priority = %d.\n",
 402                 priority));
 403
 404   return 0;
 405 }
 406
 407 int
 408 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 409 {
 410   int result = set_rt_mode ();
 411   if (result != 0)
 412     return result;
 413
 414   try
 415     {
 416       // Calibrate the timer.
 417       gsf = ACE_High_Res_Timer::global_scale_factor ();
 418
 419       // Initialize the ORB, the POA, and get the server reference.
 420       CORBA::ORB_var orb =
 421         CORBA::ORB_init (argc,
 422                          argv);
 423
 424       // Get the command line options.
 425       if (parse_args (argc, argv) != 0)
 426         {
 427           ACE_ERROR_RETURN ((LM_ERROR,
 428                              "parse_args failed\n"),
 429                             1);
 430         }
 431
 432       CORBA::Object_var obj =
 433         orb->resolve_initial_references ("ORBPolicyManager");
 434
 435       CORBA::PolicyManager_var policy_manager =
 436         CORBA::PolicyManager::_narrow (obj.in ());
 437
 438       obj = orb->resolve_initial_references ("PolicyCurrent");
 439
 440       CORBA::PolicyCurrent_var policy_current =
 441         CORBA::PolicyCurrent::_narrow (obj.in ());
 442
 443
 444       obj = orb->string_to_object (ior);
 445
 446       Test_var server = Test::_narrow (obj.in ());
 447
 448       // Print testing parameters.
 449       print_params ();
 450
 451       // Run the test.
 452       if (test_twoway)
 453         {
 454           twoway_work_test (server.in ());
 455         }
 456       else
 457         {
 458           // Set up the sync scope any.
 459           CORBA::Any sync_scope_any;
 460           sync_scope_any <<= sync_scope;
 461
 462           // Set the length of the policy list.
 463           CORBA::PolicyList sync_scope_policy_list (1);
 464           sync_scope_policy_list.length (1);
 465
 466           // Set up the sync scope policy.
 467           sync_scope_policy_list[0] =
 468             orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE,
 469                                 sync_scope_any);
 470
 471           switch (level)
 472           {
 473             case ORB_LEVEL:
 474               // Set the sync scope policy at the ORB level.
 475               policy_manager->set_policy_overrides (sync_scope_policy_list,
 476                                                     CORBA::ADD_OVERRIDE);
 477               break;
 478
 479             case THREAD_LEVEL:
 480               // Set the sync scope policy at the thread level.
 481               policy_current->set_policy_overrides (sync_scope_policy_list,
 482                                                     CORBA::ADD_OVERRIDE);
 483               break;
 484
 485             case OBJECT_LEVEL:
 486               // Set the sync scope policy at the object level.
 487               obj = server->_set_policy_overrides (sync_scope_policy_list,
 488                                                    CORBA::ADD_OVERRIDE);
 489
 490               // Get the new object reference with the updated policy.
 491               server = Test::_narrow (obj.in ());
 492               break;
 493
 494             default:
 495               break;
 496           }
 497
 498           // We are done with this policy.
 499           sync_scope_policy_list[0]->destroy ();
 500
 501           // Are we buffering the oneway requests?
 502           if (sync_scope == Messaging::SYNC_NONE)
 503             {
 504               TAO::BufferingConstraint buffering_constraint;
 505               buffering_constraint.mode = TAO::BUFFER_MESSAGE_COUNT;
 506               buffering_constraint.message_count = buffering_queue_size;
 507               buffering_constraint.message_bytes = 0;
 508               buffering_constraint.timeout = 0;
 509
 510               // Set up the buffering constraint any.
 511               CORBA::Any buffering_constraint_any;
 512               buffering_constraint_any <<= buffering_constraint;
 513
 514               // Set up the buffering constraint policy list.
 515               CORBA::PolicyList buffering_constraint_policy_list (1);
 516               buffering_constraint_policy_list.length (1);
 517
 518               // Set up the buffering constraint policy.
 519               buffering_constraint_policy_list[0] =
 520                 orb->create_policy (TAO::BUFFERING_CONSTRAINT_POLICY_TYPE,
 521                                     buffering_constraint_any);
 522
 523               // Set up the constraints (at the object level).
 524               obj = server->_set_policy_overrides (buffering_constraint_policy_list,
 525                                                    CORBA::ADD_OVERRIDE);
 526
 527               // We are done with this policy.
 528               buffering_constraint_policy_list[0]->destroy ();
 529
 530               // Get the new object reference with the updated policy.
 531               server = Test::_narrow (obj.in ());
 532             }
 533
 534           // Run the oneway test.
 535           if (payload_test)
 536             oneway_payload_test (server.in ());
 537           else
 538             oneway_work_test (server.in ());
 539         }
 540
 541       if (shutdown_server)
 542         {
 543           ACE_DEBUG ((LM_DEBUG,
 544                       "\nShutting down server\n"));
 545
 546           server->shutdown ();
 547         }
 548
 549       // Destroy the ORB. On some platforms, e.g., Win32, the socket
 550       // library is closed at the end of main().  This means that any
 551       // socket calls made after main() fail. Hence if we wait for
 552       // static destructors to flush the queues, it will be too late.
 553       // Therefore, we use explicit destruction here and flush the
 554       // queues before main() ends.
 555       orb->destroy ();
 556     }
 557   catch (const CORBA::Exception& ex)
 558     {
 559       ex._tao_print_exception ("client");
 560       return 1;
 561     }
 562
 563   return 0;
 564 }