ACE/Kokyu/DSRT_CV_Dispatcher_Impl_T.cpp

   1 #ifndef DSRT_CV_DISPATCHER_IMPL_T_CPP
   2 #define DSRT_CV_DISPATCHER_IMPL_T_CPP
   3
   4 #include "DSRT_CV_Dispatcher_Impl_T.h"
   5
   6 #if !defined (__ACE_INLINE__)
   7 //#include "DSRT_CV_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
  37 template <class DSRT_Scheduler_Traits>
  38 DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
  39 DSRT_CV_Dispatcher_Impl (ACE_Sched_Params::Policy sched_policy,
  40                          int sched_scope)
  41   :DSRT_Dispatcher_Impl<DSRT_Scheduler_Traits> (sched_policy, sched_scope),
  42    run_cond_ (run_cond_lock_)
  43 {
  44 }
  45
  46 template <class DSRT_Scheduler_Traits> int
  47 DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
  48 init_i (const DSRT_ConfigInfo&)
  49 {
  50   return 0;
  51 }
  52
  53 template <class DSRT_Scheduler_Traits>
  54 int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
  55 schedule_i (Guid_t id, const DSRT_QoSDescriptor& qos)
  56 {
  57 #ifdef KOKYU_DSRT_LOGGING
  58   ACE_DEBUG ((LM_DEBUG,
  59               "(%t|%T):schedule_i enter\n"));
  60 #endif
  61
  62   DSRT_Dispatch_Item<DSRT_Scheduler_Traits>* item;
  63   ACE_hthread_t thr_handle;
  64   ACE_Thread::self (thr_handle);
  65
  66   if (ACE_OS::thr_setprio (thr_handle,
  67                            this->blocked_prio_,
  68                            this->sched_policy_) == -1)
  69     {
  70       ACE_ERROR_RETURN ((LM_ERROR,
  71                   ACE_TEXT ("%p\n"),
  72                   ACE_TEXT ("thr_setprio failed")), -1);
  73     }
  74
  75   ACE_NEW_RETURN (item,
  76                   DSRT_Dispatch_Item<DSRT_Scheduler_Traits> (id, qos),
  77                   -1);
  78   item->thread_handle (thr_handle);
  79
  80   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
  81   if (this->ready_queue_.insert (item) == -1)
  82     return -1;
  83
  84 #ifdef KOKYU_DSRT_LOGGING
  85   this->ready_queue_.dump ();
  86
  87   ACE_DEBUG ((LM_DEBUG,
  88               "(%t|%T):schedule_i after ready_q.insert\n"));
  89 #endif
  90
  91   DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> item_var;
  92
  93   ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, cond_guard, run_cond_lock_, -1);
  94   this->ready_queue_.most_eligible (item_var);
  95
  96   guard.release ();
  97
  98   ACE_hthread_t most_eligible_thr_handle = item_var->thread_handle ();
  99
 100 #ifdef KOKYU_DSRT_LOGGING
 101       ACE_DEBUG ((LM_DEBUG,
 102                   "(%t|%T):curr thr handle = %d\n",
 103                   thr_handle));
 104       ACE_DEBUG ((LM_DEBUG,
 105                   "(%t|%T):curr scheduled thr handle = %d\n",
 106                   this->curr_scheduled_thr_handle_));
 107       ACE_DEBUG ((LM_DEBUG,
 108                   "(%t|%T):most eligible thr handle = %d\n",
 109                   most_eligible_thr_handle));
 110 #endif
 111
 112   if (this->curr_scheduled_thr_handle_ == thr_handle &&
 113       most_eligible_thr_handle != thr_handle)
 114     {
 115 #ifdef KOKYU_DSRT_LOGGING
 116       ACE_DEBUG ((LM_DEBUG,
 117                   "(%t|%T):curr sched thr handle = thr_handle & "
 118                   "most eligible thr handle != curr thr handle. "
 119                   "about to do a broadcast on CV to wake up most eligible\n"));
 120 #endif
 121       this->curr_scheduled_thr_handle_ = most_eligible_thr_handle;
 122       //wake up the most eligible thread
 123       this->run_cond_.broadcast ();
 124     }
 125
 126   //if the current thread is not the most eligible, then wait.
 127   //if the current thread is most eligible, but some thread is
 128   //scheduled currently, then wait.
 129   while (most_eligible_thr_handle != thr_handle ||
 130          (most_eligible_thr_handle == thr_handle &&
 131           this->curr_scheduled_thr_handle_ != thr_handle &&
 132           this->curr_scheduled_thr_handle_ != 0))
 133     {
 134       ACE_Time_Value tv (60,0);
 135       tv += ACE_OS::gettimeofday ();
 136       //wait a maximum of 1 min. This is an escape latch against lockups.
 137 #ifdef KOKYU_DSRT_LOGGING
 138       ACE_DEBUG ((LM_DEBUG,
 139                   "(%t|%T): About to block on cv\n"));
 140 #endif
 141       if (this->run_cond_.wait (&tv) == -1)
 142         {
 143           ACE_ERROR ((LM_ERROR,
 144                       "(%t|%T): run_cond.wait timed out -- Possible Lockup\n"));
 145         }
 146       this->ready_queue_.most_eligible (item_var);
 147       most_eligible_thr_handle = item_var->thread_handle ();
 148     }
 149   this->curr_scheduled_guid_ = item_var->guid ();
 150   this->curr_scheduled_thr_handle_ = most_eligible_thr_handle;
 151
 152 #ifdef KOKYU_DSRT_LOGGING
 153       ACE_DEBUG ((LM_DEBUG,
 154                   "(%t|%T): %d is currently running\n",
 155                   thr_handle));
 156 #endif
 157
 158   if (ACE_OS::thr_setprio (thr_handle,
 159                            this->active_prio_,
 160                            this->sched_policy_) == -1)
 161     {
 162       ACE_ERROR ((LM_ERROR,
 163                   ACE_TEXT ("%p\n"),
 164                   ACE_TEXT ("thr_setprio failed")));
 165     }
 166
 167 #ifdef KOKYU_DSRT_LOGGING
 168   ACE_DEBUG ((LM_DEBUG,
 169               "(%t|%T):schedule_i exit\n"));
 170 #endif
 171
 172   return 0;
 173 }
 174
 175 template <class DSRT_Scheduler_Traits>
 176 int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 177 update_schedule_i (Guid_t guid, const DSRT_QoSDescriptor& qos)
 178 {
 179   return this->schedule_i (guid, qos);
 180 }
 181
 182 template <class DSRT_Scheduler_Traits>
 183 int DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 184 update_schedule_i (Guid_t guid, Block_Flag_t flag)
 185 {
 186   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
 187
 188 #ifdef KOKYU_DSRT_LOGGING
 189   ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block entered\n"));
 190 #endif
 191
 192   DSRT_Dispatch_Item_var<DSRT_Scheduler_Traits> dispatch_item;
 193   ACE_hthread_t thr_handle;
 194   ACE_Thread::self (thr_handle);
 195
 196   int found = this->ready_queue_.find (guid, dispatch_item);
 197
 198 #ifdef KOKYU_DSRT_LOGGING
 199   if (found == 0)
 200       ACE_DEBUG ((LM_DEBUG, "(%t|%T): %d found in ready queue\n", thr_handle));
 201   else
 202       ACE_DEBUG ((LM_DEBUG, "(%t|%T): %d not found in ready queue\n",
 203                   thr_handle));
 204 #endif
 205
 206   if (found == 0 && flag == BLOCK)
 207     {
 208       thr_handle = dispatch_item->thread_handle ();
 209
 210 #ifdef KOKYU_DSRT_LOGGING
 211       ACE_DEBUG ((LM_DEBUG, "(%t|%T): update schedule: %d found\n", thr_handle));
 212 #endif
 213
 214       if (ACE_OS::thr_setprio (thr_handle,
 215                                this->blocked_prio_,
 216                                this->sched_policy_) == -1)
 217         {
 218           ACE_ERROR ((LM_ERROR,
 219                       ACE_TEXT ("%p\n"),
 220                       ACE_TEXT ("thr_setprio failed")));
 221         }
 222
 223       int rc = this->cancel_schedule (guid);
 224
 225 #ifdef KOKYU_DSRT_LOGGING
 226       ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
 227 #endif
 228
 229       return rc;
 230     }
 231
 232 #ifdef KOKYU_DSRT_LOGGING
 233   ACE_DEBUG ((LM_DEBUG, "(%t): update schedule for block done\n"));
 234 #endif
 235
 236   return -1;
 237 }
 238
 239 template <class DSRT_Scheduler_Traits> int
 240 DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 241 cancel_schedule_i (Guid_t guid)
 242 {
 243   ACE_GUARD_RETURN (ACE_SYNCH_RECURSIVE_MUTEX, guard, this->synch_lock_, -1);
 244
 245   ACE_hthread_t thr_handle;
 246   ACE_Thread::self (thr_handle);
 247
 248 #ifdef KOKYU_DSRT_LOGGING
 249   ACE_DEBUG ((LM_DEBUG, "(%t|%T): about to remove guid\n"));
 250 #endif
 251
 252   this->ready_queue_.remove (guid);
 253
 254 #ifdef KOKYU_DSRT_LOGGING
 255   this->ready_queue_.dump ();
 256 #endif
 257
 258   if (this->curr_scheduled_thr_handle_ == thr_handle)
 259     {
 260       this->curr_scheduled_guid_ = 0;
 261       this->curr_scheduled_thr_handle_ = 0;
 262     }
 263
 264   ACE_GUARD_RETURN (cond_lock_t,
 265                     mon, this->run_cond_lock_, 0);
 266   this->run_cond_.broadcast ();
 267   return 0;
 268 }
 269
 270 template <class DSRT_Scheduler_Traits> int
 271 DSRT_CV_Dispatcher_Impl<DSRT_Scheduler_Traits>::
 272 shutdown_i ()
 273 {
 274   this->shutdown_flagged_ = 1;
 275   return 0;
 276 }
 277
 278 }
 279
 280 #endif /* DSRT_CV_DISPATCHER_IMPL_T_CPP */