ACE/Kokyu/Default_Dispatcher_Impl.cpp

   1 #include "Default_Dispatcher_Impl.h"
   2
   3 #include "ace/Sched_Params.h"
   4 #include <utility>
   5 #include <memory>
   6
   7 #if ! defined (__ACE_INLINE__)
   8 #include "Default_Dispatcher_Impl.inl"
   9 #endif /* __ACE_INLINE__ */
  10
  11 namespace Kokyu
  12 {
  13 Default_Dispatcher_Impl::Default_Dispatcher_Impl ()
  14   : activated_ (0)
  15 {
  16 }
  17
  18 int
  19 Default_Dispatcher_Impl::init_i (const Dispatcher_Attributes& attrs)
  20 {
  21   //create and init the dispatcher tasks here
  22
  23   ACE_DEBUG ((LM_DEBUG, "entering init_t\n" ));
  24   int size = ACE_Utils::truncate_cast<int> (attrs.config_info_set_.size ());
  25
  26   if (size == 0)
  27     {
  28       return -1;
  29     }
  30
  31   this->ntasks_ = size;
  32
  33   Dispatcher_Task_Auto_Ptr * tasks_array = 0;
  34   ACE_NEW_RETURN (tasks_array, Dispatcher_Task_Auto_Ptr[ntasks_], -1);
  35
  36   //ACE_DEBUG ((LM_DEBUG, "after new on task array\n" ));
  37   tasks_.reset(tasks_array);
  38
  39   //ACE_DEBUG ((LM_DEBUG, "task array unique_ptr set\n" ));
  40
  41   ConfigInfoSet& config_set =
  42     const_cast<ConfigInfoSet&> (attrs.config_info_set_);
  43   ConfigInfoSet::ITERATOR iter(config_set);
  44   int i=0;
  45
  46   ConfigInfo* config;
  47   for (;i<size && iter.next (config);iter.advance ())
  48     {
  49       //ACE_DEBUG ((LM_DEBUG, "iter = %d\n", i));
  50       Dispatcher_Task* task=0;
  51       ACE_NEW_RETURN (task,
  52                       Dispatcher_Task (*config,
  53                                        ACE_Thread_Manager::instance()),
  54                       -1);
  55       std::unique_ptr<Dispatcher_Task> tmp_task_unique_ptr (task);
  56       tasks_[i++] = std::move(tmp_task_unique_ptr);
  57     }
  58
  59   this->thr_creation_flags_ = attrs.thread_creation_flags ();
  60
  61   if (attrs.immediate_activation_ && !this->activated_)
  62     {
  63       this->activate_i ();
  64     }
  65
  66   curr_config_info_ = attrs.config_info_set_;
  67   return 0;
  68 }
  69
  70 int
  71 Default_Dispatcher_Impl::activate_i ()
  72 {
  73   int i;
  74
  75   if (this->activated_)
  76     return 0;
  77
  78   for(i=0; i<ntasks_; ++i)
  79     {
  80       Priority_t priority =
  81         tasks_[i]->get_curr_config_info ().thread_priority_;
  82
  83       if (this->tasks_[i]->activate (this->thr_creation_flags_,
  84                                      1, 1, priority) == -1)
  85         {
  86           ACE_ERROR_RETURN ((LM_ERROR,
  87              ACE_TEXT ("EC (%P|%t) cannot activate queue.")
  88              ACE_TEXT ("Need superuser privilege to run in RT class\n")),
  89              -1);
  90         }
  91     }
  92
  93   this->activated_ = 1;
  94   return 0;
  95 }
  96
  97 Dispatcher_Task*
  98 Default_Dispatcher_Impl::find_task_with_preemption_prio (Priority_t prio)
  99 {
 100   int i;
 101
 102   if (prio >=0)
 103   {
 104     for( i=0; i<ntasks_; ++i)
 105     {
 106       if ( tasks_[i]->preemption_priority () == prio)
 107         return  tasks_[i].get();
 108     }
 109   }
 110
 111   return 0;
 112 }
 113
 114 int
 115 Default_Dispatcher_Impl::dispatch_i (const Dispatch_Command* cmd,
 116                   const QoSDescriptor& qos_info)
 117 {
 118   //delegate to the appropriate task
 119   if (qos_info.preemption_priority_ < 0)
 120     return -1;
 121
 122   Dispatcher_Task* task =
 123     find_task_with_preemption_prio (qos_info.preemption_priority_);
 124
 125   //@@VS - We should insert this into the lowest prio queue.
 126   //How do we know that the last queue is the lowest prio queue.
 127   if (task == 0)
 128     task = tasks_[ntasks_-1].get ();
 129
 130   return task->enqueue (cmd, qos_info);
 131 }
 132
 133 int
 134 Default_Dispatcher_Impl::shutdown_i ()
 135 {
 136   //This needs to be revisited based on mode transition and
 137   //consistent cut through the queues
 138
 139   //post shutdown command to all tasks
 140   int i;
 141
 142   for(i=0; i<ntasks_; ++i)
 143     {
 144       QoSDescriptor qos_info;
 145       Shutdown_Task_Command* shutdown_cmd = 0;
 146       ACE_NEW_RETURN (shutdown_cmd, Shutdown_Task_Command, -1);
 147       tasks_[i]->enqueue (shutdown_cmd, qos_info);
 148     }
 149
 150   //wait for all tasks to exit
 151   for (i=0; i<ntasks_; ++i)
 152     {
 153       tasks_[i]->wait ();
 154     }
 155
 156   return 0;
 157 }
 158
 159 int
 160 Shutdown_Task_Command::execute ()
 161 {
 162   return -1;
 163 }
 164
 165 }