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