dynsched.cpp

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

  {
    return return_status;
  }

  // mark the entry as visited
  entry.dfs_status (Task_Entry::VISITED);

  // check all the calling operations: if there is one that has not already been
  // visited, mark the return status indicating there is a cycle, print
  // an error message to that effect, and recurse on that dependant
  Task_Entry_Link **calling_entry_link;
  ACE_Unbounded_Set_Iterator <Task_Entry_Link *> i (entry.callers ());
  while (i.next (calling_entry_link) != 0)
  {
    i.advance ();
    if ((*calling_entry_link)->caller ().dfs_status () == Task_Entry::NOT_VISITED)
    {
      // indicate the two tasks are in (the same) dependency cycle
      ACE_ERROR ((LM_ERROR,
                  ACE_LIB_TEXT("Tasks \"%s\" and \"%s\" are part of a call cycle.\n"),
                  ACE_TEXT_CHAR_TO_TCHAR((*calling_entry_link)->caller ().rt_info ()->entry_point.in ()),
                  ACE_TEXT_CHAR_TO_TCHAR(entry.rt_info ()->entry_point.in ())));

      // set return status, ignore status returned by recursive call:
      // we already know there are cycles in the dependencies
      return_status = ST_CYCLE_IN_DEPENDENCIES;
      check_dependency_cycles_recurse ((*calling_entry_link)->caller ());
    }
  }

  // mark the entry as finished
  entry.dfs_status (Task_Entry::FINISHED);

  return return_status;
}


ACE_DynScheduler::status_t
ACE_DynScheduler::schedule_threads (ACE_Unbounded_Set<RtecScheduler::Scheduling_Anomaly *> &anomaly_set)
{
  // make sure there are as many thread delineator
  // entries in the set as the counter indicates
  if (threads_ != thread_delineators_->size ())
  {
    return THREAD_COUNT_MISMATCH;
  }

  // allocate an array of pointers to the thread delineators
  ACE_NEW_RETURN (ordered_thread_dispatch_entries_,
                  Dispatch_Entry * [threads_],
                  ST_VIRTUAL_MEMORY_EXHAUSTED);
  ACE_OS::memset (ordered_thread_dispatch_entries_, 0,
                  sizeof (Dispatch_Entry *) * threads_);


  // copy pointers to the thread delineators from the set to the array
  ACE_Unbounded_Set_Iterator <Dispatch_Entry *> iter (*thread_delineators_);
  for (u_int i = 0; i < threads_; ++i, iter.advance ())
  {
    Dispatch_Entry** dispatch_entry;

    if (! iter.next (dispatch_entry))
    {
      return ST_BAD_INTERNAL_POINTER;
    }

    ordered_thread_dispatch_entries_ [i] = *dispatch_entry;
  }

  // sort the thread dispatch entries into priority order
  status_t status = sort_dispatches (ordered_thread_dispatch_entries_, threads_);

  if (status == SUCCEEDED)
  {
    // assign priorities to the thread dispatch entries
    status = assign_priorities (ordered_thread_dispatch_entries_,
                                threads_, anomaly_set);
  }

  return status;
}
  // thread scheduling method: sets up array of pointers to task
  // entries that are threads, calls internal thread scheduling method

ACE_DynScheduler::status_t
ACE_DynScheduler::schedule_dispatches (ACE_Unbounded_Set<RtecScheduler::Scheduling_Anomaly *> &anomaly_set)
{
  dispatch_entry_count_ = ACE_static_cast (u_int, dispatch_entries_->size ());

  ACE_NEW_RETURN (ordered_dispatch_entries_,
                  Dispatch_Entry * [dispatch_entry_count_],
                  ST_VIRTUAL_MEMORY_EXHAUSTED);
  ACE_OS::memset (ordered_dispatch_entries_, 0,
                  sizeof (Dispatch_Entry *) * dispatch_entry_count_);

  ACE_Unbounded_Set_Iterator <Dispatch_Entry *> iter (*dispatch_entries_);
  for (u_int i = 0; i < dispatch_entry_count_; ++i, iter.advance ())
  {
    Dispatch_Entry** dispatch_entry;

    if (! iter.next (dispatch_entry))
    {
      return ST_BAD_INTERNAL_POINTER;
    }

    ordered_dispatch_entries_ [i] = *dispatch_entry;
  }

  // sort the entries in order of priority and subpriority
  sort_dispatches (ordered_dispatch_entries_, dispatch_entry_count_);

  // assign dynamic and static subpriorities to the thread dispatch entries
  return assign_subpriorities (ordered_dispatch_entries_,
                               dispatch_entry_count_, anomaly_set);
}
  // dispatch scheduling method: sets up an array of dispatch entries,
  // calls internal dispatch scheduling method.

