dynsched.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

int ACE_DynScheduler::number_of_dependencies(RT_Info* rt_info)
{
  return rt_info->dependencies.length();
}

int ACE_DynScheduler::number_of_dependencies(RT_Info& rt_info)
{
  return rt_info.dependencies.length();
}

int ACE_DynScheduler::add_dependency(RT_Info* rt_info,
                                     Dependency_Info& d)
{
  RT_Info *temp_info = 0; // temporary pointer to the caller's RT_Info

  switch (d.dependency_type)
  {
    case RtecBase::TWO_WAY_CALL:

      temp_info = rt_info;
      break;

    case RtecBase::ONE_WAY_CALL:

      // swap the handles and point to the caller instead of the called operation
      if (lookup_rt_info (d.rt_info, temp_info) != SUCCEEDED)
      {
        ACE_ERROR ((LM_ERROR,
                    ACE_LIB_TEXT("cannot find %d to add dependency\n"), d.rt_info));
        return -1;
      }

      d.rt_info = rt_info->handle;
      break;

    default:

      ACE_ERROR ((LM_ERROR,
                  ACE_LIB_TEXT("unrecognized dependency type %d for %s\n"),
                  d.dependency_type, ACE_TEXT_CHAR_TO_TCHAR(rt_info->entry_point.in ())));
      return -1;
  }

  ACE_DEBUG ((LM_DEBUG,
              ACE_LIB_TEXT("Sched (%t) adding %s dependency to caller: %s\n"),
              (const ACE_TCHAR *) ((d.dependency_type == RtecBase::TWO_WAY_CALL)
                              ? ACE_LIB_TEXT("TWO_WAY") : ACE_LIB_TEXT("ONE_WAY")),
              ACE_TEXT_CHAR_TO_TCHAR(temp_info->entry_point.in ())));

  RtecScheduler::Dependency_Set& set = temp_info->dependencies;
  int l = set.length();
  set.length(l + 1);
  set[l] = d;
  return 0;
}

void ACE_DynScheduler::export_to_file (RT_Info* info, FILE* file)
{
  ACE_DynScheduler::export_to_file (*info, file);
}

void ACE_DynScheduler::export_to_file (RT_Info& info, FILE* file)
{
  (void) ACE_OS::fprintf (file,
                          "%s\n%d\n%d\n%d\n%d\n%d\n%d\n%d\n%d\n%u\n"
                          "# begin calls\n%d\n",
                          info.entry_point.in (),
                          info.handle,
                          ACE_U64_TO_U32 (info.worst_case_execution_time),
                          ACE_U64_TO_U32 (info.typical_execution_time),
                          ACE_U64_TO_U32 (info.cached_execution_time),
                          int(info.period),
                          info.criticality,
                          info.importance,
                          ACE_U64_TO_U32 (info.quantum),
                          info.threads,
                          number_of_dependencies(info));

  for (int i = 0; i < number_of_dependencies(info); ++i)
    {
      RT_Info tmp;
      (void) ACE_OS::fprintf (file, "%s, %d\n",
                              (const char*)tmp.entry_point,
                              info.dependencies[i].number_of_calls);

    }

  (void) ACE_OS::fprintf (file, "# end calls\n%d\n%d\n\n",
                          info.priority,
                          info.preemption_subpriority);


}


int
ACE_DynScheduler::dispatch_configuration (const Preemption_Priority & p_priority,
                                          OS_Priority & priority,
                                          Dispatching_Type & d_type)
{
  // look up the stored configuration info for the given priority level
  Config_Info *config_info;
  if (lookup_config_info (p_priority, config_info) != SUCCEEDED)
  {
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_LIB_TEXT("Config info for priority %lu could not be found\n"),
                       p_priority),
                      -1);
  }

  priority = config_info->thread_priority;
  d_type = config_info->dispatching_type;

  return 0;
}
  // provide the thread priority and queue type for the given priority level


