TAO/tests/Oneway_Timeouts/client.cpp

   1 #include "TestS.h"
   2 #include "tao/Strategies/advanced_resource.h"
   3 #include "tao/Messaging/Messaging.h"
   4 #include "tao/AnyTypeCode/TAOA.h"
   5 #include "tao/AnyTypeCode/Any.h"
   6
   7 #include "ace/streams.h"
   8 #include "ace/High_Res_Timer.h"
   9 #include "ace/Arg_Shifter.h"
  10
  11 #include <cmath>
  12
  13 using namespace CORBA;
  14 using namespace PortableServer;
  15
  16 namespace
  17 {
  18   const ACE_TCHAR *non_existent_ior = ACE_TEXT("corbaloc:iiop:1.2@63.246.9.65:12345/test");
  19   const int TIME_THRESHOLD = 100; //ms
  20
  21   int request_timeout = 0;
  22   Messaging::SyncScope sync_scope;
  23   bool use_buf_constraints = false;
  24   bool use_sync_scope = false;
  25   int bc_mode = TAO::BUFFER_FLUSH;
  26   int bc_count = 0;
  27   int bc_bytes = 0;
  28   int bc_timeout = 0;
  29   int num_requests = 10;
  30   int request_interval = 50;
  31   int connect_timeout = 0;
  32   int run_orb_delay = 0;
  33   int run_orb_time = 0;
  34   bool force_timeout = false;
  35   // This will force a blocking connection before starting the test
  36   // by sending the num_requests as a twoway.
  37   bool force_connect = false;
  38   bool use_sleep = false;
  39   unsigned int max_request_time = 0;
  40   bool use_twoway = false;
  41   bool retry_transients = false;
  42   bool retry_timeouts = false;
  43   bool make_request_queued = false;
  44   const ACE_TCHAR *server_ior = ACE_TEXT ("file://test.ior");
  45
  46   void print_usage (const ACE_TCHAR *argv0)
  47   {
  48     ACE_ERROR ((LM_ERROR,
  49                 "%s [-request_timeout ms=0] "
  50                 "[-connect_timeout ms=0] "
  51                 "[-request_interval ms=100] "
  52                 "[-run_orb_delay ms=0] "
  53                 "[-run_orb_time ms=0] "
  54                 "[-max_request_time ms=0] "
  55                 "[-num_requests n=10] "
  56                 "[-use_twoway] "
  57                 "[-retry_transients] "
  58                 "[-retry_timeouts] "
  59                 "[-use_sleep] "
  60                 "[-force_timeout] "
  61                 "[-force_connect] "
  62                 "[-buffer_count n=0]"
  63                 "[-buffer_bytes n=0] "
  64                 "[-buffer_timeout ms=0] "
  65                 "[-sync delayed|none] "
  66                 "[-make_request_queued] "
  67                 "[-server_ior <ior>]\n",
  68                 argv0));
  69   }
  70
  71   bool parse_command_line (int ac, ACE_TCHAR *av[])
  72   {
  73     ACE_Arg_Shifter args (ac, av);
  74     args.consume_arg ();
  75
  76     while (args.is_anything_left ())
  77       {
  78         if (args.cur_arg_strncasecmp (ACE_TEXT("-request_timeout")) == 0)
  79           {
  80             args.consume_arg ();
  81             request_timeout = ACE_OS::atoi (args.get_current ());
  82             args.consume_arg ();
  83           }
  84         else if (args.cur_arg_strncasecmp (ACE_TEXT("-connect_timeout")) == 0)
  85           {
  86             args.consume_arg ();
  87             connect_timeout = ACE_OS::atoi (args.get_current ());
  88             args.consume_arg ();
  89           }
  90         else if (args.cur_arg_strncasecmp (ACE_TEXT("-request_interval")) == 0)
  91           {
  92             args.consume_arg ();
  93             request_interval = ACE_OS::atoi (args.get_current ());
  94             args.consume_arg ();
  95           }
  96         else if (args.cur_arg_strncasecmp (ACE_TEXT("-run_orb_delay")) == 0)
  97           {
  98             args.consume_arg ();
  99             run_orb_delay = ACE_OS::atoi (args.get_current ());
 100             args.consume_arg ();
 101           }
 102         else if (args.cur_arg_strncasecmp (ACE_TEXT("-run_orb_time")) == 0)
 103           {
 104             args.consume_arg ();
 105             run_orb_time = ACE_OS::atoi(args.get_current ());
 106             args.consume_arg ();
 107           }
 108         else if (args.cur_arg_strncasecmp (ACE_TEXT("-max_request_time")) == 0)
 109           {
 110             args.consume_arg ();
 111             max_request_time = ACE_OS::atoi (args.get_current ());
 112             args.consume_arg ();
 113           }
 114         else if (args.cur_arg_strncasecmp (ACE_TEXT("-num_requests")) == 0)
 115           {
 116             args.consume_arg ();
 117             num_requests = ACE_OS::atoi (args.get_current ());
 118             args.consume_arg ();
 119           }
 120         else if (args.cur_arg_strncasecmp (ACE_TEXT("-use_twoway")) == 0)
 121           {
 122             use_twoway = true;
 123             args.consume_arg ();
 124           }
 125         else if (args.cur_arg_strncasecmp (ACE_TEXT("-retry_transients")) == 0)
 126           {
 127             retry_transients = true;
 128             args.consume_arg ();
 129           }
 130         else if (args.cur_arg_strncasecmp (ACE_TEXT("-retry_timeouts")) == 0)
 131           {
 132             retry_timeouts = true;
 133             args.consume_arg ();
 134           }
 135         else if (args.cur_arg_strncasecmp (ACE_TEXT("-use_sleep")) == 0)
 136           {
 137             use_sleep = true;
 138             args.consume_arg ();
 139           }
 140         else if (args.cur_arg_strncasecmp (ACE_TEXT("-force_timeout")) == 0)
 141           {
 142             force_timeout = true;
 143             args.consume_arg ();
 144           }
 145         else if (args.cur_arg_strncasecmp (ACE_TEXT("-force_connect")) == 0)
 146           {
 147             force_connect = true;
 148             args.consume_arg ();
 149           }
 150         else if (args.cur_arg_strncasecmp (ACE_TEXT("-buffer_count")) == 0)
 151           {
 152             args.consume_arg ();
 153             use_buf_constraints = true;
 154             bc_count =  ACE_OS::atoi (args.get_current ());
 155             args.consume_arg ();
 156           }
 157         else if (args.cur_arg_strncasecmp (ACE_TEXT("-buffer_bytes")) == 0)
 158           {
 159             args.consume_arg ();
 160             use_buf_constraints = true;
 161             bc_bytes = ACE_OS::atoi (args.get_current ());
 162             args.consume_arg ();
 163           }
 164         else if (args.cur_arg_strncasecmp (ACE_TEXT("-buffer_timeout")) == 0)
 165           {
 166             args.consume_arg ();
 167             use_buf_constraints = true;
 168             bc_timeout = ACE_OS::atoi (args.get_current ());
 169             args.consume_arg ();
 170           }
 171         else if (args.cur_arg_strncasecmp (ACE_TEXT("-sync")) == 0)
 172         {
 173           args.consume_arg ();
 174           if (args.cur_arg_strncasecmp (ACE_TEXT("delayed")) == 0)
 175           {
 176             sync_scope = TAO::SYNC_DELAYED_BUFFERING;
 177             use_sync_scope = true;
 178           }
 179           else if (args.cur_arg_strncasecmp (ACE_TEXT("none")) == 0)
 180           {
 181             sync_scope = Messaging::SYNC_NONE;
 182             use_sync_scope = true;
 183           }
 184           else
 185           {
 186             print_usage (av[0]);
 187             return false;
 188           }
 189
 190           args.consume_arg ();
 191         }
 192         else if (args.cur_arg_strncasecmp (ACE_TEXT("-make_request_queued")) == 0)
 193         {
 194           make_request_queued = true;
 195           args.consume_arg ();
 196         }
 197         else if (args.cur_arg_strncasecmp (ACE_TEXT("-server_ior")) == 0)
 198           {
 199             args.consume_arg ();
 200             server_ior = args.get_current ();
 201             args.consume_arg ();
 202           }
 203         else
 204           {
 205             ACE_ERROR ((LM_ERROR, "Error: Unknown argument \"%s\"\n",
 206                         args.get_current ()));
 207             print_usage (av[0]);
 208             return false;
 209           }
 210
 211       }
 212
 213     return true;
 214   }
 215
 216   Tester_ptr set_request_timeout (Tester_ptr tst, ORB_ptr orb)
 217   {
 218     if (request_timeout <= 0)
 219       {
 220         return Tester::_duplicate (tst);
 221       }
 222
 223     Any a;
 224     a <<= static_cast<TimeBase::TimeT> (request_timeout * 10000);
 225     PolicyList pols (1);
 226     pols.length (1);
 227     pols[0] =
 228       orb->create_policy (Messaging::RELATIVE_RT_TIMEOUT_POLICY_TYPE, a);
 229     Object_var obj = tst->_set_policy_overrides (pols, ADD_OVERRIDE);
 230     pols[0]->destroy ();
 231     return Tester::_unchecked_narrow (obj.in ());
 232   }
 233
 234
 235   void set_connect_timeout (ORB_ptr orb)
 236   {
 237     if (connect_timeout <= 0)
 238       return;
 239     Object_var obj = orb->resolve_initial_references ("PolicyCurrent");
 240     PolicyCurrent_var policy_current = PolicyCurrent::_narrow (obj.in ());
 241     Any a;
 242     a <<= static_cast<TimeBase::TimeT> (connect_timeout * 10000);
 243     PolicyList pols (1);
 244     pols.length (1);
 245     pols[0] = orb->create_policy (TAO::CONNECTION_TIMEOUT_POLICY_TYPE, a);
 246     policy_current->set_policy_overrides (pols, ADD_OVERRIDE);
 247     pols[0]->destroy ();
 248   }
 249
 250
 251   void set_buffering (ORB_ptr orb)
 252   {
 253     Object_var obj = orb->resolve_initial_references ("PolicyCurrent");
 254     PolicyCurrent_var policy_current = PolicyCurrent::_narrow (obj.in ());
 255     PolicyList pols (1);
 256     pols.length (1);
 257
 258     if (use_sync_scope)
 259       {
 260         Any a;
 261         if (make_request_queued)
 262           a <<= Messaging::SYNC_NONE;
 263         else
 264           a <<= sync_scope;
 265         pols[0] = orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE, a);
 266         policy_current->set_policy_overrides (pols, ADD_OVERRIDE);
 267         pols[0]->destroy ();
 268       }
 269
 270
 271     if (use_buf_constraints)
 272       {
 273         TAO::BufferingConstraint bc;
 274         if (bc_count > 0)
 275           {
 276             bc_mode |= TAO::BUFFER_MESSAGE_COUNT;
 277           }
 278
 279         if (bc_bytes > 0)
 280           {
 281             bc_mode |= TAO::BUFFER_MESSAGE_BYTES;
 282           }
 283
 284         if (bc_timeout > 0)
 285           {
 286             bc_mode |= TAO::BUFFER_TIMEOUT;
 287           }
 288
 289         bc.mode = bc_mode;
 290         bc.message_count = bc_count;
 291         bc.message_bytes = bc_bytes;
 292         bc.timeout = static_cast<TimeBase::TimeT> (bc_timeout * 10000);
 293         Any a;
 294         a <<= bc;
 295         pols[0] =
 296           orb->create_policy (TAO::BUFFERING_CONSTRAINT_POLICY_TYPE, a);
 297         policy_current->set_policy_overrides (pols, ADD_OVERRIDE);
 298         pols[0]->destroy ();
 299       }
 300
 301   }
 302
 303
 304
 305 void reset_buffering (ORB_ptr orb)
 306 {
 307   Object_var obj = orb->resolve_initial_references ("PolicyCurrent");
 308   PolicyCurrent_var policy_current = PolicyCurrent::_narrow (obj.in ());
 309   PolicyList pols (1);
 310   pols.length (1);
 311
 312   Any a;
 313   a <<= sync_scope;
 314   pols[0] = orb->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE, a);
 315   policy_current->set_policy_overrides (pols, ADD_OVERRIDE);
 316   pols[0]->destroy ();
 317 }
 318
 319
 320 }
 321
 322 int ACE_TMAIN (int ac, ACE_TCHAR *av[])
 323 {
 324
 325   ACE_Time_Value before = ACE_High_Res_Timer::gettimeofday_hr ();
 326
 327   int num_requests_sent = 0;
 328
 329   try
 330     {
 331       ORB_var orb = ORB_init (ac, av);
 332
 333       if (!parse_command_line (ac, av))
 334         {
 335           return 1;
 336         }
 337
 338       set_connect_timeout (orb.in ());
 339       set_buffering (orb.in ());
 340
 341       ACE_TString ior (server_ior);
 342       if (force_timeout)
 343         {
 344           ior = non_existent_ior;
 345         }
 346
 347       Object_var obj = orb->string_to_object (ior.c_str ());
 348
 349       ACE_ASSERT (! is_nil (obj.in ()));
 350
 351       Tester_var tmp_tester;
 352       if (force_connect)
 353         {
 354           tmp_tester = Tester::_narrow (obj.in ());
 355           tmp_tester->test2 (-2);
 356           ACE_DEBUG ((LM_DEBUG, "Connected...\n"));
 357         }
 358       else
 359         tmp_tester = Tester::_unchecked_narrow (obj.in ());
 360
 361       Tester_var tester = set_request_timeout (tmp_tester.in (), orb.in ());
 362
 363       ACE_ASSERT (! is_nil (tester.in ()));
 364
 365       Long i = 0;
 366
 367       // Using make_request_queued option to work around test failure due to
 368       // different connection establishment behavior between OS.  Some system
 369       // can connect immediately and some may take longer time. With the flag on,
 370       // the test sets the SYNC_NONE scope and sends a request so the transport
 371       // queue is not empty for some SYNC_DELAYED_BUFFERING test case and hence
 372       // the requests are all queued and will be received by server continuously
 373       // during a short period.
 374       if (make_request_queued)
 375       {
 376         //Send this message while using SYNC_NONE.
 377         //This would leave the request in transport queue.
 378         tester->test (-3);
 379         //Set to SYNC_DELAYED_BUFFERING.
 380         //The requests will be queued since queue is not
 381         //empty.
 382         reset_buffering (orb.in ());
 383       }
 384
 385       for (; i < num_requests; ++i)
 386         {
 387           before = ACE_High_Res_Timer::gettimeofday_hr ();
 388           try
 389             {
 390               if (use_twoway)
 391                 {
 392                   tester->test2 (i);
 393                 }
 394               else
 395                 {
 396                   tester->test (i);
 397                 }
 398
 399             }
 400           catch (const CORBA::TRANSIENT&)
 401             {
 402               ACE_DEBUG ((LM_DEBUG,
 403                           "Transient exception during test () invocation %d\n",
 404                           i));
 405               if (retry_transients)
 406                 ACE_DEBUG ((LM_DEBUG,"retrying\n"));
 407               else
 408                 throw;
 409             }
 410           catch (const CORBA::TIMEOUT&)
 411             {
 412               ACE_DEBUG ((LM_DEBUG,
 413                           "Timeout exception during test () invocation %d\n",
 414                           i));
 415
 416               // This exception is expected with forced timeouts, since the
 417               // IOR is invalid. Unless we want to retry the invocation
 418               // go ahead and rethrow and let the outer catch deal with it.
 419               // Likewise if force_timeouts is not set, then throw it anyway
 420               // because the exception should not occur because these are
 421               // oneways.
 422               if (retry_timeouts)
 423                 ACE_DEBUG ((LM_DEBUG,"retrying\n"));
 424               else
 425                 throw;
 426             }
 427
 428           ++num_requests_sent;
 429
 430           ACE_Time_Value after = ACE_High_Res_Timer::gettimeofday_hr ();
 431           if (max_request_time > 0 &&
 432               (after - before).msec () > max_request_time)
 433             {
 434               ACE_DEBUG ((LM_DEBUG,
 435                           "note: test() took %d ms, max is %d ms\n",
 436                           (after - before).msec (), max_request_time));
 437             }
 438
 439           ACE_DEBUG ((LM_DEBUG, "c%d\n", i));
 440
 441           if (request_interval > 0)
 442             {
 443               ACE_Time_Value tv (0, request_interval * 1000);
 444               ACE_Time_Value done = tv +
 445                 ACE_High_Res_Timer::gettimeofday_hr ();
 446               if (! use_sleep)
 447                 {
 448                   orb->run (tv);
 449                 }
 450               else
 451                 {
 452                   ACE_OS::sleep (tv);
 453                 }
 454
 455               while (ACE_High_Res_Timer::gettimeofday_hr () < done)
 456                 {
 457                   ACE_OS::sleep (0);
 458                 }
 459             }
 460         }
 461
 462       ACE_DEBUG ((LM_DEBUG,"request loop complete\n"));
 463
 464
 465       if (run_orb_delay > 0)
 466         {
 467           ACE_Time_Value tv (0, run_orb_delay * 1000);
 468           ACE_OS::sleep (tv);
 469         }
 470
 471
 472       if (run_orb_time > 0)
 473         {
 474           ACE_Time_Value tv (0, run_orb_time * 1000);
 475           orb->run (tv);
 476         }
 477
 478       ACE_DEBUG ((LM_DEBUG,"Sending synch request to shutdown server\n"));
 479       use_twoway = true;
 480       use_sync_scope = false;
 481
 482       if (force_timeout)
 483         {
 484           // we have one more invocation that may time out.
 485           before = ACE_High_Res_Timer::gettimeofday_hr ();
 486           tester->test2 (-1);
 487         }
 488       else
 489         {
 490           // At this point, we aren't interested in the time it takes, we
 491           // want the peer to shut down, so use the non-timeout reference.
 492           // BUT IF THIS DOES raise a timeout, it will be reported as an
 493           // error.
 494           tmp_tester->test2 (-1);
 495         }
 496
 497       orb->shutdown (1);
 498
 499       orb->destroy ();
 500
 501       if (force_timeout)
 502         {
 503           ACE_ERROR_RETURN ((LM_ERROR,
 504                              "Error: Connection did not time out.\n"),
 505                             1);
 506         }
 507
 508       return 0;
 509     }
 510   catch (const CORBA::TRANSIENT &)
 511     {
 512       ACE_DEBUG ((LM_DEBUG, "caught transient exception\n"));
 513       if (force_timeout)
 514         {
 515           ACE_Time_Value after = ACE_High_Res_Timer::gettimeofday_hr ();
 516           long ms = (after - before).msec ();
 517           if ( (use_twoway || !use_sync_scope)
 518               && request_timeout > 0
 519               && request_timeout < connect_timeout)
 520             {
 521               connect_timeout = request_timeout;
 522             }
 523           else if (use_sync_scope && !use_sleep)
 524             {
 525               if (ms > TIME_THRESHOLD)
 526                 {
 527                   ACE_DEBUG ((LM_DEBUG,
 528                               "note: Buffered request took %dms\n", ms));
 529                 }
 530
 531               ms = num_requests_sent * request_interval;
 532             }
 533
 534           if (ms - connect_timeout > TIME_THRESHOLD ||
 535               connect_timeout - ms > TIME_THRESHOLD)
 536             {
 537               ACE_DEBUG ((LM_DEBUG,
 538                           "note: Timeout expected in %d ms, "
 539                           "but took %d ms\n", connect_timeout, ms));
 540             }
 541
 542           return 0;
 543         }
 544       else
 545         {
 546           ACE_ERROR_RETURN ((LM_ERROR, "Error: Unexpected\n"), 1);
 547         }
 548     }
 549   catch (const CORBA::TIMEOUT &)
 550     {
 551       ACE_DEBUG ((LM_DEBUG, "caught timeout exception\n"));
 552       if (force_timeout)
 553         {
 554           ACE_Time_Value after = ACE_High_Res_Timer::gettimeofday_hr ();
 555           long ms = (after - before).msec ();
 556           if ( (use_twoway || !use_sync_scope)
 557               && request_timeout > 0
 558               && request_timeout < connect_timeout)
 559             {
 560               connect_timeout = request_timeout;
 561             }
 562           else if (use_sync_scope && !use_sleep)
 563             {
 564               if (ms > TIME_THRESHOLD)
 565                 {
 566                   ACE_DEBUG ((LM_DEBUG,
 567                               "note: Buffered request took %d ms\n",
 568                               ms));
 569                 }
 570
 571               ms = num_requests_sent * request_interval;
 572             }
 573
 574           if (ms - connect_timeout > TIME_THRESHOLD ||
 575               connect_timeout - ms > TIME_THRESHOLD)
 576             {
 577               ACE_DEBUG ((LM_DEBUG,
 578                           "note: Timeout expected in %d ms, "
 579                           "but took %d ms\n", connect_timeout, ms));
 580             }
 581
 582           return 0;
 583         }
 584       else
 585         {
 586           ACE_ERROR_RETURN ((LM_ERROR, "Error: Unexpected\n"), 1);
 587         }
 588
 589     }
 590   catch (Exception &ex)
 591     {
 592       ACE_ERROR ((LM_ERROR, "client: %s\n\nLast operation took %d ms.\n",
 593                   ex._name(),
 594                   (ACE_High_Res_Timer::gettimeofday_hr () - before).msec ()));
 595     }
 596
 597   return 1;
 598 }