ACE_DynScheduler::status_t
ACE_DynScheduler::store_assigned_info (void)
{
  for  (u_int i = 0; i < dispatch_entry_count_; ++i)
    {
      if ((! ordered_dispatch_entries_) || (! (ordered_dispatch_entries_[i])) ||
          (! (ordered_dispatch_entries_[i]->task_entry ().rt_info ())))
        {
          ACE_ERROR_RETURN ((LM_ERROR,
                             ACE_LIB_TEXT("ACE_DynScheduler::store_assigned_info () could not store ")
                             ACE_LIB_TEXT("priority information (error in internal representation)")),
                             ST_BAD_INTERNAL_POINTER);
        }

  // set OS priority and Scheduler preemption priority and static
  // preemption subpriority in underlying RT_Info
  ordered_dispatch_entries_ [i]->task_entry ().rt_info ()->priority =
    ordered_dispatch_entries_ [i]->OS_priority ();
  ordered_dispatch_entries_ [i]->task_entry ().rt_info ()->preemption_priority =
    ordered_dispatch_entries_ [i]->priority ();
  ordered_dispatch_entries_ [i]->task_entry ().rt_info ()->preemption_subpriority =
    ordered_dispatch_entries_ [i]->static_subpriority ();
  }

  return SUCCEEDED;
}
  // = store assigned information back into the RT_Infos


ACE_DynScheduler::status_t
ACE_DynScheduler::create_timeline ()
{
  // queue of previously scheduled entries that need to be rescheduled
  ACE_Unbounded_Queue <Dispatch_Entry *> reschedule_queue;

  status_t status = SUCCEEDED;

  ACE_NEW_RETURN(timeline_, ACE_Ordered_MultiSet <TimeLine_Entry_Link>,
                 ST_VIRTUAL_MEMORY_EXHAUSTED);

  ACE_NEW_RETURN(expanded_dispatches_, ACE_Unbounded_Set <Dispatch_Entry *>,
                     ST_VIRTUAL_MEMORY_EXHAUSTED);

  // start with the id of the first entry in the array
  min_dispatch_id_ = ordered_dispatch_entries_[0]->dispatch_id ();
  max_dispatch_id_ = ordered_dispatch_entries_[0]->dispatch_id ();

  for (u_long i = 0; i < dispatch_entry_count_; ++i)
  {
    // update the minimal and maximal id values for the schedule
    if (ordered_dispatch_entries_[i]->dispatch_id () < min_dispatch_id_)
    {
      min_dispatch_id_ = ordered_dispatch_entries_[i]->dispatch_id ();
    }
    if (ordered_dispatch_entries_[i]->dispatch_id () > max_dispatch_id_)
    {
      max_dispatch_id_ = ordered_dispatch_entries_[i]->dispatch_id ();
    }

    // only put OPERATION and REMOTE_DEPENDANT dispatches into the timeline.
    if ((ordered_dispatch_entries_[i]->task_entry().info_type () !=
                RtecScheduler::OPERATION) &&
        (ordered_dispatch_entries_[i]->task_entry().info_type () !=
                RtecScheduler::REMOTE_DEPENDANT))
    {
      continue;
    }

    // schedule the current dispatch entry into the timeline
    status = schedule_timeline_entry (*(ordered_dispatch_entries_[i]),
                                      reschedule_queue);
    if (status != SUCCEEDED)
    {
      break;
    }

    // iterate through the set of dispatch entries that need to be rescheduled
    Dispatch_Entry *rescheduled_entry;
    while (reschedule_queue.is_empty () == 0)
    {

      if (reschedule_queue.dequeue_head (rescheduled_entry) < 0)
      {
        status = ST_BAD_INTERNAL_POINTER;
        break;
      }

      status = schedule_timeline_entry (*rescheduled_entry, reschedule_queue);
      if (status != SUCCEEDED)
      {
        break;
      }
    }
    if (status != SUCCEEDED)
    {
      break;
    }

    // Schedule additional dispatches of the entry
    // over the total frame size into the timeline.
    u_long current_frame_offset = 0;
    u_long task_period =
      ordered_dispatch_entries_[i]->task_entry ().effective_period ();
    for (current_frame_offset = task_period;
         current_frame_offset < frame_size_;
                        current_frame_offset += task_period)
    {
      Dispatch_Entry *new_dispatch_entry;

      // create a new dispatch entry at the current sub-frame offset
      // Just use low 32 bits of arrival and deadline.  This will
      // have to change when TimeBase.idl is finalized.
      const TimeBase::TimeT arrival =
        ordered_dispatch_entries_[i]->arrival () +
        ACE_static_cast (ACE_UINT32, current_frame_offset);
      const TimeBase::TimeT deadline=
        ordered_dispatch_entries_[i]->deadline () +
        ACE_static_cast (ACE_UINT32, current_frame_offset);

      ACE_NEW_RETURN (
        new_dispatch_entry,
        Dispatch_Entry (arrival,
                        deadline,
                        ordered_dispatch_entries_[i]->priority (),
                        ordered_dispatch_entries_[i]->OS_priority (),
                        ordered_dispatch_entries_[i]->task_entry (),
                        ordered_dispatch_entries_[i]),
        ST_VIRTUAL_MEMORY_EXHAUSTED);

      // add the new dispatch entry to the set of expanded dispatches
      expanded_dispatches_->insert (new_dispatch_entry);

      // schedule the new dispatch entry into the timeline
      status = schedule_timeline_entry (*new_dispatch_entry, reschedule_queue);
      if (status != SUCCEEDED)
      {
        break;
      }

      while (reschedule_queue.is_empty () == 0)
      {
        if (reschedule_queue.dequeue_head (rescheduled_entry) < 0)
        {
          status = ST_BAD_INTERNAL_POINTER;
          break;
        }
        status = schedule_timeline_entry (*rescheduled_entry, reschedule_queue);
        if (status != SUCCEEDED)
        {
          break;
        }
      }
      if (status != SUCCEEDED)
      {
        break;
      }
    }

    if (status != SUCCEEDED)
    {
      break;
    }
  }

  return status;
}
  // Create a timeline.



