TAO/orbsvcs/examples/RtEC/Kokyu/Service.cpp

   1 #include "orbsvcs/Sched/Reconfig_Scheduler.h"
   2 #include "orbsvcs/Runtime_Scheduler.h"
   3 //#include "orbsvcs/Event/Module_Factory.h"
   4 //#include "orbsvcs/Event/Event_Channel.h"
   5 #include "orbsvcs/Event_Service_Constants.h"
   6 #include "orbsvcs/Event_Utilities.h"
   7 #include "orbsvcs/Scheduler_Factory.h"
   8 #include "orbsvcs/Event/EC_Event_Channel.h"
   9 #include "orbsvcs/Event/EC_Default_Factory.h"
  10 #include "orbsvcs/Event/EC_Kokyu_Factory.h"
  11 #include "Consumer.h"
  12 #include "Supplier.h"
  13
  14
  15 #include "ace/Get_Opt.h"
  16 #include "ace/Sched_Params.h"
  17 #include <memory>
  18 #include "ace/SString.h"
  19 #include "ace/OS_NS_strings.h"
  20 #include "ace/Thread.h"
  21
  22
  23 namespace
  24 {
  25   int config_run = 0;
  26   ACE_CString sched_type ="rms";
  27 }
  28
  29 inline RtecScheduler::Period_t time_val_to_period (const ACE_Time_Value &tv)
  30 {
  31   //100s of nanoseconds
  32   return static_cast<RtecScheduler::Period_t> (tv.sec () * 1000000 + tv.usec ())*10;
  33 }
  34
  35 int parse_args (int argc, ACE_TCHAR *argv[]);
  36
  37 typedef TAO_Reconfig_Scheduler<TAO_RMS_FAIR_Reconfig_Sched_Strategy, TAO_SYNCH_MUTEX> RECONFIG_RMS_SCHED_TYPE;
  38
  39 typedef TAO_Reconfig_Scheduler<TAO_MUF_FAIR_Reconfig_Sched_Strategy, TAO_SYNCH_MUTEX> RECONFIG_MUF_SCHED_TYPE;
  40
  41 int
  42 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  43 {
  44   ACE_UNUSED_ARG (config_run);
  45
  46   //TAO_EC_Default_Factory::init_svcs ();
  47
  48   TAO_EC_Kokyu_Factory::init_svcs ();
  49
  50
  51   try
  52     {
  53       // ORB initialization boiler plate...
  54       CORBA::ORB_var orb =
  55         CORBA::ORB_init (argc, argv);
  56
  57       if (parse_args (argc, argv) == -1)
  58         {
  59           ACE_ERROR ((LM_ERROR,
  60                       "Usage: Service [-o IOR_file_name]\n"));
  61           return 1;
  62         }
  63
  64       CORBA::Object_var object =
  65         orb->resolve_initial_references ("RootPOA");
  66       PortableServer::POA_var poa =
  67         PortableServer::POA::_narrow (object.in ());
  68       PortableServer::POAManager_var poa_manager =
  69         poa->the_POAManager ();
  70       poa_manager->activate ();
  71
  72       // ****************************************************************
  73
  74       // Create an scheduling service
  75       POA_RtecScheduler::Scheduler* sched_impl = 0;
  76
  77       if (ACE_OS::strcasecmp(sched_type.c_str(), "rms") == 0)
  78         {
  79           ACE_DEBUG ((LM_DEBUG, "Creating RMS scheduler\n"));
  80           ACE_NEW_RETURN (sched_impl,
  81                           RECONFIG_RMS_SCHED_TYPE,
  82                           1);
  83         }
  84       else if (ACE_OS::strcasecmp(sched_type.c_str(), "muf") == 0)
  85         {
  86           ACE_DEBUG ((LM_DEBUG, "Creating MUF scheduler\n"));
  87           ACE_NEW_RETURN (sched_impl,
  88                           RECONFIG_MUF_SCHED_TYPE,
  89                           1);
  90         }
  91
  92       RtecScheduler::Scheduler_var scheduler =
  93         sched_impl->_this ();
  94
  95       // ****************************************************************
  96       TAO_EC_Event_Channel_Attributes attributes (poa.in (),
  97                                                   poa.in ());
  98       attributes.scheduler = scheduler.in (); // no need to dup
  99
 100       TAO_EC_Event_Channel ec_impl (attributes);
 101       RtecEventChannelAdmin::EventChannel_var event_channel =
 102         ec_impl._this ();
 103       // ****************************************************************
 104
 105       // Create a consumer, intialize its RT_Info structures, and
 106       // connnect to the event channel....
 107
 108       Consumer consumer_impl1, consumer_impl2;
 109
 110       RtecScheduler::handle_t consumer1_rt_info =
 111         scheduler->create ("consumer1");
 112
 113       RtecScheduler::handle_t consumer2_rt_info =
 114         scheduler->create ("consumer2");
 115
 116       //consumer's rate will get propagated from the supplier.
 117       //so no need to specify a period here. Specifying
 118       //criticality is crucial since it propagates from
 119       //consumer to supplier.
 120       ACE_Time_Value tv (0,0);
 121       TimeBase::TimeT tmp;
 122       ORBSVCS_Time::Time_Value_to_TimeT (tmp, tv);
 123       scheduler->set (consumer1_rt_info,
 124                       RtecScheduler::VERY_LOW_CRITICALITY,
 125                       tmp, tmp, tmp,
 126                       time_val_to_period (tv),
 127                       RtecScheduler::VERY_LOW_IMPORTANCE,
 128                       tmp,
 129                       0,
 130                       RtecScheduler::OPERATION);
 131
 132       scheduler->set (consumer2_rt_info,
 133                       RtecScheduler::VERY_HIGH_CRITICALITY,
 134                       tmp, tmp, tmp,
 135                       time_val_to_period (tv),
 136                       RtecScheduler::VERY_HIGH_IMPORTANCE,
 137                       tmp,
 138                       0,
 139                       RtecScheduler::OPERATION);
 140
 141       ACE_ConsumerQOS_Factory consumer_qos1, consumer_qos2;
 142       //consumer_qos.start_disjunction_group ();
 143       // The types int the range [0,ACE_ES_EVENT_UNDEFINED) are
 144       // reserved for the EC...
 145       consumer_qos1.insert_type (ACE_ES_EVENT_UNDEFINED,
 146                                 consumer1_rt_info);
 147
 148       RtecEventChannelAdmin::ConsumerQOS qos =
 149         consumer_qos1.get_ConsumerQOS ();
 150 /*
 151       for (int i=0;i<qos.dependencies.length (); ++i)
 152         {
 153           ACE_DEBUG ((LM_DEBUG,
 154                       "consumer_qos1[%d] event.header.type = %d, "
 155                       "consumer_qos1[%d] rt_info = %d, "
 156                       "consumer_qos1[%d] event.header.source = %d\n",
 157                       i,qos.dependencies[i].event.header.type,
 158                       i,qos.dependencies[i].rt_info,
 159                       i,qos.dependencies[i].event.header.source));
 160         }
 161 */
 162
 163       consumer_qos2.insert_type (ACE_ES_EVENT_UNDEFINED + 1,
 164                                 consumer2_rt_info);
 165
 166       // The canonical protocol to connect to the EC
 167       RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
 168         event_channel->for_consumers ();
 169
 170       RtecEventChannelAdmin::ProxyPushSupplier_var supplier_proxy1 =
 171         consumer_admin->obtain_push_supplier ();
 172
 173       RtecEventChannelAdmin::ProxyPushSupplier_var supplier_proxy2 =
 174         consumer_admin->obtain_push_supplier ();
 175
 176       RtecEventComm::PushConsumer_var consumer1 =
 177         consumer_impl1._this ();
 178
 179       RtecEventComm::PushConsumer_var consumer2 =
 180         consumer_impl2._this ();
 181
 182       ACE_DEBUG ((LM_DEBUG, "connecting consumers\n"));
 183       ACE_DEBUG ((LM_DEBUG, "connecting consumer1\n"));
 184       supplier_proxy1->connect_push_consumer (consumer1.in (),
 185                                              consumer_qos1.get_ConsumerQOS ());
 186
 187       ACE_DEBUG ((LM_DEBUG, "connecting consumer2\n"));
 188       supplier_proxy2->connect_push_consumer (consumer2.in (),
 189                                              consumer_qos2.get_ConsumerQOS ());
 190
 191       ACE_DEBUG ((LM_DEBUG, "consumers connected\n"));
 192
 193       // ****************************************************************
 194
 195       RtecScheduler::handle_t supplier1_rt_info =
 196         scheduler->create ("supplier1");
 197
 198       RtecScheduler::handle_t supplier2_rt_info =
 199         scheduler->create ("supplier2");
 200
 201       RtecEventComm::EventSourceID supplier_id1 = 1, supplier_id2 = 2;
 202       ACE_SupplierQOS_Factory supplier_qos1, supplier_qos2;
 203       supplier_qos1.insert (supplier_id1,
 204                            ACE_ES_EVENT_UNDEFINED,
 205                            supplier1_rt_info,
 206                            1 /* number of calls, but what does that mean? */);
 207       supplier_qos2.insert (supplier_id2,
 208                            ACE_ES_EVENT_UNDEFINED + 1,
 209                            supplier2_rt_info,
 210                            1 /* number of calls, but what does that mean? */);
 211
 212       // The canonical protocol to connect to the EC
 213       RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
 214         event_channel->for_suppliers ();
 215
 216       RtecEventChannelAdmin::ProxyPushConsumer_var consumer_proxy1 =
 217         supplier_admin->obtain_push_consumer ();
 218
 219       RtecEventChannelAdmin::ProxyPushConsumer_var consumer_proxy2 =
 220         supplier_admin->obtain_push_consumer ();
 221
 222       Supplier supplier_impl1(supplier_id1, consumer_proxy1.in ());
 223       Supplier supplier_impl2(supplier_id2, consumer_proxy2.in ());
 224
 225       RtecEventComm::PushSupplier_var supplier1 =
 226         supplier_impl1._this ();
 227
 228       RtecEventComm::PushSupplier_var supplier2 =
 229         supplier_impl2._this ();
 230
 231       ACE_DEBUG ((LM_DEBUG, "connecting suppliers\n"));
 232       ACE_DEBUG ((LM_DEBUG, "connecting supplier1\n"));
 233       consumer_proxy1->connect_push_supplier (supplier1.in (),
 234                                              supplier_qos1.get_SupplierQOS ());
 235
 236       ACE_DEBUG ((LM_DEBUG, "connecting supplier2\n"));
 237       consumer_proxy2->connect_push_supplier (supplier2.in (),
 238                                              supplier_qos2.get_SupplierQOS ());
 239       ACE_DEBUG ((LM_DEBUG, "suppliers connected\n"));
 240
 241       // ****************************************************************
 242
 243       //Timer Registration part
 244
 245       //Timeout consumers for the two suppliers.
 246       Timeout_Consumer timeout_consumer_impl1(&supplier_impl1);
 247       Timeout_Consumer timeout_consumer_impl2(&supplier_impl2);
 248
 249       RtecScheduler::handle_t supplier1_timeout_consumer_rt_info =
 250         scheduler->create ("supplier1_timeout_consumer");
 251
 252       //Period = 1sec
 253       tv.set (1,0);
 254       ORBSVCS_Time::Time_Value_to_TimeT (tmp, tv);
 255
 256       scheduler->set (supplier1_timeout_consumer_rt_info,
 257                       RtecScheduler::VERY_LOW_CRITICALITY,
 258                       tmp, tmp, tmp,
 259                       time_val_to_period (tv),
 260                       RtecScheduler::VERY_LOW_IMPORTANCE,
 261                       tmp,
 262                       0,
 263                       RtecScheduler::OPERATION);
 264
 265       RtecScheduler::handle_t supplier2_timeout_consumer_rt_info =
 266         scheduler->create ("supplier2_timeout_consumer");
 267
 268       //Period = 3sec
 269       tv.set (3, 0);
 270       ORBSVCS_Time::Time_Value_to_TimeT (tmp, tv);
 271
 272       scheduler->set (supplier2_timeout_consumer_rt_info,
 273                       RtecScheduler::VERY_HIGH_CRITICALITY,
 274                       tmp, tmp, tmp,
 275                       time_val_to_period (tv),
 276                       RtecScheduler::VERY_HIGH_IMPORTANCE,
 277                       tmp,
 278                       0,
 279                       RtecScheduler::OPERATION);
 280
 281       ACE_ConsumerQOS_Factory timer_qos1, timer_qos2;
 282       timer_qos1.insert_time (ACE_ES_EVENT_INTERVAL_TIMEOUT,
 283                               10000000, //in 100s of nanosec
 284                               supplier1_timeout_consumer_rt_info);
 285       timer_qos2.insert_time (ACE_ES_EVENT_INTERVAL_TIMEOUT,
 286                               30000000, //in 100s of nanosec
 287                               supplier2_timeout_consumer_rt_info);
 288
 289       RtecEventChannelAdmin::ProxyPushSupplier_var timeout_supplier_proxy1 =
 290         consumer_admin->obtain_push_supplier ();
 291
 292       RtecEventChannelAdmin::ProxyPushSupplier_var timeout_supplier_proxy2 =
 293         consumer_admin->obtain_push_supplier ();
 294
 295       RtecEventComm::PushConsumer_var safe_timeout_consumer1 =
 296         timeout_consumer_impl1._this ();
 297
 298       RtecEventComm::PushConsumer_var safe_timeout_consumer2 =
 299         timeout_consumer_impl2._this ();
 300
 301       ACE_DEBUG ((LM_DEBUG, "connecting timeout consumers\n"));
 302       timeout_supplier_proxy1->
 303         connect_push_consumer (safe_timeout_consumer1.in (),
 304                                timer_qos1.get_ConsumerQOS ());
 305
 306       timeout_supplier_proxy2->
 307         connect_push_consumer (safe_timeout_consumer2.in (),
 308                                timer_qos2.get_ConsumerQOS ());
 309
 310       ACE_DEBUG ((LM_DEBUG, "timeout consumers connected\n"));
 311
 312       // ****************************************************************
 313       //Registering dependency between timeout consumers and our suppliers
 314       //with the scheduler
 315
 316       scheduler->add_dependency (supplier1_timeout_consumer_rt_info,
 317                                  supplier1_rt_info,
 318                                  1,
 319                                  RtecBase::TWO_WAY_CALL);
 320
 321       scheduler->add_dependency (supplier2_timeout_consumer_rt_info,
 322                                  supplier2_rt_info,
 323                                  1,
 324                                  RtecBase::TWO_WAY_CALL);
 325
 326       // ****************************************************************
 327
 328       // At this point the consumer and supplier are connected to the
 329       // EC, they have provided their QoS info to the Scheduling
 330       // Service and the EC has informed the Scheduler about the
 331       // dependencies between them.
 332       // We can now compute the schedule for this configuration...
 333
 334       // The schedule is returned in this variables....
 335
 336       ACE_DEBUG ((LM_DEBUG, "Computing schedule\n"));
 337       RtecScheduler::RT_Info_Set_var infos;
 338       RtecScheduler::Config_Info_Set_var configs;
 339       RtecScheduler::Dependency_Set_var dependencies;
 340       RtecScheduler::Scheduling_Anomaly_Set unsafe_anomalies;
 341       RtecScheduler::Scheduling_Anomaly_Set_var anomalies;
 342
 343       scheduler->get_rt_info_set (infos.out() );
 344       scheduler->get_dependency_set (dependencies.out() );
 345       scheduler->get_config_info_set (configs.out() );
 346
 347       ACE_DEBUG ((LM_DEBUG, "Printing intermediate results\n"));
 348       ACE_Scheduler_Factory::dump_schedule (infos.in (),
 349                                             dependencies.in (),
 350                                             configs.in (),
 351                                             unsafe_anomalies,
 352                                             ACE_TEXT("schedule.out"));
 353
 354       // Obtain the range of valid priorities in the current
 355       // platform, the scheduler hard-code this values in the
 356       // generated file, but in the future we may just use the
 357       // "logical" priorities and define the mapping to OS
 358       // priorities at run-time.
 359       int min_os_priority =
 360         ACE_Sched_Params::priority_min (ACE_SCHED_FIFO,
 361                                         ACE_SCOPE_THREAD);
 362       int max_os_priority =
 363         ACE_Sched_Params::priority_max (ACE_SCHED_FIFO,
 364                                         ACE_SCOPE_THREAD);
 365       scheduler->compute_scheduling (min_os_priority,
 366                                      max_os_priority,
 367                                      infos.out (),
 368                                      dependencies.out (),
 369                                      configs.out (),
 370                                      anomalies.out ());
 371
 372       // Dump the schedule to a file..
 373       ACE_Scheduler_Factory::dump_schedule (infos.in (),
 374                                             dependencies.in (),
 375                                             configs.in (),
 376                                             anomalies.in (),
 377                                             ACE_TEXT("schedule.out"));
 378
 379       // ****************************************************************
 380       ACE_DEBUG ((LM_DEBUG, "Pushing events\n"));
 381
 382       ACE_hthread_t thr_handle;
 383       ACE_Thread::self (thr_handle);
 384
 385       int prio = ACE_Sched_Params::priority_max (ACE_SCHED_FIFO);
 386       ACE_OS::thr_setprio (thr_handle, prio);
 387
 388 //     // Generate a few events....
 389 //       RtecEventComm::EventSet event1 (1);
 390 //       event1.length (1);
 391 //       event1[0].header.type   = ACE_ES_EVENT_UNDEFINED;
 392 //       event1[0].header.source = supplier_id1;
 393 //       event1[0].header.ttl    = 1;
 394
 395 //       RtecEventComm::EventSet event2 (1);
 396 //       event2.length (1);
 397 //       event2[0].header.type   = ACE_ES_EVENT_UNDEFINED + 1;
 398 //       event2[0].header.source = supplier_id2;
 399 //       event2[0].header.ttl    = 1;
 400
 401 //       for (int i = 0; i != 200; ++i)
 402 //         {
 403 //           if (i % 2 == 0)
 404 //             {
 405 //               consumer_proxy1->push (event1);
 406 //             }
 407 //           else
 408 //             {
 409 //               consumer_proxy2->push (event2);
 410 //             }
 411
 412 //           ACE_Time_Value rate (0, 10000);
 413 //           ACE_OS::sleep (rate);
 414 //         }
 415
 416       ACE_DEBUG ((LM_DEBUG, "(%t) activating EC\n"));
 417       ec_impl.activate ();
 418       ACE_DEBUG ((LM_DEBUG, "EC activated\n"));
 419
 420       orb->run ();
 421
 422       // ****************************************************************
 423
 424       // We should do a lot of cleanup (disconnect from the EC,
 425       // deactivate all the objects with the POA, etc.) but this is
 426       // just a simple demo so we are going to be lazy.
 427
 428     }
 429   catch (const CORBA::Exception& ex)
 430     {
 431       ex._tao_print_exception ("Service");
 432       return 1;
 433     }
 434   return 0;
 435 }
 436
 437 // ****************************************************************
 438
 439 int parse_args (int argc, ACE_TCHAR *argv[])
 440 {
 441   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("cs:"));
 442   int c;
 443
 444   while ((c = get_opts ()) != -1)
 445     switch (c)
 446       {
 447       case 's':
 448         sched_type = ACE_TEXT_ALWAYS_CHAR(get_opts.opt_arg ());
 449         break;
 450
 451       case '?':
 452       default:
 453         ACE_ERROR_RETURN ((LM_ERROR,
 454                            "usage:  %s %s"
 455                            "\n",
 456                            argv [0],
 457                            "-s <rms|muf>"),
 458                           -1);
 459       }
 460   // Indicates successful parsing of the command line
 461   return 0;
 462 }
 463