ACE_DynScheduler::status_t
ACE_DynScheduler::lookup_rt_info (handle_t handle,
                                  RT_Info*& rtinfo)
{
  if (handle < 0 || (size_t) handle > rt_info_entries_.size ())
  {
    return ST_UNKNOWN_TASK;
  }

  RT_Info** entry;
  ACE_Unbounded_Set_Iterator <RT_Info *> i (rt_info_entries_);
  while (i.next (entry) != 0)
    {
      i.advance ();
      RT_Info* info_ptr = *entry;
      if (info_ptr->handle == handle)
        {
          rtinfo = info_ptr;
          return SUCCEEDED;
        }
    }

  return ST_UNKNOWN_TASK;
}
  // obtains an RT_Info based on its "handle".

ACE_DynScheduler::status_t
ACE_DynScheduler::lookup_config_info (Preemption_Priority priority,
                                      Config_Info* &config_info)
{
  if (config_info_entries_ == 0)
  {
    return NOT_SCHEDULED;
  }

  if (priority < 0 || (size_t) priority > config_info_entries_->size ())
  {
    return ST_UNKNOWN_PRIORITY;
  }

  Config_Info** entry;
  ACE_Unbounded_Set_Iterator <Config_Info *> i (*config_info_entries_);
  while (i.next (entry) != 0)
    {
      i.advance ();
      Config_Info* config_ptr = *entry;
      if (config_ptr->preemption_priority == priority)
        {
          config_info = config_ptr;
          return SUCCEEDED;
        }
    }

  return ST_UNKNOWN_PRIORITY;
}
  // Obtains a Config_Info based on its priority.


void
ACE_DynScheduler::reset ()
{
  // if the schedule is up to date, free resources
  // and mark schedule as not being up to date
  if (up_to_date_)
  {
    delete [] task_entries_;
    task_entries_ = 0;

    delete [] ordered_task_entries_;
    ordered_task_entries_ = 0;

    delete thread_delineators_;
    thread_delineators_ = 0;

    delete [] ordered_thread_dispatch_entries_;
    ordered_thread_dispatch_entries_ = 0;

    if (dispatch_entries_)
    {
      // free all the dispatch entries in the list, then the list itself
      ACE_Unbounded_Set_Iterator <Dispatch_Entry *> iter (*dispatch_entries_);
      Dispatch_Entry **entry = 0;
      for (iter.first (); ! iter.done (); iter.advance (), entry = 0)
      {
        if ((iter.next (entry) != 0) && (entry) && (*entry))
        {
          delete (*entry);
        }
      }
      delete dispatch_entries_;
      dispatch_entries_ = 0;
    }

    if (config_info_entries_)
    {
      // free all the config info entries in the list, then the list itself
      ACE_Unbounded_Set_Iterator <Config_Info *> iter (*config_info_entries_);
      Config_Info **entry = 0;
      for (iter.first (); ! iter.done (); iter.advance (), entry = 0)
      {
        if ((iter.next (entry) != 0) && (entry) && (*entry))
        {
          delete (*entry);
        }
      }
      delete config_info_entries_;
      config_info_entries_ = 0;
    }


    if (expanded_dispatches_)
    {
      // free all the dispatch entries in the list, then the list itself
      ACE_Unbounded_Set_Iterator <Dispatch_Entry *> expanded_iter (*expanded_dispatches_);
      Dispatch_Entry **expanded_entry = 0;
      for (expanded_iter.first (); ! expanded_iter.done ();
           expanded_iter.advance (), expanded_entry = 0)
      {
        if ((expanded_iter.next (expanded_entry) != 0) &&
            (expanded_entry) && (*expanded_entry))
        {
          delete (*expanded_entry);
        }
      }
      delete expanded_dispatches_;
      expanded_dispatches_ = 0;
    }

    delete [] ordered_dispatch_entries_;
    ordered_dispatch_entries_ = 0;

    dispatch_entry_count_ = 0;
    threads_ = 0;

    status_ = NOT_SCHEDULED;

    frame_size_ = 1;
    critical_set_frame_size_ = 0;
    utilization_ = 0.0;
    critical_set_utilization_ = 0.0;
    minimum_priority_queue_ = 0;
    minimum_guaranteed_priority_queue_ = -1;

    if (timeline_)
    {
      // iterate over and delete the set of timeline entries
      ACE_Ordered_MultiSet_Iterator <TimeLine_Entry_Link> t_iter (*timeline_);
      TimeLine_Entry_Link *t_entry = 0;
      for (t_iter.first (); ! t_iter.done (); t_iter.advance (), t_entry = 0)
      {
        if ((t_iter.next (t_entry) != 0) && (t_entry))
        {
          delete &(t_entry->entry ());
        }
      }
      delete timeline_;
      timeline_ = 0;
         }

    up_to_date_ = 0;
  }
}