ACE_DynScheduler::status_t
ACE_DynScheduler::output_dispatch_priorities (const char *filename)
{
  status_t status = UNABLE_TO_OPEN_SCHEDULE_FILE;

  // open the file
  FILE *file = ACE_OS::fopen (ACE_TEXT_CHAR_TO_TCHAR(filename), ACE_LIB_TEXT("w"));
  if (file)
  {
    status = output_dispatch_priorities (file);
    fclose (file);
  }
  else
  {
    ACE_ERROR ((LM_ERROR,
                ACE_LIB_TEXT("ACE_DynScheduler::output_dispatch_priorities: ")
                ACE_LIB_TEXT("Could not open schedule file (\"%s\")"),
                ACE_TEXT_CHAR_TO_TCHAR(filename)));
  }

  return status;
}


ACE_DynScheduler::status_t
ACE_DynScheduler::output_dispatch_priorities (FILE *file)
{

  u_long dispatch_count = 0;
  u_long i = 0;
  for (i = 0; i < dispatch_entry_count_; ++i)
  {
    dispatch_count +=
     frame_size_
     / ordered_dispatch_entries_[i]->task_entry ().effective_period ();
  }

  if (ACE_OS::fprintf (
      file, "\n\nSCHEDULING RESULTS:\n\n"
            "Number of dispatches:              %3lu\n"
            "Number of threads:                 %3u\n"
            "Number of tasks:                   %3u\n"
            "Scheduler Status:                    [%d] %s\n"
            "Total Frame Size:                    %lu nsec (%f Hz)\n"
            "Critical Set Frame Size:             %lu nsec (%f Hz)\n"
            "Utilization:                         %f\n"
            "Critical Set Utilization:            %f\n"
            "Minimum Priority Queue:            %3d\n"
            "Minimum Guaranteed Priority Queue: %3d\n"
            "Minimum Critical Priority:         %3d\n\n\n"

            "DISPATCH PRIORITIES:\n\n"
            "                                  (critical              \n"
            "                                   instant)              \n"
            "             dispatch              dynamic      static   \n"
            "operation          ID  priority  subpriority  subpriority\n"
            "---------    --------  --------  -----------  -----------\n",
      dispatch_count, threads_, tasks_, status_,
      status_message(status_), frame_size_,
      (double) (10000000.0 / ((double) frame_size_)),
      critical_set_frame_size_,
      (double) (10000000.0 / ((double) critical_set_frame_size_)),
      utilization_, critical_set_utilization_,
      int(minimum_priority_queue_),
      int(minimum_guaranteed_priority_queue_),
      int(minimum_critical_priority ())) < 0)

  {
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_LIB_TEXT("ACE_DynScheduler::output_dispatch_priorities: ")
                       ACE_LIB_TEXT("Could not write to schedule file\n")),
                      UNABLE_TO_WRITE_SCHEDULE_FILE);
  }

  for (i = 0; i < dispatch_entry_count_; ++i)
  {
    if (ACE_OS::fprintf (file, "%-11s  %8lu  %8u  %11u  %11u\n",
        ordered_dispatch_entries_[i]->task_entry ().rt_info ()->
          entry_point.in (),
        ordered_dispatch_entries_[i]->dispatch_id (),
        ordered_dispatch_entries_[i]->priority (),
        ordered_dispatch_entries_[i]->dynamic_subpriority (),
        ordered_dispatch_entries_[i]->static_subpriority ()) < 0)
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         ACE_LIB_TEXT("ACE_DynScheduler::output_dispatch_priorities: ")
                         ACE_LIB_TEXT("Could not write to schedule file\n")),
                        UNABLE_TO_WRITE_SCHEDULE_FILE);
    }
  }

  return SUCCEEDED;
}


ACE_DynScheduler::status_t
ACE_DynScheduler::output_dispatch_timeline (const char *filename)
{
  status_t status = UNABLE_TO_OPEN_SCHEDULE_FILE;

  // open the file
  FILE *file = ACE_OS::fopen (ACE_TEXT_CHAR_TO_TCHAR(filename), ACE_LIB_TEXT("w"));
  if (file)
  {
    status = output_dispatch_timeline (file);
    fclose (file);
  }
  else
  {
    ACE_ERROR ((LM_ERROR,
                ACE_LIB_TEXT("ACE_DynScheduler::output_dispatch_timeline: ")
                ACE_LIB_TEXT("Could not open schedule file (\"%s\")"),
                filename));
  }

  return status;
}

ACE_DynScheduler::status_t
ACE