TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Supplier.cpp

   1 #include "ace/Get_Opt.h"
   2 #include <memory>
   3 #include "ace/Sched_Params.h"
   4 #include "ace/ACE.h"
   5 #include "ace/OS_NS_unistd.h"
   6
   7 #include "tao/Timeprobe.h"
   8 #include "tao/CDR.h"
   9 #include "orbsvcs/Event_Utilities.h"
  10 #include "orbsvcs/Event_Service_Constants.h"
  11 #include "orbsvcs/Time_Utilities.h"
  12 #include "orbsvcs/CosNamingC.h"
  13 #include "ECM_Supplier.h"
  14 #include "ECM_Data.h"
  15 #include "ace/OS_NS_errno.h"
  16
  17
  18 ECMS_Driver::ECMS_Driver ()
  19   : n_suppliers_ (1),
  20     event_count_ (100),
  21     event_period_ (100),
  22     event_size_ (32),
  23     event_a_ (ACE_ES_EVENT_UNDEFINED),
  24     event_b_ (ACE_ES_EVENT_UNDEFINED + 1),
  25     pid_file_name_ (0)
  26 {
  27 }
  28
  29 // ****************************************************************
  30
  31 int
  32 ECMS_Driver::run (int argc, ACE_TCHAR* argv[])
  33 {
  34   try
  35     {
  36       CORBA::ORB_var orb =
  37         CORBA::ORB_init (argc, argv);
  38
  39       CORBA::Object_var poa_object =
  40         orb->resolve_initial_references("RootPOA");
  41       if (CORBA::is_nil (poa_object.in ()))
  42         ACE_ERROR_RETURN ((LM_ERROR,
  43                            " (%P|%t) Unable to initialize the POA.\n"),
  44                           1);
  45
  46       PortableServer::POA_var root_poa =
  47         PortableServer::POA::_narrow (poa_object.in ());
  48
  49       PortableServer::POAManager_var poa_manager =
  50         root_poa->the_POAManager ();
  51
  52       if (this->parse_args (argc, argv))
  53         return 1;
  54
  55       ACE_DEBUG ((LM_DEBUG,
  56                   "Execution parameters:\n"
  57                   "  suppliers = <%d>\n"
  58                   "  event count = <%d>\n"
  59                   "  event period = <%d>\n"
  60                   "  event size = <%d>\n"
  61                   "  supplier Event A = <%d>\n"
  62                   "  supplier Event B = <%d>\n"
  63                   "  pid file name = <%s>\n",
  64
  65                   this->n_suppliers_,
  66                   this->event_count_,
  67                   this->event_period_,
  68                   this->event_size_,
  69                   this->event_a_,
  70                   this->event_b_,
  71
  72                   this->pid_file_name_?this->pid_file_name_:ACE_TEXT("nil")));
  73
  74       if (this->pid_file_name_ != 0)
  75         {
  76           FILE* pid = ACE_OS::fopen (this->pid_file_name_, "w");
  77           if (pid != 0)
  78             {
  79               ACE_OS::fprintf (pid, "%ld\n",
  80                                static_cast<long> (ACE_OS::getpid ()));
  81               ACE_OS::fclose (pid);
  82             }
  83         }
  84
  85       int min_priority =
  86         ACE_Sched_Params::priority_min (ACE_SCHED_FIFO);
  87       // Enable FIFO scheduling
  88
  89       if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
  90                                                   min_priority,
  91                                                   ACE_SCOPE_PROCESS)) != 0)
  92         {
  93           if (ACE_OS::last_error () == EPERM)
  94             ACE_DEBUG ((LM_DEBUG,
  95                         "%s: user is not superuser, "
  96                         "so remain in time-sharing class\n", argv[0]));
  97           else
  98             ACE_ERROR ((LM_ERROR,
  99                         "%s: ACE_OS::sched_params failed\n", argv[0]));
 100         }
 101
 102       if (ACE_OS::thr_setprio (min_priority) == -1)
 103         {
 104           ACE_ERROR ((LM_ERROR, "(%P|%t) main thr_setprio failed,"
 105                       "no real-time features\n"));
 106         }
 107
 108       CORBA::Object_var naming_obj =
 109         orb->resolve_initial_references ("NameService");
 110       if (CORBA::is_nil (naming_obj.in ()))
 111         ACE_ERROR_RETURN ((LM_ERROR,
 112                            " (%P|%t) Unable to get the Naming Service.\n"),
 113                           1);
 114
 115       CosNaming::NamingContext_var naming_context =
 116         CosNaming::NamingContext::_narrow (naming_obj.in ());
 117
 118       CosNaming::Name name (1);
 119       name.length (1);
 120       name[0].id = CORBA::string_dup ("EventService");
 121
 122       CORBA::Object_var ec_obj =
 123         naming_context->resolve (name);
 124
 125       RtecEventChannelAdmin::EventChannel_var channel;
 126       if (CORBA::is_nil (ec_obj.in ()))
 127         channel = RtecEventChannelAdmin::EventChannel::_nil ();
 128       else
 129         channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in ());
 130
 131       poa_manager->activate ();
 132
 133       this->connect_suppliers (channel.in ());
 134
 135       ACE_DEBUG ((LM_DEBUG, "connected supplier(s)\n"));
 136
 137       this->activate_suppliers ();
 138
 139       ACE_DEBUG ((LM_DEBUG, "suppliers are active\n"));
 140
 141       // Wait for the supplier threads...
 142       if (ACE_Thread_Manager::instance ()->wait () == -1)
 143         {
 144           ACE_ERROR ((LM_ERROR, "Thread_Manager wait failed\n"));
 145           return 1;
 146         }
 147
 148       ACE_DEBUG ((LM_DEBUG, "suppliers finished\n"));
 149
 150       this->disconnect_suppliers ();
 151
 152       // @@ Deactivate the suppliers (as CORBA Objects?)
 153     }
 154   catch (const CORBA::SystemException& sys_ex)
 155     {
 156       sys_ex._tao_print_exception ("SYS_EX in Supplier");
 157     }
 158   catch (const CORBA::Exception& ex)
 159     {
 160       ex._tao_print_exception ("NON SYS EX in Supplier");
 161     }
 162   return 0;
 163 }
 164
 165 int
 166 ECMS_Driver::supplier_task (Test_Supplier *supplier,
 167                             void* /* cookie */)
 168 {
 169   try
 170     {
 171       ACE_Time_Value tv (0, this->event_period_);
 172
 173       CORBA::ULong n = this->event_size_;
 174
 175       ECM_IDLData::Info info;
 176       info.mobile_name = CORBA::string_dup ("test");
 177       info.mobile_speed = 1;
 178       info.trajectory.length (n);
 179
 180       ECM_Data other;
 181       other.description = CORBA::string_dup ("some data");
 182
 183       for (CORBA::ULong j = 0; j < n; ++j)
 184         {
 185           info.trajectory[j].x = j;
 186           info.trajectory[j].y = j * j;
 187           other.inventory.bind (j, j + 1);
 188         }
 189
 190       ACE_DEBUG ((LM_DEBUG,
 191                   "The inventory contains (%d) elements\n",
 192                   other.inventory.current_size ()));
 193
 194       // We have to make it big enough so we get a contiguous block,
 195       // otherwise the octet sequence will not work correctly.
 196       // NOTE: we could pre-allocate enough memory in the CDR stream
 197       // but we want to show that chaining works!
 198       TAO_OutputCDR cdr;
 199
 200       CORBA::Boolean byte_order = TAO_ENCAP_BYTE_ORDER;
 201       cdr << CORBA::Any::from_boolean (byte_order);
 202
 203       // The typecode name standard, the encode method is not (in
 204       // general the CDR interface is not specified).
 205       if (!(cdr << info))
 206         throw CORBA::MARSHAL ();
 207
 208       // Here we marshall a non-IDL type.
 209       cdr << other;
 210
 211       if (!cdr.good_bit ())
 212         ACE_ERROR ((LM_ERROR, "Problem marshalling C++ data\n"));
 213
 214       const ACE_Message_Block* mb = cdr.begin ();
 215       // NOTE: total_length () return the length of the complete
 216       // chain.
 217       CORBA::ULong mblen = cdr.total_length ();
 218
 219       for (CORBA::Long i = 0; i < this->event_count_; ++i)
 220         {
 221           RtecEventComm::EventSet event (1);
 222           event.length (1);
 223           event[0].header.source = supplier->supplier_id ();
 224           event[0].header.ttl = 1;
 225
 226           ACE_hrtime_t t = ACE_OS::gethrtime ();
 227           ORBSVCS_Time::hrtime_to_TimeT (event[0].header.creation_time, t);
 228
 229           if (i == static_cast<CORBA::Long> (this->event_count_) - 1)
 230             event[0].header.type = ACE_ES_EVENT_SHUTDOWN;
 231           else if (i % 2 == 0)
 232             event[0].header.type = this->event_a_;
 233           else
 234             event[0].header.type = this->event_b_;
 235
 236           // We use replace to minimize the copies, this should result
 237           // in just one memory allocation;
 238 #if (TAO_NO_COPY_OCTET_SEQUENCES == 1)
 239           event[0].data.payload.replace (mblen, mb);
 240 #else
 241           // If the replace method is not available, we will need
 242           // to do the copy manually.  First, set the octet sequence length.
 243           event[0].data.payload.length (mblen);
 244
 245           // Now copy over each byte.
 246           char* base = mb->data_block ()->base ();
 247           for(CORBA::ULong i = 0; i < mblen; i++)
 248             {
 249               event[0].data.payload[i] = base[i];
 250             }
 251 #endif /* TAO_NO_COPY_OCTET_SEQUENCES == 1 */
 252
 253           supplier->consumer_proxy ()->push(event);
 254
 255           // ACE_DEBUG ((LM_DEBUG, "(%t) supplier push event\n"));
 256
 257           ACE_OS::sleep (tv);
 258         }
 259     }
 260   catch (const CORBA::SystemException& sys_ex)
 261     {
 262       sys_ex._tao_print_exception ("SYS_EX");
 263     }
 264   catch (const CORBA::Exception& ex)
 265     {
 266       ex._tao_print_exception ("NON SYS EX");
 267     }
 268   return 0;
 269 }
 270
 271 void
 272 ECMS_Driver::connect_suppliers (RtecEventChannelAdmin::EventChannel_ptr channel)
 273 {
 274   for (int i = 0; i < this->n_suppliers_; ++i)
 275     {
 276       char buf[BUFSIZ];
 277       ACE_OS::sprintf (buf, "supplier_%02d", i);
 278
 279       ACE_NEW (this->suppliers_[i], Test_Supplier (this));
 280
 281       this->suppliers_[i]->connect (buf,
 282                                     this->event_a_,
 283                                     this->event_b_,
 284                                     channel);
 285     }
 286 }
 287
 288 void
 289 ECMS_Driver::activate_suppliers ()
 290 {
 291   for (int i = 0; i < this->n_suppliers_; ++i)
 292     {
 293       this->suppliers_[i]->activate ();
 294     }
 295 }
 296
 297 void
 298 ECMS_Driver::disconnect_suppliers ()
 299 {
 300   for (int i = 0; i < this->n_suppliers_; ++i)
 301     {
 302       this->suppliers_[i]->disconnect ();
 303     }
 304 }
 305
 306 int
 307 ECMS_Driver::parse_args (int argc, ACE_TCHAR *argv [])
 308 {
 309   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("ds:n:t:h:p:b:"));
 310   int opt;
 311
 312   while ((opt = get_opt ()) != EOF)
 313     {
 314       switch (opt)
 315         {
 316         case 's':
 317           this->n_suppliers_ = ACE_OS::atoi (get_opt.opt_arg ());
 318           break;
 319
 320         case 'n':
 321           this->event_count_ = ACE_OS::atoi (get_opt.opt_arg ());
 322           break;
 323
 324         case 't':
 325           this->event_period_ = ACE_OS::atoi (get_opt.opt_arg ());
 326           break;
 327
 328         case 'b':
 329           this->event_size_ = ACE_OS::atoi (get_opt.opt_arg ());
 330           break;
 331
 332         case 'h':
 333           {
 334             char* aux;
 335                 char* arg = ACE_OS::strtok_r (ACE_TEXT_ALWAYS_CHAR(get_opt.opt_arg ()), ",", &aux);
 336
 337             this->event_a_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 338                 arg = ACE_OS::strtok_r (0, ",", &aux);
 339             this->event_b_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 340           }
 341           break;
 342
 343         case 'p':
 344           this->pid_file_name_ = get_opt.opt_arg ();
 345           break;
 346
 347         case '?':
 348         default:
 349           ACE_DEBUG ((LM_DEBUG,
 350                       "Usage: %s "
 351                       "[ORB options] "
 352                       "-s <nsuppliers> "
 353                       "-n <event count> "
 354                       "-t <event period (usecs)> "
 355                       "-h <eventa,eventb> "
 356                       "-p <pid file name> "
 357                       "\n",
 358                       argv[0]));
 359           return -1;
 360         }
 361     }
 362
 363   if (this->event_count_ <= 0)
 364     {
 365       ACE_DEBUG ((LM_DEBUG,
 366                   "%s: event count (%d) is out of range, "
 367                   "reset to default (%d)\n",
 368                   argv[0], this->event_count_,
 369                   100));
 370       this->event_count_ = 100;
 371     }
 372
 373   if (this->event_size_ < 0)
 374     {
 375       ACE_DEBUG ((LM_DEBUG,
 376                   "%s: event size (%d) is out of range, "
 377                   "reset to default (%d)\n",
 378                   argv[0], this->event_size_,
 379                   32));
 380       this->event_count_ = 32;
 381     }
 382
 383   if (this->n_suppliers_ <= 0)
 384     {
 385       this->n_suppliers_ = 1;
 386       ACE_ERROR_RETURN ((LM_ERROR,
 387                          "%s: number of suppliers out of range, "
 388                          "reset to default (%d)\n",
 389                          argv[0], 1), -1);
 390     }
 391
 392   return 0;
 393 }
 394
 395 Test_Supplier::Test_Supplier (ECMS_Driver *driver)
 396   :  driver_ (driver),
 397      supplier_ (this)
 398 {
 399 }
 400
 401 void
 402 Test_Supplier::connect (const char* name,
 403                         int event_a,
 404                         int event_b,
 405                         RtecEventChannelAdmin::EventChannel_ptr ec)
 406 {
 407   this->supplier_id_ = ACE::crc32 (name);
 408   ACE_DEBUG ((LM_DEBUG,
 409               "ID for <%s> is %04.4x\n",
 410               name,
 411               this->supplier_id_));
 412
 413   ACE_SupplierQOS_Factory qos;
 414   qos.insert (this->supplier_id_,
 415               event_a,
 416               0, 1);
 417   qos.insert (this->supplier_id_,
 418               event_b,
 419               0, 1);
 420   qos.insert (this->supplier_id_,
 421               ACE_ES_EVENT_SHUTDOWN,
 422               0, 1);
 423
 424   RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
 425     ec->for_suppliers ();
 426
 427   this->consumer_proxy_ =
 428     supplier_admin->obtain_push_consumer ();
 429
 430   RtecEventComm::PushSupplier_var objref =
 431     this->supplier_._this ();
 432
 433   this->consumer_proxy_->connect_push_supplier (objref.in (),
 434                                                 qos.get_SupplierQOS ());
 435 }
 436
 437 void
 438 Test_Supplier::disconnect ()
 439 {
 440   if (CORBA::is_nil (this->consumer_proxy_.in ()))
 441     return;
 442
 443   RtecEventChannelAdmin::ProxyPushConsumer_var proxy =
 444     this->consumer_proxy_._retn ();
 445
 446   try
 447     {
 448       proxy->disconnect_push_consumer ();
 449     }
 450   catch (const CORBA::OBJECT_NOT_EXIST&)
 451     {
 452       // Ignore, the EC can shutdown before we get a chance to
 453       // disconnect
 454     }
 455   catch (const CORBA::TRANSIENT&)
 456     {
 457       // Ignore, the EC can shutdown before we get a chance to
 458       // disconnect
 459     }
 460   catch (const CORBA::Exception&)
 461     {
 462       throw;
 463     }
 464 }
 465
 466 int
 467 Test_Supplier::svc ()
 468 {
 469   return this->driver_->supplier_task (this, this->cookie_);
 470 }
 471
 472 void
 473 Test_Supplier::disconnect_push_supplier ()
 474 {
 475   this->consumer_proxy_ =
 476     RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
 477 }
 478
 479 int Test_Supplier::supplier_id () const
 480 {
 481   return this->supplier_id_;
 482 }
 483
 484 RtecEventChannelAdmin::ProxyPushConsumer_ptr
 485 Test_Supplier::consumer_proxy ()
 486 {
 487   return this->consumer_proxy_.in ();
 488 }
 489
 490 int
 491 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 492 {
 493   ECMS_Driver driver;
 494   return driver.run (argc, argv);
 495 }