TAO/performance-tests/Protocols/distributor.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "ace/OS_NS_stdio.h"
   3 #include "ace/OS_NS_string.h"
   4 #include "ace/OS_NS_unistd.h"
   5 #include "tao/RTCORBA/RTCORBA.h"
   6 #include "tao/ORB_Constants.h"
   7 #include "tao/Policy_CurrentC.h"
   8 #include "tao/debug.h"
   9 #include "testS.h"
  10
  11 static const ACE_TCHAR *ior_file = ACE_TEXT ("distributor.ior");
  12 static const ACE_TCHAR *ior = ACE_TEXT ("file://receiver.ior");
  13 static int number_of_connection_attempts = 20;
  14
  15 static int
  16 parse_args (int argc, ACE_TCHAR **argv)
  17 {
  18   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("f:k:"));
  19   int c;
  20
  21   while ((c = get_opts ()) != -1)
  22     switch (c)
  23       {
  24       case 'k':
  25         ior = get_opts.opt_arg ();
  26         break;
  27
  28       case 'f':
  29         ior_file = get_opts.opt_arg ();
  30         break;
  31
  32       default:
  33         ACE_ERROR_RETURN ((LM_ERROR,
  34                            "usage: %s\n"
  35                            "\t-f <ior_file> (defaults to %s)\n"
  36                            "\t-k <ior> (defaults to %s)\n"
  37                            "\n",
  38                            argv[0],
  39                            ior_file,
  40                            ior),
  41                           -1);
  42       }
  43
  44   return 0;
  45 }
  46
  47 class test_i :
  48   public POA_test
  49 {
  50 public:
  51   test_i (CORBA::ORB_ptr orb,
  52           PortableServer::POA_ptr poa,
  53           RTCORBA::RTORB_ptr rtorb,
  54           CORBA::PolicyManager_ptr policy_manager,
  55           test_ptr receiver);
  56
  57   ~test_i ();
  58
  59   void start_test (CORBA::Long session_id,
  60                    const char *protocol,
  61                    CORBA::ULong invocation_rate,
  62                    CORBA::ULong message_size,
  63                    CORBA::ULong iterations);
  64
  65   void end_test ();
  66
  67   void oneway_sync ();
  68
  69   void twoway_sync ();
  70
  71   void oneway_method (CORBA::Long session_id,
  72                       CORBA::ULong iteration,
  73                       const ::test::octets &payload);
  74
  75   void twoway_method (CORBA::Long &session_id,
  76                       CORBA::ULong &iteration,
  77                       ::test::octets &payload);
  78
  79   //FUZZ: disable check_for_lack_ACE_OS
  80   void shutdown ();
  81   //FUZZ: enable check_for_lack_ACE_OS
  82
  83   PortableServer::POA_ptr _default_POA ();
  84
  85 private:
  86   CORBA::ORB_var orb_;
  87   PortableServer::POA_var poa_;
  88   RTCORBA::RTORB_var rtorb_;
  89   CORBA::PolicyManager_var policy_manager_;
  90   test_var receiver_;
  91
  92   CORBA::PolicyList base_protocol_policy_;
  93   CORBA::PolicyList test_protocol_policy_;
  94 };
  95
  96 test_i::test_i (CORBA::ORB_ptr orb,
  97                 PortableServer::POA_ptr poa,
  98                 RTCORBA::RTORB_ptr rtorb,
  99                 CORBA::PolicyManager_ptr policy_manager,
 100                 test_ptr receiver)
 101   : orb_ (CORBA::ORB::_duplicate (orb)),
 102     poa_ (PortableServer::POA::_duplicate (poa)),
 103     rtorb_ (RTCORBA::RTORB::_duplicate (rtorb)),
 104     policy_manager_ (CORBA::PolicyManager::_duplicate (policy_manager)),
 105     receiver_ (test::_duplicate (receiver))
 106 {
 107   // Base protocol is used for setting up and tearing down the test.
 108   this->base_protocol_policy_.length (1);
 109
 110   // Test protocol is the one being tested.
 111   this->test_protocol_policy_.length (1);
 112
 113   RTCORBA::ProtocolList protocols;
 114   protocols.length (1);
 115   protocols[0].transport_protocol_properties =
 116     RTCORBA::ProtocolProperties::_nil ();
 117   protocols[0].orb_protocol_properties =
 118     RTCORBA::ProtocolProperties::_nil ();
 119
 120   // IIOP is always used for the base protocol.
 121   protocols[0].protocol_type = 0;
 122
 123   // User decides the test protocol.
 124   this->base_protocol_policy_[0] =
 125     this->rtorb_->create_client_protocol_policy (protocols);
 126 }
 127
 128 test_i::~test_i ()
 129 {
 130 }
 131
 132 void
 133 test_i::start_test (CORBA::Long session_id,
 134                     const char *protocol,
 135                     CORBA::ULong invocation_rate,
 136                     CORBA::ULong message_size,
 137                     CORBA::ULong iterations)
 138 {
 139   RTCORBA::ProtocolList protocols;
 140   protocols.length (1);
 141   protocols[0].transport_protocol_properties =
 142     RTCORBA::ProtocolProperties::_nil ();
 143   protocols[0].orb_protocol_properties =
 144     RTCORBA::ProtocolProperties::_nil ();
 145
 146   if (ACE_OS::strcmp (protocol, "DIOP") == 0)
 147     {
 148       if (TAO_debug_level) ACE_DEBUG ((LM_DEBUG, "test protocol is DIOP\n"));
 149       protocols[0].protocol_type = TAO_TAG_DIOP_PROFILE;
 150     }
 151   else if (ACE_OS::strcmp (protocol, "SCIOP") == 0)
 152     {
 153       if (TAO_debug_level) ACE_DEBUG ((LM_DEBUG, "test protocol is SCIOP\n"));
 154       protocols[0].protocol_type = TAO_TAG_SCIOP_PROFILE;
 155     }
 156   else
 157     {
 158       if (TAO_debug_level) ACE_DEBUG ((LM_DEBUG, "test protocol is IIOP\n"));
 159       protocols[0].protocol_type = 0;
 160     }
 161
 162   this->test_protocol_policy_[0] =
 163     this->rtorb_->create_client_protocol_policy (protocols);
 164
 165   // Make sure we have a connection to the server using the test
 166   // protocol.
 167   this->policy_manager_->set_policy_overrides (this->test_protocol_policy_,
 168                                                CORBA::SET_OVERRIDE);
 169
 170   // Since the network maybe unavailable temporarily, make sure to try
 171   // for a few times before giving up.
 172   for (int j = 0;;)
 173     {
 174     test_protocol_setup:
 175
 176       try
 177         {
 178           // Send a message to ensure that the connection is setup.
 179           this->receiver_->oneway_sync ();
 180
 181           goto test_protocol_success;
 182         }
 183       catch (const CORBA::TRANSIENT&)
 184         {
 185           ++j;
 186
 187           if (j < number_of_connection_attempts)
 188             {
 189               ACE_OS::sleep (1);
 190               goto test_protocol_setup;
 191             }
 192         }
 193
 194       ACE_ERROR ((LM_ERROR,
 195                   "Cannot setup test protocol\n"));
 196
 197       throw CORBA::TRANSIENT (CORBA::OMGVMCID | 2, CORBA::COMPLETED_NO);
 198     }
 199
 200  test_protocol_success:
 201
 202   // Use IIOP for setting up the test since the test protocol maybe
 203   // unreliable.
 204   this->policy_manager_->set_policy_overrides (this->base_protocol_policy_,
 205                                                CORBA::SET_OVERRIDE);
 206
 207   // Since the network maybe unavailable temporarily, make sure to try
 208   // for a few times before giving up.
 209   for (int k = 0;;)
 210     {
 211     base_protocol_setup:
 212
 213       try
 214         {
 215           // Let the server know what to expect..
 216           this->receiver_->start_test (session_id,
 217                                        protocol,
 218                                        invocation_rate,
 219                                        message_size,
 220                                        iterations);
 221
 222           goto base_protocol_success;
 223         }
 224       catch (const CORBA::TRANSIENT&)
 225         {
 226           ACE_OS::sleep (1);
 227
 228           if (k < number_of_connection_attempts)
 229             {
 230               ACE_OS::sleep (1);
 231               goto base_protocol_setup;
 232             }
 233         }
 234
 235       ACE_ERROR ((LM_ERROR,
 236                   "Cannot setup base protocol\n"));
 237
 238       throw CORBA::TRANSIENT (CORBA::OMGVMCID | 2, CORBA::COMPLETED_NO);
 239     }
 240
 241  base_protocol_success:
 242
 243   // Select the test protocol for these invocation.
 244   this->policy_manager_->set_policy_overrides (this->test_protocol_policy_,
 245                                                CORBA::SET_OVERRIDE);
 246 }
 247
 248 void
 249 test_i::end_test ()
 250 {
 251   // Use IIOP to indicate end of test to server.
 252   this->policy_manager_->set_policy_overrides (this->base_protocol_policy_,
 253                                                CORBA::SET_OVERRIDE);
 254
 255   this->receiver_->end_test ();
 256 }
 257
 258 void
 259 test_i::oneway_sync ()
 260 {
 261   this->receiver_->oneway_sync ();
 262 }
 263
 264 void
 265 test_i::twoway_sync ()
 266 {
 267   this->receiver_->twoway_sync ();
 268 }
 269
 270 void
 271 test_i::oneway_method (CORBA::Long session_id,
 272                        CORBA::ULong iteration,
 273                        const ::test::octets &payload)
 274 {
 275   this->receiver_->oneway_method (session_id,
 276                                   iteration,
 277                                   payload);
 278 }
 279
 280 void
 281 test_i::twoway_method (CORBA::Long &session_id,
 282                        CORBA::ULong &iteration,
 283                        ::test::octets &payload)
 284 {
 285   this->receiver_->twoway_method (session_id,
 286                                   iteration,
 287                                   payload);
 288 }
 289
 290 PortableServer::POA_ptr
 291 test_i::_default_POA ()
 292 {
 293   return PortableServer::POA::_duplicate (this->poa_.in ());
 294 }
 295
 296 void
 297 test_i::shutdown ()
 298 {
 299   this->receiver_->shutdown ();
 300
 301   this->orb_->shutdown (false);
 302 }
 303
 304 int
 305 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 306 {
 307   try
 308     {
 309       CORBA::ORB_var orb =
 310         CORBA::ORB_init (argc, argv);
 311
 312       CORBA::Object_var object =
 313         orb->resolve_initial_references ("RTORB");
 314
 315       RTCORBA::RTORB_var rtorb =
 316         RTCORBA::RTORB::_narrow (object.in ());
 317
 318       object =
 319         orb->resolve_initial_references ("ORBPolicyManager");
 320
 321       CORBA::PolicyManager_var policy_manager =
 322         CORBA::PolicyManager::_narrow (object.in ());
 323
 324       int parse_args_result =
 325         parse_args (argc, argv);
 326       if (parse_args_result != 0)
 327         return parse_args_result;
 328
 329       object =
 330         orb->resolve_initial_references ("RootPOA");
 331
 332       PortableServer::POA_var root_poa =
 333         PortableServer::POA::_narrow (object.in ());
 334
 335       PortableServer::POAManager_var poa_manager =
 336         root_poa->the_POAManager ();
 337
 338       object =
 339         orb->string_to_object (ior);
 340
 341       test_var receiver =
 342         test::_narrow (object.in ());
 343
 344       test_i *servant =
 345         new test_i (orb.in (),
 346                     root_poa.in (),
 347                     rtorb.in (),
 348                     policy_manager.in (),
 349                     receiver.in ());
 350       PortableServer::ServantBase_var safe_servant (servant);
 351       ACE_UNUSED_ARG (safe_servant);
 352
 353       test_var test =
 354         servant->_this ();
 355
 356       CORBA::String_var ior =
 357         orb->object_to_string (test.in ());
 358
 359       FILE *output_file =
 360         ACE_OS::fopen (ior_file, "w");
 361       ACE_ASSERT (output_file != 0);
 362
 363       u_int result =
 364         ACE_OS::fprintf (output_file,
 365                          "%s",
 366                          ior.in ());
 367       ACE_ASSERT (result == ACE_OS::strlen (ior.in ()));
 368       ACE_UNUSED_ARG (result);
 369
 370       ACE_OS::fclose (output_file);
 371
 372       poa_manager->activate ();
 373
 374       orb->run ();
 375       ACE_OS::sleep(1);
 376     }
 377   catch (const CORBA::Exception& ex)
 378     {
 379       ex._tao_print_exception ("Exception caught");
 380       return -1;
 381     }
 382
 383   return 0;
 384 }