ACE_DynScheduler::status_t
ACE_DynScheduler::schedule (
  ACE_Unbounded_Set<RtecScheduler::Scheduling_Anomaly *> &anomaly_set)
{
  ACE_GUARD_RETURN (LOCK, ace_mon, lock_, ACE_DynScheduler::FAILED);

  RtecScheduler::Anomaly_Severity severity = RtecScheduler::ANOMALY_NONE;
  RtecScheduler::Anomaly_Severity temp_severity = RtecScheduler::ANOMALY_NONE;
  status_t temp_status = SUCCEEDED;
  Scheduling_Anomaly *anomaly = 0;
  ACE_CString unresolved_locals (""), unresolved_remotes ("");

  if (up_to_date_)
  {
    // do nothing if the RT_Infos have not changed
    // since the last valid schedule was generated
    return SUCCEEDED;
  }
  else
  {
    // save the total number of registered RT_Infos
    tasks (ACE_static_cast (u_int, rt_info_entries_.size ()));
  }

  // set up the task entry data structures
  status_ = setup_task_entries ();
  if (status_ != SUCCEEDED)
    {
      // Create an anomaly, add it to anomaly set
      anomaly = create_anomaly (status_);
      if (anomaly)
        {
          anomaly_set.insert (anomaly);
        }
      else
        {
          return ST_VIRTUAL_MEMORY_EXHAUSTED;
        }

      switch (anomaly->severity)
        {
          case RtecScheduler::ANOMALY_FATAL :
            return status_;

          default:
            severity = anomaly->severity;
            break;
        }
    }

  // check for cycles in the dependency graph: as a side effect, leaves
  // the ordered_task_entries_ pointer array sorted in topological order,
  // which is used by propagate_dispatches () to ensure that dispatches
  // are propagated top down in the call graph.
  temp_status = check_dependency_cycles ();
  if (temp_status != SUCCEEDED)
    {
      // Create an anomaly, add it to anomaly set
      anomaly = create_anomaly (temp_status);
      if (anomaly)
        {
          anomaly_set.insert (anomaly);
        }
      else
        {
          return ST_VIRTUAL_MEMORY_EXHAUSTED;
        }

      switch (anomaly->severity)
        {
          case RtecScheduler::ANOMALY_FATAL :
            status_ = temp_status;
            return status_;

          case RtecScheduler::ANOMALY_ERROR :
            severity = anomaly->severity;
            status_ = temp_status;
            break;

          case RtecScheduler::ANOMALY_WARNING :
            if (severity == RtecScheduler::ANOMALY_NONE)
              {
                severity = anomaly->severity;
                status_ = temp_status;
              }
            break;

          default:
            break;
        }
    }

  // task entries are related, now threads can be found
  temp_status = identify_threads (unresolved_locals,
                                  unresolved_remotes);
  if (temp_status != SUCCEEDED)
    {
      temp_severity = anomaly_severity (temp_status);
      switch (temp_severity)
        {
          case RtecScheduler::ANOMALY_FATAL :
            status_ = temp_status;
            return status_;

          case RtecScheduler::ANOMALY_ERROR :
            severity = temp_severity;
            status_ = temp_status;
            break;

          case RtecScheduler::ANOMALY_WARNING :
            if (severity == RtecScheduler::ANOMALY_NONE)
              {
                severity = temp_severity;
                status_ = temp_status;
              }
            break;

          default:
            break;
        }
    }

  // invoke the internal thread scheduling method of the strategy
  temp_status = schedule_threads (anomaly_set);
  if (temp_status != SUCCEEDED)
    {
      temp_severity = anomaly_severity (temp_status);
      switch (temp_severity)