TAO/tests/Oneway_Send_Timeouts/Client.cpp

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