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 (void);
  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
 197       Worker worker1 (orb.in (), server.in (), current.in (), scheduler, 10, 15);
 198
 199       if (worker1.activate (flags, 1, 0, max_prio) != 0)
 200         {
 201           ACE_ERROR ((LM_ERROR,
 202                       "(%t|%T) cannot activate worker thread.\n"));
 203         }
 204
 205       ACE_OS::sleep(2);
 206
 207       Worker worker2 (orb.in (), server.in (), current.in (), scheduler, 12, 5);
 208
 209       if (worker2.activate (flags, 1, 0, max_prio) != 0)
 210         {
 211           ACE_ERROR ((LM_ERROR,
 212                       "(%t|%T) cannot activate scheduler thread in RT mode.\n"));
 213         }
 214
 215       worker1.wait ();
 216       worker2.wait ();
 217
 218       ACE_DEBUG ((LM_DEBUG,
 219                   "(%t): wait for worker threads done in main thread\n"));
 220
 221       if (do_shutdown)
 222         {
 223           if (enable_dynamic_scheduling)
 224             {
 225               MIF_Scheduling::SchedulingParameter sched_param;
 226               sched_param.importance = 0;
 227               CORBA::Policy_var sched_param_policy =
 228                 scheduler->create_scheduling_parameter (sched_param);
 229               CORBA::Policy_ptr implicit_sched_param = 0;
 230               current->begin_scheduling_segment (0,
 231                                                  sched_param_policy.in (),
 232                                                  implicit_sched_param);
 233             }
 234
 235             ACE_DEBUG ((LM_DEBUG, "(%t): about to call server shutdown\n"));
 236             server->shutdown ();
 237
 238             ACE_DEBUG ((LM_DEBUG, "after shutdown call in main thread\n"));
 239
 240
 241             if (enable_dynamic_scheduling)
 242             {
 243               current->end_scheduling_segment (0);
 244             }
 245         }
 246
 247       scheduler->shutdown ();
 248       ACE_DEBUG ((LM_DEBUG, "scheduler shutdown done\n"));
 249
 250       orb->destroy ();
 251     }
 252   catch (const CORBA::Exception& ex)
 253     {
 254       ex._tao_print_exception ("Exception caught:");
 255       return 1;
 256     }
 257
 258   return 0;
 259 }
 260
 261 // ****************************************************************
 262
 263 Worker::Worker (CORBA::ORB_ptr orb,
 264                 Simple_Server_ptr server_ptr,
 265                 RTScheduling::Current_ptr current,
 266                 MIF_Scheduler* scheduler,
 267                 CORBA::Short importance,
 268                 CORBA::Long server_load)
 269   //                int sleep_time)
 270   :  orb_ (CORBA::ORB::_duplicate (orb)),
 271      server_ (Simple_Server::_duplicate (server_ptr)),
 272      scheduler_current_ (RTScheduling::Current::_duplicate (current)),
 273      scheduler_ (scheduler),
 274      importance_ (importance),
 275      server_load_ (server_load)
 276      //     sleep_time_ (sleep_time)
 277 {
 278 }
 279
 280 int
 281 Worker::svc (void)
 282 {
 283   const char * name = 0;
 284   /*
 285   ACE_DEBUG ((LM_DEBUG, "(%t|%T):about to sleep for %d sec\n", sleep_time_));
 286   ACE_OS::sleep (sleep_time_);
 287   ACE_DEBUG ((LM_DEBUG, "(%t|%T):woke up from sleep for %d sec\n", sleep_time_));
 288   */
 289   ACE_hthread_t thr_handle;
 290   ACE_Thread::self (thr_handle);
 291   int prio;
 292
 293   if (ACE_Thread::getprio (thr_handle, prio) == -1)
 294     {
 295       if (errno == ENOTSUP)
 296         {
 297           ACE_ERROR((LM_ERROR,
 298                      ACE_TEXT ("getprio not supported\n")));
 299         }
 300       else
 301         {
 302           ACE_ERROR ((LM_ERROR,
 303                       ACE_TEXT ("%p\n")
 304                       ACE_TEXT ("thr_getprio failed")));
 305         }
 306     }
 307
 308   ACE_DEBUG ((LM_DEBUG, "(%t|%T) worker activated with prio %d\n", prio));
 309
 310   if (enable_dynamic_scheduling)
 311     {
 312       MIF_Scheduling::SchedulingParameter sched_param;
 313       CORBA::Policy_var sched_param_policy;
 314       sched_param.importance = importance_;
 315       sched_param_policy = scheduler_->create_scheduling_parameter (sched_param);
 316       CORBA::Policy_ptr implicit_sched_param = 0;
 317       ACE_DEBUG ((LM_DEBUG, "(%t|%T):before begin_sched_segment\n"));
 318       scheduler_current_->begin_scheduling_segment (name,
 319                                                     sched_param_policy.in (),
 320                                                     implicit_sched_param);
 321       ACE_DEBUG ((LM_DEBUG, "(%t|%T):after begin_sched_segment\n"));
 322     }
 323
 324   ACE_DEBUG ((LM_DEBUG, "(%t|%T):about to make two way call\n"));
 325   server_->test_method (server_load_);
 326   ACE_DEBUG ((LM_DEBUG, "(%t|%T):two way call done\n"));
 327
 328   if (enable_dynamic_scheduling)
 329     {
 330       scheduler_current_->end_scheduling_segment (name);
 331     }
 332
 333   ACE_DEBUG ((LM_DEBUG, "client worker thread (%t) done\n"));
 334
 335   return 0;
 336 }