TAO/examples/Kokyu_dsrt_schedulers/muf_example/client.cpp

   1 #include "ace/Get_Opt.h"
   2 #include "ace/Task.h"
   3 #include "ace/High_Res_Timer.h"
   4 #include "ace/OS_NS_unistd.h"
   5 #include "tao/RTScheduling/RTScheduler_Manager.h"
   6 #include "testC.h"
   7 #include "MUF_Scheduler.h"
   8 #include "orbsvcs/Time_Utilities.h"
   9
  10 const ACE_TCHAR *ior = ACE_TEXT("file://test.ior");
  11 int niterations = 5;
  12 int do_shutdown = 0;
  13 int enable_dynamic_scheduling = 0;
  14 int enable_yield = 1;
  15
  16 /**
  17  * Run a server thread
  18  *
  19  * Use the ACE_Task_Base class to run server threads
  20  */
  21 class Worker : public ACE_Task_Base
  22 {
  23 public:
  24   Worker (CORBA::ORB_ptr orb,
  25           Simple_Server_ptr server_ptr,
  26           RTScheduling::Current_ptr current,
  27           MUF_Scheduler* scheduler,
  28           TimeBase::TimeT deadline,
  29           TimeBase::TimeT estimated_initial_execution_time,
  30           long criticality,
  31           CORBA::Long server_load);
  32     //          int sleep_time);
  33   // ctor
  34
  35   virtual int svc (void);
  36   // The thread entry point.
  37
  38 private:
  39   CORBA::ORB_var orb_;
  40   // The orb
  41
  42   Simple_Server_var server_;
  43   RTScheduling::Current_var scheduler_current_;
  44   MUF_Scheduler* scheduler_;
  45   TimeBase::TimeT deadline_;
  46   TimeBase::TimeT estimated_initial_execution_time_;
  47   long criticality_;
  48   CORBA::Long server_load_;
  49   int sleep_time_;
  50 };
  51
  52 int
  53 parse_args (int argc, ACE_TCHAR *argv[])
  54 {
  55   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("xk:i:ds"));
  56   int c;
  57
  58   while ((c = get_opts ()) != -1)
  59     switch (c)
  60       {
  61       case 'x':
  62         do_shutdown = 1;
  63         break;
  64
  65       case 'k':
  66         ior = get_opts.opt_arg ();
  67         break;
  68
  69       case 'i':
  70         niterations = ACE_OS::atoi (get_opts.opt_arg ());
  71         break;
  72
  73       case 'd':
  74         enable_dynamic_scheduling = 1;
  75         break;
  76
  77       case 's':
  78         enable_yield = 0;
  79         break;
  80
  81       case '?':
  82       default:
  83         ACE_ERROR_RETURN ((LM_ERROR,
  84                            "usage:  %s "
  85                            "-k <ior> "
  86                            "-i <niterations> "
  87                            "-d (enable dynamic scheduling)"
  88                            "-s (disable yield)"
  89                            "\n",
  90                            argv [0]),
  91                           -1);
  92       }
  93   // Indicates successful parsing of the command line
  94   return 0;
  95 }
  96
  97 int
  98 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  99 {
 100   MUF_Scheduler* scheduler=0;
 101   RTScheduling::Current_var current;
 102   int prio;
 103   int max_prio;
 104   ACE_Sched_Params::Policy  sched_policy = ACE_SCHED_RR;
 105   int sched_scope = ACE_SCOPE_THREAD;
 106   long flags;
 107
 108   if (sched_policy == ACE_SCHED_RR)
 109     flags = THR_NEW_LWP | THR_BOUND | THR_JOINABLE | THR_SCHED_RR;
 110   else
 111     flags = THR_NEW_LWP | THR_BOUND | THR_JOINABLE | THR_SCHED_FIFO;
 112
 113   ACE_hthread_t main_thr_handle;
 114   ACE_Thread::self (main_thr_handle);
 115
 116   max_prio = ACE_Sched_Params::priority_max (sched_policy,
 117                                              sched_scope);
 118
 119   //FUZZ: disable check_for_lack_ACE_OS
 120   ACE_Sched_Params sched_params (sched_policy, max_prio);
 121   //FUZZ: enable check_for_lack_ACE_OS
 122
 123   ACE_OS::sched_params (sched_params);
 124
 125   if (ACE_Thread::getprio (main_thr_handle, prio) == -1)
 126     {
 127       if (errno == ENOTSUP)
 128         {
 129           ACE_ERROR((LM_ERROR,
 130                      ACE_TEXT ("getprio not supported\n")));
 131         }
 132       else
 133         {
 134           ACE_ERROR ((LM_ERROR,
 135                       ACE_TEXT ("%p\n")
 136                       ACE_TEXT ("thr_getprio failed")));
 137         }
 138     }
 139
 140   ACE_DEBUG ((LM_DEBUG, "(%t): main thread prio is %d\n", prio));
 141
 142   try
 143     {
 144       RTScheduling::Scheduler_var sched_owner;
 145
 146       CORBA::ORB_var orb =
 147         CORBA::ORB_init (argc, argv);
 148
 149       if (parse_args (argc, argv) != 0)
 150         return 1;
 151
 152       CORBA::Object_var object =
 153         orb->string_to_object (ior);
 154
 155       Simple_Server_var server =
 156         Simple_Server::_narrow (object.in ());
 157
 158       if (CORBA::is_nil (server.in ()))
 159         {
 160           ACE_ERROR_RETURN ((LM_ERROR,
 161                              "Object reference <%s> is nil.\n",
 162                              ior),
 163                             1);
 164         }
 165
 166       if (enable_dynamic_scheduling)
 167         {
 168           ACE_DEBUG ((LM_DEBUG, "Dyn Sched enabled\n"));
 169           CORBA::Object_var manager_obj =
 170             orb->resolve_initial_references ("RTSchedulerManager");
 171
 172           TAO_RTScheduler_Manager_var manager =
 173             TAO_RTScheduler_Manager::_narrow (manager_obj.in ());
 174
 175           Kokyu::DSRT_Dispatcher_Impl_t disp_impl_type;
 176           if (enable_yield)
 177             {
 178               disp_impl_type = Kokyu::DSRT_CV_BASED;
 179             }
 180           else
 181             {
 182               disp_impl_type = Kokyu::DSRT_OS_BASED;
 183             }
 184
 185           ACE_NEW_RETURN (scheduler,
 186                           MUF_Scheduler (orb.in (),
 187                                          disp_impl_type,
 188                                          sched_policy,
 189                                          sched_scope), -1);
 190           sched_owner = scheduler;
 191
 192           manager->rtscheduler (scheduler);
 193
 194           CORBA::Object_var object =
 195             orb->resolve_initial_references ("RTScheduler_Current");
 196
 197           current  =
 198             RTScheduling::Current::_narrow (object.in ());
 199
 200         }
 201
 202       TimeBase::TimeT deadline;
 203       TimeBase::TimeT exec_time;
 204       int criticality=0;
 205
 206       ORBSVCS_Time::Time_Value_to_TimeT (deadline,
 207                                          ACE_OS::gettimeofday () +
 208                                          ACE_Time_Value (50,0) );
 209
 210       ORBSVCS_Time::Time_Value_to_TimeT (exec_time,
 211                                          ACE_OS::gettimeofday () +
 212                                          ACE_Time_Value (10,0) );
 213
 214       Worker worker1 (orb.in (),
 215                       server.in (),
 216                       current.in (),
 217                       scheduler,
 218                       deadline,
 219                       exec_time,
 220                       criticality,
 221                       30);
 222
 223       if (worker1.activate (flags, 1, 0, max_prio) != 0)
 224         {
 225           ACE_ERROR ((LM_ERROR,
 226                       "(%t|%T) cannot activate worker thread.\n"));
 227         }
 228
 229       ACE_OS::sleep(2);
 230
 231       ORBSVCS_Time::Time_Value_to_TimeT (deadline,
 232                                          ACE_OS::gettimeofday () +
 233                                          ACE_Time_Value (30,0) );
 234
 235       ORBSVCS_Time::Time_Value_to_TimeT (exec_time,
 236                                          ACE_OS::gettimeofday () +
 237                                          ACE_Time_Value (10,0) );
 238
 239       criticality = 0;
 240       Worker worker2 (orb.in (),
 241                       server.in (),
 242                       current.in (),
 243                       scheduler,
 244                       deadline,
 245                       exec_time,
 246                       criticality,
 247                       10);
 248
 249       if (worker2.activate (flags, 1, 0, max_prio) != 0)
 250         {
 251           ACE_ERROR ((LM_ERROR,
 252                       "(%t|%T) cannot activate scheduler thread in RT mode.\n"));
 253         }
 254
 255       ORBSVCS_Time::Time_Value_to_TimeT (deadline,
 256                                          ACE_OS::gettimeofday () +
 257                                          ACE_Time_Value (100,0) );
 258
 259       ORBSVCS_Time::Time_Value_to_TimeT (exec_time,
 260                                          ACE_OS::gettimeofday () +
 261                                          ACE_Time_Value (10,0) );
 262       criticality = 1;
 263
 264       Worker worker3 (orb.in (),
 265                       server.in (),
 266                       current.in (),
 267                       scheduler,
 268                       deadline,
 269                       exec_time,
 270                       criticality,
 271                       5);
 272       if (worker3.activate (flags, 1, 0, max_prio) != 0)
 273         {
 274           ACE_ERROR ((LM_ERROR,
 275                       "(%t|%T) cannot activate scheduler thread in RT mode.\n"));
 276         }
 277
 278       worker1.wait ();
 279       worker2.wait ();
 280       worker3.wait ();
 281
 282       ACE_DEBUG ((LM_DEBUG,
 283                   "(%t): wait for worker threads done in main thread\n"));
 284
 285       if (do_shutdown)
 286         {
 287           if (enable_dynamic_scheduling)
 288             {
 289               MUF_Scheduling::SchedulingParameter sched_param;
 290               sched_param.criticality = 0;
 291               sched_param.deadline = 0;
 292               sched_param.estimated_initial_execution_time = 0;
 293               CORBA::Policy_var sched_param_policy =
 294                 scheduler->create_scheduling_parameter (sched_param);
 295               CORBA::Policy_ptr implicit_sched_param = 0;
 296               current->begin_scheduling_segment (0,
 297                                                  sched_param_policy.in (),
 298                                                  implicit_sched_param);
 299             }
 300
 301             ACE_DEBUG ((LM_DEBUG, "(%t): about to call server shutdown\n"));
 302             server->shutdown ();
 303
 304             ACE_DEBUG ((LM_DEBUG, "after shutdown call in main thread\n"));
 305
 306
 307             if (enable_dynamic_scheduling)
 308             {
 309               current->end_scheduling_segment (0);
 310             }
 311         }
 312
 313       scheduler->shutdown ();
 314       ACE_DEBUG ((LM_DEBUG, "scheduler shutdown done\n"));
 315
 316       orb->destroy ();
 317     }
 318   catch (const CORBA::Exception& ex)
 319     {
 320       ex._tao_print_exception ("Exception caught:");
 321       return 1;
 322     }
 323
 324   return 0;
 325 }
 326
 327 // ****************************************************************
 328
 329 Worker::Worker (CORBA::ORB_ptr orb,
 330                 Simple_Server_ptr server_ptr,
 331                 RTScheduling::Current_ptr current,
 332                 MUF_Scheduler* scheduler,
 333                 TimeBase::TimeT deadline,
 334                 TimeBase::TimeT estimated_initial_execution_time,
 335                 long criticality,
 336                 CORBA::Long server_load)
 337   //                int sleep_time)
 338   :  orb_ (CORBA::ORB::_duplicate (orb)),
 339      server_ (Simple_Server::_duplicate (server_ptr)),
 340      scheduler_current_ (RTScheduling::Current::_duplicate (current)),
 341      scheduler_ (scheduler),
 342      deadline_ (deadline),
 343      estimated_initial_execution_time_ ( estimated_initial_execution_time),
 344      criticality_ (criticality),
 345      server_load_ (server_load)
 346      //     sleep_time_ (sleep_time)
 347 {
 348 }
 349
 350 int
 351 Worker::svc (void)
 352 {
 353   const char * name = 0;
 354   /*
 355   ACE_DEBUG ((LM_DEBUG, "(%t|%T):about to sleep for %d sec\n", sleep_time_));
 356   ACE_OS::sleep (sleep_time_);
 357   ACE_DEBUG ((LM_DEBUG, "(%t|%T):woke up from sleep for %d sec\n", sleep_time_));
 358   */
 359   ACE_hthread_t thr_handle;
 360   ACE_Thread::self (thr_handle);
 361   int prio;
 362
 363   if (ACE_Thread::getprio (thr_handle, prio) == -1)
 364     {
 365       if (errno == ENOTSUP)
 366         {
 367           ACE_ERROR((LM_ERROR,
 368                      ACE_TEXT ("getprio not supported\n")));
 369         }
 370       else
 371         {
 372           ACE_ERROR ((LM_ERROR,
 373                       ACE_TEXT ("%p\n")
 374                       ACE_TEXT ("thr_getprio failed")));
 375         }
 376     }
 377
 378   ACE_DEBUG ((LM_DEBUG, "(%t|%T) worker activated with prio %d\n", prio));
 379
 380   if (enable_dynamic_scheduling)
 381     {
 382       MUF_Scheduling::SchedulingParameter sched_param;
 383       CORBA::Policy_var sched_param_policy;
 384       sched_param.criticality = criticality_;
 385       sched_param.deadline = deadline_;
 386       sched_param.estimated_initial_execution_time = 0;
 387       sched_param_policy = scheduler_->create_scheduling_parameter (sched_param);
 388       CORBA::Policy_ptr implicit_sched_param = 0;
 389       ACE_DEBUG ((LM_DEBUG, "(%t|%T):before begin_sched_segment\n"));
 390       scheduler_current_->begin_scheduling_segment (name,
 391                                                     sched_param_policy.in (),
 392                                                     implicit_sched_param);
 393       ACE_DEBUG ((LM_DEBUG, "(%t|%T):after begin_sched_segment\n"));
 394     }
 395
 396   ACE_DEBUG ((LM_DEBUG, "(%t|%T):about to make two way call\n"));
 397   server_->test_method (server_load_);
 398   ACE_DEBUG ((LM_DEBUG, "(%t|%T):two way call done\n"));
 399
 400   if (enable_dynamic_scheduling)
 401     {
 402       scheduler_current_->end_scheduling_segment (name);
 403     }
 404
 405   ACE_DEBUG ((LM_DEBUG, "client worker thread (%t) done\n"));
 406
 407   return 0;
 408 }