ACE/Kokyu/DSRT_Direct_Dispatcher_Impl_T.cpp

   1 #ifndef DSRT_DIRECT_DISPATCHER_IMPL_T_CPP
   2 #define DSRT_DIRECT_DISPATCHER_IMPL_T_CPP
   3
   4 #include "DSRT_Direct_Dispatcher_Impl_T.h"
   5
   6 #if !defined (__ACE_INLINE__)
   7 //#include "DSRT_Direct_Dispatcher_Impl_T.i"
   8 #endif /* __ACE_INLINE__ */
   9
  10 namespace Kokyu
  11 {
  12 /*
  13 //@@VS: This is somehow not being recognized by MSVC, which results
  14 //in a link error. For now, the definition has been moved to the .h
  15 //file. Needs further investigation.
  16
  17 template <class DSRT_Scheduler_Traits>
  18 int Comparator_Adapter_Generator<DSRT_Scheduler_Traits>::MoreEligible::
  19 operator ()(const DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits>& item1,
  20             const DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits>& item2)
  21 {
  22   int rc = qos_comparator_ (item1->qos (), item2->qos ());
  23
  24   //more eligible
  25   if (rc == 1)
  26     return 1;
  27
  28   //if equally eligible, then resolve tie with the creation time of
  29   //the item
  30   if (rc == 0 && item1->insertion_time () < item2->insertion_time ())
  31     return 1;
  32
  33   return 0;
  34 }
  35 */
  36 template <class DSRT_Scheduler_Traits>
  37 DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
  38 DSRT_Direct_Dispatcher_Impl (ACE_Sched_Params::Policy sched_policy,
  39                          int sched_scope)
  40   :DSRT_Dispatcher_Impl<DSRT_Scheduler_Traits>(sched_policy, sched_scope),
  41    sched_queue_modified_ (0),
  42    sched_queue_modified_cond_ (sched_queue_modified_cond_lock_)
  43 {
  44   //Run scheduler thread at highest priority
  45   if (this->activate (this->rt_thr_flags_, 1, 0, this->executive_prio_) == -1)
  46     {
  47       ACE_ERROR ((LM_ERROR,
  48                   "(%t|%T) cannot activate scheduler thread in RT mode."
  49                   "Trying in non RT mode\n"));
  50       if (this->activate (this->non_rt_thr_flags_) == -1)
  51         ACE_ERROR ((LM_ERROR,
  52                     "(%t|%T) cannot activate scheduler thread\n"));
  53     }
  54 }
  55
  56 template <class DSRT_Scheduler_Traits> int
  57 DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
  58 init_i (const DSRT_ConfigInfo&)
  59 {
  60   return 0;
  61 }
  62
  63 template <class DSRT_Scheduler_Traits> int
  64 DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::svc ()
  65 {
  66   ACE_hthread_t scheduler_thr_handle;
  67   ACE_Thread::self (scheduler_thr_handle);
  68
  69 #ifdef KOKYU_DSRT_LOGGING
  70   int prio;
  71   ACE_DEBUG ((LM_DEBUG,
  72               ACE_TEXT ("max prio=%d\n")
  73               ACE_TEXT ("min prio=%d\n")
  74               ACE_TEXT ("active prio=%d\n")
  75               ACE_TEXT ("inactive prio=%d\n"),
  76               max_prio_,
  77               min_prio_,
  78               active_prio_,
  79               inactive_prio_));
  80
  81   if (ACE_OS::thr_getprio (scheduler_thr_handle, prio) == -1)
  82     {
  83       if (errno == ENOTSUP)
  84         {
  85           ACE_ERROR((LM_ERROR,
  86                      ACE_TEXT ("getprio not supported\n")
  87                      ));
  88         }
  89       else
  90         {
  91           ACE_ERROR ((LM_ERROR,
  92                       ACE_TEXT ("%p\n")
  93                       ACE_TEXT ("thr_getprio failed")));
  94         }
  95     }
  96
  97   ACE_DEBUG ((LM_DEBUG, "(%t): Scheduler thread prio is %d\n", prio));
  98 #endif /*DSRT_LOGGING*/
  99
 100   while(1)
 101     {
 102       ACE_GUARD_RETURN (cond_lock_t,
 103                         mon,
 104                         sched_queue_modified_cond_lock_,
 105                         0);
 106
 107       if (this->shutdown_flagged_)
 108         break;
 109
 110       while (!sched_queue_modified_)
 111         {
 112 #ifdef KOKYU_DSRT_LOGGING
 113           ACE_DEBUG ((LM_DEBUG,
 114                       "(%t|%T): sched thread about to wait on cv\n"));
 115 #endif
 116           sched_queue_modified_cond_.wait ();
 117         }
 118
 119 #ifdef KOKYU_DSRT_LOGGING
 120       ACE_DEBUG ((LM_DEBUG, "(%t|%T): sched thread done waiting on cv\n"));
 121 #endif
 122
 123       sched_queue_modified_ = 0;
 124
 125       ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX,
 126                         synch_lock_mon,
 127                         this->synch_lock_,
 128                         0);
 129
 130       if (this->ready_queue_.current_size () <= 0)
 131         continue;
 132
 133 #ifdef KOKYU_DSRT_LOGGING
 134       ACE_DEBUG ((LM_DEBUG, "(%t|%T):Sched Queue contents===>\n"));
 135       this->ready_queue_.dump ();
 136 #endif
 137       DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> item_var;
 138       this->ready_queue_.most_eligible (item_var);
 139
 140       ACE_hthread_t most_eligible_thr_handle = item_var->thread_handle ();
 141
 142 #ifdef KOKYU_DSRT_LOGGING
 143       ACE_DEBUG ((LM_DEBUG,
 144                   "(%t|%T):curr scheduled thr handle = %d\n",
 145                   this->curr_scheduled_thr_handle_));
 146       ACE_DEBUG ((LM_DEBUG,
 147                   "(%t|%T):most eligible thr handle = %d\n",
 148                   most_eligible_thr_handle));
 149 #endif
 150
 151       if (this->curr_scheduled_thr_handle_ != most_eligible_thr_handle)
 152         {
 153           if (this->curr_scheduled_thr_handle_ != 0)
 154             {
 155               if (ACE_OS::thr_setprio (this->curr_scheduled_thr_handle_,
 156                                        this->inactive_prio_,
 157                                        this->sched_policy_) == -1)
 158                 {
 159                   ACE_ERROR ((LM_ERROR,
 160                               ACE_TEXT ("%p\n"),
 161                               ACE_TEXT ("thr_setprio on curr_scheduled_thr_handle_ failed.")));
 162                   ACE_DEBUG ((LM_DEBUG, "thr_handle = %d, prio = %d\n",
 163                               this->curr_scheduled_thr_handle_,
 164                               this->inactive_prio_));
 165                 }
 166             }
 167
 168           if (ACE_OS::thr_setprio (most_eligible_thr_handle,
 169                                    this->active_prio_, this->sched_policy_) == -1)
 170             {
 171               ACE_ERROR ((LM_ERROR,
 172                           ACE_TEXT ("%p\n"),
 173                           ACE_TEXT ("thr_setprio on most_eligible_thr_handle failed")));
 174             }
 175
 176           this->curr_scheduled_thr_handle_ = most_eligible_thr_handle;
 177           this->curr_scheduled_guid_ = item_var->guid ();
 178         }
 179           /*change all threads in blocked_prio_ to inactive_prio_*/
 180           this->ready_queue_.change_prio(this->blocked_prio_, this->inactive_prio_,this->sched_policy_);
 181     }
 182
 183 #ifdef KOKYU_DSRT_LOGGING
 184   ACE_DEBUG ((LM_DEBUG, "(%t): sched thread exiting\n"));
 185 #endif
 186
 187   return 0;
 188 }
 189
 190 template <class DSRT_Scheduler_Traits>
 191 int DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 192 schedule_i (Guid_t id, const DSRT_QoSDescriptor& qos)
 193 {
 194   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
 195
 196 #ifdef KOKYU_DSRT_LOGGING
 197   ACE_DEBUG ((LM_DEBUG,
 198               "(%t|%T):schedule_i enter\n"));
 199 #endif
 200
 201   DSRT_Dispatch_Item<DSRT_Scheduler_Traits>* item;
 202   ACE_hthread_t thr_handle;
 203   ACE_Thread::self (thr_handle);
 204
 205   ACE_NEW_RETURN (item,
 206                   DSRT_Dispatch_Item<DSRT_Scheduler_Traits> (id, qos),
 207                   -1);
 208   item->thread_handle (thr_handle);
 209
 210   if (this->ready_queue_.insert (item) == -1)
 211     return -1;
 212
 213 #ifdef KOKYU_DSRT_LOGGING
 214   ACE_DEBUG ((LM_DEBUG,
 215               "(%t|%T):schedule_i after ready_q.insert\n"));
 216 #endif
 217
 218   if (ACE_OS::thr_setprio (thr_handle,
 219                            this->blocked_prio_,
 220                            this->sched_policy_) == -1)
 221     {
 222       ACE_ERROR_RETURN ((LM_ERROR,
 223                   ACE_TEXT ("%p\n"),
 224                   ACE_TEXT ("thr_setprio failed")), -1);
 225     }
 226
 227 #ifdef KOKYU_DSRT_LOGGING
 228   ACE_DEBUG ((LM_DEBUG,
 229               "(%t|%T):schedule_i after thr_setprio\n"));
 230 #endif
 231
 232   //ready_queue_.dump ();
 233
 234   /*first release ready_queue_ lock. Otherwise if the scheduler gets the
 235   sched_queue_modified_cond_lock first, then try to get the ready_queue_ lock
 236   just when one thread who gets the ready_queue_ lock first, then try to get
 237   sched_queue_modified_cond_lock. Deadlock happens.
 238   */
 239   guard.release ();
 240
 241   //@@ Perhaps the lock could be moved further down just before
 242   //setting the condition variable?
 243   ACE_GUARD_RETURN (cond_lock_t,
 244                     mon, this->sched_queue_modified_cond_lock_, 0);
 245
 246 #ifdef KOKYU_DSRT_LOGGING
 247   ACE_DEBUG ((LM_DEBUG,
 248               "(%t|%T):schedule_i after acquiring cond lock\n"));
 249 #endif
 250
 251   this->sched_queue_modified_ = 1;
 252   this->sched_queue_modified_cond_.signal ();
 253
 254 #ifdef KOKYU_DSRT_LOGGING
 255   ACE_DEBUG ((LM_DEBUG,
 256               "(%t|%T):schedule_i exit\n"));
 257 #endif
 258
 259   return 0;
 260 }
 261
 262 template <class DSRT_Scheduler_Traits>
 263 int DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 264 update_schedule_i (Guid_t guid, const DSRT_QoSDescriptor& qos)
 265 {
 266   return this->schedule (guid, qos);
 267 }
 268
 269 template <class DSRT_Scheduler_Traits>
 270 int DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 271 update_schedule_i (Guid_t guid, Block_Flag_t flag)
 272 {
 273   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
 274
 275 #ifdef KOKYU_DSRT_LOGGING
 276   ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block entered\n"));
 277 #endif
 278
 279   DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> dispatch_item;
 280   ACE_hthread_t thr_handle;
 281
 282   int found = this->ready_queue_.find (guid, dispatch_item);
 283   if (found == 0 && flag == BLOCK)
 284     {
 285       thr_handle = dispatch_item->thread_handle ();
 286       if (ACE_OS::thr_setprio (thr_handle,
 287                                this->blocked_prio_,
 288                                this->sched_policy_) == -1)
 289         {
 290           ACE_ERROR ((LM_ERROR,
 291                       ACE_TEXT ("%p\n"),
 292                       ACE_TEXT ("thr_setprio failed")));
 293         }
 294
 295       //monitor released because cancel_schedule would acquire the
 296       //lock. Using recursive mutex creates lock up.
 297       //
 298       //@@ Need to investigate this further. Also we can consider
 299       //using the Thread-Safe interface pattern.
 300       //mon.release ();
 301       int rc = this->cancel_schedule (guid);
 302
 303 #ifdef KOKYU_DSRT_LOGGING
 304       ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
 305 #endif
 306
 307       return rc;
 308     }
 309
 310 #ifdef KOKYU_DSRT_LOGGING
 311   ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
 312 #endif
 313
 314   return -1;
 315 }
 316
 317 template <class DSRT_Scheduler_Traits> int
 318 DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 319 cancel_schedule_i (Guid_t guid)
 320 {
 321   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
 322
 323 #ifdef KOKYU_DSRT_LOGGING
 324   ACE_DEBUG ((LM_DEBUG, "(%t): about to remove guid\n"));
 325 #endif
 326
 327   this->ready_queue_.remove (guid);
 328
 329 #ifdef KOKYU_DSRT_LOGGING
 330   this->ready_queue_.dump ();
 331 #endif
 332
 333   if (this->curr_scheduled_guid_ == guid)
 334     {
 335       this->curr_scheduled_guid_ = 0;
 336       this->curr_scheduled_thr_handle_ = 0;
 337     }
 338
 339   //release ready_queue_ lock first before getting another lock
 340   guard.release ();
 341
 342   ACE_GUARD_RETURN (cond_lock_t,
 343                     mon, this->sched_queue_modified_cond_lock_, 0);
 344   this->sched_queue_modified_ = 1;
 345   this->sched_queue_modified_cond_.signal ();
 346   return 0;
 347 }
 348
 349 template <class DSRT_Scheduler_Traits> int
 350 DSRT_Direct_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 351 shutdown_i ()
 352 {
 353   this->shutdown_flagged_ = 1;
 354
 355   ACE_GUARD_RETURN (cond_lock_t, mon, this->sched_queue_modified_cond_lock_, 0);
 356   this->sched_queue_modified_ = 1;
 357   this->sched_queue_modified_cond_.signal ();
 358   // We have to wait until the scheduler executive thread shuts
 359   // down. But we have acquired the lock and if we wait without
 360   // releasing it, the scheduler thread will try to acquire it after
 361   // it gets woken up by the above signal and it fails to acquire the
 362   // lock. This will lead to a deadlock. So release the lock before we
 363   // wait.
 364   mon.release ();
 365   this->wait ();
 366   return 0;
 367 }
 368
 369 }
 370
 371 #endif /* DSRT_DIRECT_DISPATCHER_IMPL_T_CPP */