TAO/tests/Oneway_Send_Timeouts/Client.cpp

   1 #include "tao/Messaging/Messaging.h"
   2 #include "tao/AnyTypeCode/Any.h"
   3
   4 #include <memory>
   5 #include "ace/Get_Opt.h"
   6 #include "ace/Time_Value.h"
   7 #include "ace/OS_NS_unistd.h"
   8 #include "ace/OS_NS_strings.h"
   9
  10 #include "Client.h"
  11 #include "TestC.h"
  12
  13 Client::Client (int argc, ACE_TCHAR* argv[])
  14   : init_ (false), one_way_test_ (false)
  15 {
  16   if (this->init (argc, argv)) {
  17     init_ = true;
  18   }
  19 }
  20
  21 Client::~Client ()
  22 {
  23   if (!CORBA::is_nil (orb_.in())) {
  24     orb_->shutdown (true);
  25     orb_->destroy ();
  26     orb_ = CORBA::ORB::_nil();
  27   }
  28
  29   ACE_DEBUG ((LM_DEBUG, "(%P|%t) ~Client>\n"));
  30 }
  31
  32 bool
  33 Client::init (int argc, ACE_TCHAR* argv[])
  34 {
  35   try {
  36     orb_ = CORBA::ORB_init (argc, argv, "Client");
  37     if (CORBA::is_nil (orb_.in())) {
  38       ACE_ERROR ((LM_ERROR, "Client::init> ORB initialization failed.\n"));
  39       return false;
  40     }
  41
  42     if (!this->parse_args (argc, argv)) {
  43       return false;
  44     }
  45
  46     // Timeout
  47     TimeBase::TimeT timeout = 5 * 1000000; // 5 seconds
  48     CORBA::Any any_object;
  49     any_object <<= timeout;
  50
  51     CORBA::PolicyList policy_list (2);
  52     policy_list.length (2);
  53     policy_list[0] =
  54       orb_->create_policy (Messaging::RELATIVE_RT_TIMEOUT_POLICY_TYPE,
  55                            any_object);
  56
  57
  58     // Timeout with SYNC_SCOPE SYNC_WITH_TRANSPORT
  59     Messaging::SyncScope sync_with_transport = Messaging::SYNC_WITH_TRANSPORT;
  60     CORBA::Any sync_with_transport_any;
  61     sync_with_transport_any <<= sync_with_transport;
  62     policy_list[1] = orb_->create_policy
  63       (Messaging::SYNC_SCOPE_POLICY_TYPE, sync_with_transport_any);
  64     CORBA::Object_var obj = test_obj_->_set_policy_overrides
  65       (policy_list, CORBA::SET_OVERRIDE);
  66     test_obj_transport_timeout_ = Test::_narrow (obj.in ());
  67     policy_list[1]->destroy ();
  68
  69
  70     // Timeout with SYNC_SCOPE SYNC_NONE
  71     Messaging::SyncScope sync_none = Messaging::SYNC_NONE;
  72     CORBA::Any sync_none_any;
  73     sync_none_any <<= sync_none;
  74     policy_list[1] = orb_->create_policy
  75       (Messaging::SYNC_SCOPE_POLICY_TYPE, sync_none_any);
  76     // Apply the policy at the object level
  77     obj = test_obj_->_set_policy_overrides
  78       (policy_list, CORBA::SET_OVERRIDE);
  79     test_obj_none_timeout_ = Test::_narrow (obj.in ());
  80     policy_list[1]->destroy ();
  81
  82
  83     // Timeout with TAO specific SYNC_SCOPE SYNC_EAGER_BUFFERING
  84     //Messaging::SyncScope eager_buffering = TAO::SYNC_EAGER_BUFFERING;
  85     Messaging::SyncScope eager_buffering = Messaging::SYNC_NONE;
  86     CORBA::Any eager_any;
  87     eager_any <<= eager_buffering;
  88     policy_list[1] = orb_->create_policy
  89       (Messaging::SYNC_SCOPE_POLICY_TYPE, eager_any);
  90     obj = test_obj_->_set_policy_overrides
  91       (policy_list, CORBA::SET_OVERRIDE);
  92     test_obj_eager_timeout_ = Test::_narrow (obj.in ());
  93     policy_list[1]->destroy ();
  94
  95
  96     // Timeout with TAO specific SYNC_SCOPE SYNC_DELAYED_BUFFERING
  97     Messaging::SyncScope delayed_buffering = TAO::SYNC_DELAYED_BUFFERING;
  98     CORBA::Any delayed_any;
  99     delayed_any <<= delayed_buffering;
 100     policy_list[1] = orb_->create_policy
 101       (Messaging::SYNC_SCOPE_POLICY_TYPE, delayed_any);
 102     obj = test_obj_->_set_policy_overrides
 103       (policy_list, CORBA::SET_OVERRIDE);
 104     test_obj_delayed_timeout_ = Test::_narrow (obj.in ());
 105     policy_list[1]->destroy ();
 106
 107
 108     // Timeout with default SYNC_SCOPE SYNC_WITH_SERVER
 109     Messaging::SyncScope sync_with_server = Messaging::SYNC_WITH_SERVER;
 110     CORBA::Any sync_with_server_any;
 111     sync_with_server_any <<= sync_with_server;
 112     policy_list[1] = orb_->create_policy
 113       (Messaging::SYNC_SCOPE_POLICY_TYPE, sync_with_server_any);
 114     obj = test_obj_->_set_policy_overrides
 115       (policy_list, CORBA::SET_OVERRIDE);
 116     test_obj_server_timeout_ = Test::_narrow (obj.in ());
 117     policy_list[1]->destroy ();
 118
 119
 120     // Timeout with default SYNC_SCOPE (SYNC_WITH_TARGET)
 121     Messaging::SyncScope sync_with_target = Messaging::SYNC_WITH_TARGET;
 122     CORBA::Any sync_with_target_any;
 123     sync_with_target_any <<= sync_with_target;
 124     policy_list[1] = orb_->create_policy
 125       (Messaging::SYNC_SCOPE_POLICY_TYPE, sync_with_target_any);
 126     // Apply the policy at the object level
 127     obj = test_obj_->_set_policy_overrides
 128       (policy_list, CORBA::SET_OVERRIDE);
 129     test_obj_target_timeout_ = Test::_narrow (obj.in ());
 130
 131     policy_list[0]->destroy ();
 132     policy_list[1]->destroy ();
 133     policy_list.length(0);
 134   }
 135   catch( CORBA::Exception& ex) {
 136     ACE_ERROR ((LM_ERROR, "(%P|%t) Client::Init> Caught CORBA::Exception %s"
 137                 , ex._info().c_str()));
 138     return false;
 139   }
 140
 141   return true;
 142 }
 143
 144 bool
 145 Client::parse_args (int argc, ACE_TCHAR* argv[])
 146 {
 147   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("b:k:1::f:"), 0);
 148   int c;
 149   std::string test_ior;
 150   std::string back_ior;
 151   std::string flush_strategy ("lf");
 152
 153   while ((c = get_opts ()) != -1) {
 154     switch (c)
 155       {
 156       case '1':
 157         one_way_test_ = true;
 158         break;
 159       case 'k':
 160         test_ior = ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ());
 161         break;
 162       case 'b':
 163         back_ior = ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ());
 164         break;
 165       case 'f':
 166         flush_strategy = ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ());
 167         break;
 168       default:
 169         ACE_ERROR_RETURN ((LM_ERROR, "Invalid option \'-%c\'\n", c)
 170                           , false);
 171       }
 172   }
 173
 174   CORBA::Object_var obj = orb_->string_to_object (test_ior.c_str());
 175   test_obj_ = Test::_narrow (obj.in ());
 176
 177   obj = orb_->string_to_object (back_ior.c_str());
 178   management_ = Test::_narrow (obj.in ());
 179
 180   if (ACE_OS::strcasecmp ("lf", flush_strategy.c_str()) == 0) {
 181     flush_strategy_ = LF;
 182   }
 183   else if (ACE_OS::strcasecmp ("blocking", flush_strategy.c_str()) == 0) {
 184     flush_strategy_ = BLOCKING;
 185   }
 186   else if (ACE_OS::strcasecmp ("reactive", flush_strategy.c_str()) == 0) {
 187     flush_strategy_ = REACTIVE;
 188   }
 189
 190   return true;
 191 }
 192
 193 bool
 194 Client::run ()
 195 {
 196   bool status = true;
 197
 198   if (CORBA::is_nil (orb_.in())) {
 199       ACE_ERROR ((LM_ERROR, "Client::run> nil ORB found.\n"));
 200       return false;
 201     }
 202
 203   try {
 204     try {
 205       if (one_way_test_)
 206         {
 207           if (!this->test_oneway_timeout (true)) {
 208             ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: Client::run> "
 209                         "test_oneway_timeout failed.\n"));
 210             status = false;
 211           }
 212         }
 213
 214       // now run same test without the transport flooded.
 215       if (!this->test_oneway_timeout (false)) {
 216         ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: Client::run> "
 217                     "test_oneway_timeout 2 failed.\n"));
 218         status = false;
 219       }
 220
 221       test_obj_->shutdown ();
 222     }
 223     catch( CORBA::Exception& ex) {
 224       management_->unsleep (); // remote side could be asleep
 225       ACE_ERROR ((LM_ERROR, "(%P|%t) Client::run> Caught during test logic CORBA::Exception %s"
 226                   , ex._info().c_str()));
 227       status = false;
 228     }
 229   }
 230   catch( CORBA::Exception& ex) {
 231     ACE_ERROR ((LM_ERROR, "(%P|%t) Client::run> Caught during test shutdown CORBA::Exception %s"
 232                 , ex._info().c_str()));
 233     status = false;
 234   }
 235
 236   return status;
 237 }
 238
 239 bool
 240 Client::test_oneway_timeout (bool flood)
 241 {
 242   bool status = true;
 243
 244   std::unique_ptr<char[]> tmp (new char [6000000]);
 245   char* msg = tmp.get();
 246
 247   ACE_OS::memset (msg,'A',5999999);
 248   msg[5999999] = 0;
 249
 250   test_obj_->dummy_two_way (); // connection establishment
 251
 252   ACE_Time_Value tv (0);
 253   if (flood && !this->flood_connection(tv)) {
 254     ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> flooding failed.\n"));
 255   }
 256
 257   // Timeout with SYNC_SCOPE SYNC_NONE
 258   try {
 259     std::string scope_name ("SYNC_NONE");
 260     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 261     test_obj_none_timeout_->dummy_one_way (msg);
 262
 263     if (flood && flush_strategy_ == BLOCKING) {
 264       // block flushing gives a oneway SYNCH_WITH_TRANSPORT semantics
 265       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: A Timeout was expected for SYNC_NONE.\n"));
 266     }
 267     else {
 268       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_NONE\n"));
 269     }
 270   }
 271   catch (const CORBA::TIMEOUT&) {
 272     if (flood && flush_strategy_ == BLOCKING) {
 273       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected a timeout was received for SYNC_NONE.\n"));
 274     }
 275     else {
 276       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 277                   "timeout exception with synch scope SYNC_NONE.\n"));
 278       status = false;
 279     }
 280   }
 281
 282
 283   // Timeout with TAO specific SYNC_SCOPE SYNC_EAGER_BUFFERING
 284   try {
 285     std::string scope_name ("SYNC_EAGER_BUFFERING");
 286     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 287
 288     /* BLOCKed flushing has SYNCH_WITH_TRANSPORT semantics. With flooding turned on
 289         you would have received a TIMEOUT in the previous test (SYNC_NONE). Even without
 290         flooding there is a chance to get back a TIMEOUT. The TIMEOUT has the side-effect
 291         closing out the connection. Therefore when flush_strategy is set to BLOCK we want
 292         to re-establish connection before each test. With flooding turned on we need
 293         to first unsleep the test server, re-establish connection and put it back to sleep.
 294         With flooding=0 we simple re-establish connection. This trick is performed at beginning
 295         of every test for flush_strategy == BLOCKING.
 296      */
 297     if (flush_strategy_ == BLOCKING) {
 298       if (flood) {
 299         management_->unsleep ();
 300         test_obj_->sleep (0, 0); // rebuild connection and put server thread to sleep
 301       }
 302       else {
 303         // else simply re-establish connection
 304         test_obj_->dummy_two_way ();
 305       }
 306     }
 307
 308     test_obj_eager_timeout_->dummy_one_way (msg);
 309
 310     if (flood && flush_strategy_ == BLOCKING) {
 311       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: A Timeout was expected for SYNC_EAGER_BUFFERING\n"));
 312     }
 313     else {
 314       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_EAGER_BUFFERING\n"));
 315     }
 316   }
 317   catch (const CORBA::TIMEOUT&) {
 318     if (flood && flush_strategy_ == BLOCKING) {
 319       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected a timeout was received for SYNC_EAGER_BUFFERING\n"));
 320     }
 321     else {
 322       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 323                   "timeout exception with synch scope SYNC_EAGER_BUFFERING.\n"));
 324       status = false;
 325     }
 326   }
 327
 328   // Timeout with TAO specific SYNC_SCOPE SYNC_DELAYED_BUFFERING
 329   try {
 330     std::string scope_name ("SYNC_DELAYED_BUFFERING");
 331     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 332
 333     if (flush_strategy_ == BLOCKING) {
 334       if (flood) {
 335         management_->unsleep ();
 336         test_obj_->sleep (0, 0);
 337       }
 338       else {
 339         test_obj_->dummy_two_way ();
 340       }
 341     }
 342
 343     test_obj_delayed_timeout_->dummy_one_way (msg);
 344
 345     if (flood && flush_strategy_ == BLOCKING) {
 346       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: A Timeout was expected for SYNC_DELAYED_BUFFERING\n"));
 347     }
 348     else {
 349       test_obj_delayed_timeout_->dummy_one_way ("SYNC_DELAYED_BUFFERING");
 350       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_DELAYED_BUFFERING\n"));
 351     }
 352   }
 353   catch (const CORBA::TIMEOUT&) {
 354     if (flood && flush_strategy_ == BLOCKING) {
 355       ACE_DEBUG ((LM_DEBUG, "(%P|%t) Expected timeout received for SYNC_DELAYED_BUFFERING\n"));
 356     }
 357     else {
 358       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 359                 "timeout exception with synch scope SYNC_DELAYED_BUFFERING.\n"));
 360       status = false;
 361     }
 362   }
 363
 364   /* Cleanup queue before the synchronous tests. We don't want the test
 365       results affected by leftovers from previous runs.
 366    */
 367   ACE_Time_Value tv_tmp (1);
 368   orb_->run (tv_tmp);
 369
 370   // Timeout with SYNC_SCOPE SYNC_WITH_TRANSPORT
 371   try {
 372     std::string scope_name ("SYNC_WITH_TRANSPORT");
 373     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 374
 375     if (flush_strategy_ == BLOCKING) {
 376       if (flood) {
 377         management_->unsleep ();
 378         test_obj_->sleep (0, 0);
 379       }
 380       else {
 381         test_obj_->dummy_two_way ();
 382       }
 383     }
 384
 385     test_obj_transport_timeout_->dummy_one_way (msg);
 386
 387     if (flood) {
 388       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Expected "
 389                   "timeout not received for synch scope SYNC_WITH_TRANSPORT.\n"
 390                   ));
 391       status = false;
 392     }
 393     else {
 394       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_WITH_TRANSPORT\n"));
 395     }
 396   }
 397   catch (const CORBA::TIMEOUT&) {
 398     if (flood) {
 399       ACE_DEBUG ((LM_DEBUG, "(%P|%t) Expected Timeout received for SYNC_WITH_TRANSPORT\n"));
 400     }
 401     else {
 402       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 403                   "timeout exception with synch scope SYNC_WITH_TRANSPORT.\n"));
 404       status = false;
 405     }
 406   }
 407
 408 // Timeout with default SYNC_SCOPE SYNC_WITH_SERVER
 409   try {
 410     std::string scope_name ("SYNC_WITH_SERVER");
 411     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 412
 413     if (flush_strategy_ == BLOCKING) {
 414       if (flood) {
 415         management_->unsleep ();
 416         test_obj_->sleep (0, 0);
 417       }
 418       else {
 419         test_obj_->dummy_two_way ();
 420       }
 421     }
 422
 423     test_obj_server_timeout_->dummy_one_way (msg);
 424
 425     if (flood) {
 426       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Expected "
 427                   "timeout not received for SYNC_SCOPE SYNC_WITH_SERVER.\n"));
 428       status = false;
 429     }
 430     else {
 431       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_WITH_SERVER\n"));
 432     }
 433   }
 434   catch (const CORBA::TIMEOUT&) {
 435     if (flood) {
 436       ACE_DEBUG ((LM_DEBUG, "(%P|%t) Expected Timeout received for SYNC_WITH_SERVER\n"));
 437     }
 438     else {
 439       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 440                   "timeout exception with synch scope SYNC_WITH_SERVER.\n"));
 441       status = false;
 442     }
 443   }
 444
 445
 446   // Timeout with default SYNC_SCOPE (SYNC_WITH_TARGET)
 447   try {
 448     std::string scope_name ("SYNC_WITH_TARGET");
 449     ACE_OS::strncpy (msg, scope_name.c_str(), scope_name.length());
 450
 451     if (flush_strategy_ == BLOCKING) {
 452       if (flood) {
 453         management_->unsleep ();
 454         test_obj_->sleep (0, 0);
 455       }
 456       else {
 457         test_obj_->dummy_two_way ();
 458       }
 459     }
 460
 461     test_obj_target_timeout_->dummy_one_way (msg);
 462
 463     if (flood) {
 464       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Expected "
 465                   "timeout not received for SYNC_SCOPE SYNC_WITH_TARGET.\n"));
 466       status = false;
 467     }
 468     else {
 469       ACE_DEBUG ((LM_DEBUG, "(%P|%t) As expected no Timeout received for SYNC_WITH_TARGET\n"));
 470     }
 471   }
 472   catch (const CORBA::TIMEOUT&) {
 473     if (flood) {
 474       ACE_DEBUG ((LM_DEBUG, "(%P|%t) Expected Timeout received for SYNC_WITH_TARGET\n"));
 475     }
 476     else {
 477       ACE_ERROR ((LM_ERROR, "(%P|%t) ERROR: test_oneway_timeout> Unexpected "
 478                   "timeout exception with synch scope SYNC_WITH_TARGET.\n"));
 479       status = false;
 480     }
 481   }
 482
 483   if (flood) {
 484     management_->unsleep ();
 485   }
 486
 487   return status;
 488 }
 489
 490 bool
 491 Client::flood_connection (ACE_Time_Value& tv)
 492 {
 493   // Block flushing currently blocks even on SYNC_DELAYED_BUFFERING
 494   //  so we can't use it to flood connections.
 495
 496   // Set the policy value.
 497   // SYNC_DELAYED_BUFFERING is used to ensure that the tcp buffer gets filled before
 498   //  buffering starts.
 499   Messaging::SyncScope sync_scope = TAO::SYNC_DELAYED_BUFFERING;
 500   //Messaging::SyncScope sync_scope = Messaging::SYNC_NONE;
 501   CORBA::Any sync_scope_any;
 502   sync_scope_any <<= sync_scope;
 503
 504   CORBA::PolicyList policy_list (1);
 505   policy_list.length (1);
 506   policy_list[0] = orb_->create_policy
 507     (Messaging::SYNC_SCOPE_POLICY_TYPE, sync_scope_any);
 508   // Apply the policy at the object level
 509   CORBA::Object_var obj = test_obj_->_set_policy_overrides
 510     (policy_list, CORBA::SET_OVERRIDE);
 511   Test_var mod_test_obj = Test::_narrow (obj.in ());
 512
 513   policy_list[0]->destroy ();
 514   policy_list.length(0);
 515
 516   std::unique_ptr<char[]> tmp (new char [2000000]);
 517   char* msg = tmp.get();
 518
 519   ACE_OS::memset (msg,'A',1999999);
 520   msg[1999999] = 0;
 521
 522   test_obj_->sleep (static_cast<CORBA::Long>(tv.sec())
 523                     , static_cast<CORBA::Long>(tv.msec()));
 524
 525   /* BLOCK flush startegy always has SYNC_WITH_TRANSPORT semantics.
 526       Trying to flood a BLOCKed flushing connection can lead to a TIMEOUT
 527       exception being thrown. This will close out the connection and
 528       the whole flooding attempt fails. Therefore in BLOCK flushing case
 529       don't attempt to flood (unless BLOCK flush accepts SYNC_WITH_TRANSPORT
 530       semantics).
 531    */
 532   if (flush_strategy_ != BLOCKING)
 533     {
 534       mod_test_obj->dummy_one_way (msg);
 535
 536       // attempt again to flood connection.
 537       ACE_Time_Value tv_tmp (2);
 538       orb_->perform_work (tv_tmp);
 539     }
 540
 541   return true;
 542 }