strategy_scheduler.cpp

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

// Strategy_Scheduler.cpp,v 1.24 2002/03/24 20:06:12 spark Exp
//
// ============================================================================
//
// = LIBRARY
//    sched
//
// = FILENAME
//    Strategy_Scheduler.cpp
//
// = CREATION DATE
//    22 December 1997
//
// = AUTHOR
//    Chris Gill
//
// ============================================================================

#include "Strategy_Scheduler.h"
#include "ace/Sched_Params.h"

#if ! defined (__ACE_INLINE__)
#include "Strategy_Scheduler.i"
#endif /* __ACE_INLINE__ */

ACE_RCSID(Sched, Strategy_Scheduler, "Strategy_Scheduler.cpp,v 1.24 2002/03/24 20:06:12 spark Exp")

//////////////////////////////////////////////
// Helper function type definition for sort //
//////////////////////////////////////////////

#if defined (ACE_HAS_WINCE)
typedef int (_cdecl* COMP_FUNC) (const void*, const void*);
#else
// This is awkward, but it makes MSVC++ happy.
extern "C"
{
typedef int (*COMP_FUNC) (const void*, const void*);
}
#endif  // ACE_HAS_WINCE

///////////////////////////////////////////////////
// class ACE_Strategy_Scheduler member functions //
///////////////////////////////////////////////////

// = Constructor.

ACE_Strategy_Scheduler::ACE_Strategy_Scheduler (ACE_Scheduler_Strategy &strategy)
  : ACE_DynScheduler (),
    strategy_ (strategy)
{
}


// = Virtual destructor.

ACE_Strategy_Scheduler::~ACE_Strategy_Scheduler ()
{
}


// = Sets up the schedule in the order generated
//   by the strategy's comparison operators.

ACE_DynScheduler::status_t
ACE_Strategy_Scheduler::sort_dispatches (Dispatch_Entry **dispatches,
                                         u_int count)
{
  // Sort the entries in order of priority and subpriority.
  strategy_.sort (dispatches, count);

  return ACE_DynScheduler::SUCCEEDED;
}

// = Assigns priorities and sub-priorities to the sorted schedule,
//   according to the strategy's priority comparison operator.

ACE_DynScheduler::status_t
ACE_Strategy_Scheduler::assign_priorities (
  Dispatch_Entry **dispatches,
  u_int count,
  ACE_Unbounded_Set<RtecScheduler::Scheduling_Anomaly *> &anomaly_set)
{
  // Start with happy status.
  ACE_DynScheduler::status_t status = SUCCEEDED;

  RtecScheduler::Scheduling_Anomaly * anomaly = 0;

  // Start with the highest OS priority in the given range and work downward:
  // if we run out of values to assign, return an error.
  int current_OS_priority = maximum_priority_;

  // Start scheduler priority at 0 (highest priority queue number)
  // NOTE: 0 is highest for priority, lowest for dynamic and static subpriority.
  Preemption_Priority current_scheduler_priority = 0;

  // Value the OS and scheduler priorities in 0th dispatch entry.
  dispatches[0]->OS_priority (current_OS_priority);
  dispatches[0]->priority (current_scheduler_priority);

  // Store the dispatch configuration for the highest priority level.
  Config_Info *config_ptr;
  ACE_NEW_RETURN (config_ptr, Config_Info, ST_VIRTUAL_MEMORY_EXHAUSTED);
  config_ptr->preemption_priority = current_scheduler_priority;
  config_ptr->thread_priority = current_OS_priority;
  config_ptr->dispatching_type = strategy_.dispatch_type (*(dispatches[0]));
  if (config_info_entries_->insert (config_ptr) < 0)
  {
    return ST_VIRTUAL_MEMORY_EXHAUSTED;
  }

  // Traverse ordered dispatch entry array, assigning priority
  // (array is sorted from highest to lowest priority).
  for (u_int i = 1; i < count; ++i)
  {
    switch (strategy_.priority_comp (*(dispatches[i-1]),
                                     *(dispatches[i])))
    {
      case -1:  // The current entry is at lower priority than the previous.
                {
        // Decrease priority by incrementing the current scheduling priority
        // number: 0 is the highest priority number.
        ++current_scheduler_priority;

        // Check for OS priority level boundaries: because OS priority values
        // can run in either increasing or decreasing order, there is no easy,
        // portable way to check other than exact comparison to the bounds
        // that were given in init () or that came from the platform itself.
        if ((current_OS_priority == minimum_priority_) ||
            (current_OS_priority == ACE_Sched_Params::previous_priority (
                                      ACE_SCHED_FIFO,
                                      current_OS_priority,
                                      ACE_SCOPE_PROCESS)))
        {
          // If we have run out of priority levels to assign, indicate
          // this in the return status, unless a more severe problem is
          // already reflected there.  Log an anomaly but keep right on
          // assigning the minimum OS priority in the range to the remaining
          // tasks.
          status = (status == SUCCEEDED)
                   ? ST_INSUFFICIENT_THREAD_PRIORITY_LEVELS
                   : status;

          // Log the anomaly.
          anomaly =
            create_anomaly (ST_INSUFFICIENT_THREAD_PRIORITY_LEVELS);
          if (anomaly)
            {
              anomaly_set.insert (anomaly);
            }
          else
            {
              return ST_VIRTUAL_MEMORY_EXHAUSTED;
            }
        }
        else
        {
          // We're still in range, so decrement the current OS priority level.
          current_OS_priority =
            ACE_Sched_Params::previous_priority (ACE_SCHED_FIFO,
                                                 current_OS_priority,
                                                 ACE_SCOPE_PROCESS);
        }

        // Store the dispatch configuration for the new priority level.
        ACE_NEW_RETURN(config_ptr, Config_Info, ST_VIRTUAL_MEMORY_EXHAUSTED);
        config_ptr->preemption_priority = current_scheduler_priority;
        config_ptr->thread_priority = current_OS_priority;
        config_ptr->dispatching_type = strategy_.dispatch_type (*(dispatches[i]));
        if (config_info_entries_->insert (config_ptr) < 0)
        {
          return ST_VIRTUAL_MEMORY_EXHAUSTED;
        }

                break;
                }
      case 0:  // Still at the same priority level.

        break;

      default: // Should never reach here: something *bad* has happened.

        ACE_ERROR ((
          LM_ERROR,
          "Priority assignment failure: tasks"
          " \"%s\" and \"%s\" are out of order.\n",
          dispatches [i-1]->task_entry ().rt_info ()->entry_point.in (),
          dispatches [i]->task_entry ().rt_info ()->entry_point.in ()));

          status = ACE_DynScheduler::ST_INVALID_PRIORITY_ORDERING;

          // Log the anomaly.
          anomaly =
            create_anomaly (ST_INVALID_PRIORITY_ORDERING);
          if (anomaly)
            {
              anomaly_set.insert (anomaly);
            }
          else
            {
              return ST_VIRTUAL_MEMORY_EXHAUSTED;
            }
    }

    // Set OS priority of the current dispatch entry.
    dispatches[i]->OS_priority (current_OS_priority);

    // Set scheduler priority of the current dispatch entry.
    dispatches[i]->priority (current_scheduler_priority);
  }

  return status;
}


// = Assigns dynamic and static sub-priorities to the sorted dispatch
//   schedule, according to the strategy's subpriority comparisons.

ACE_DynScheduler::status_t
ACE_Strategy_Scheduler::assign_subpriorities (
  Dispatch_Entry **dispatches,
  u_int count,
  ACE_Unbounded_Set<RtecScheduler::Scheduling_Anomaly *> &anomaly_set)
{
  ACE_DynScheduler::status_t status = ACE_DynScheduler::SUCCEEDED;
  RtecScheduler::Scheduling_Anomaly * anomaly = 0;

  // Start subpriority levels and element counts at 1, set level values in
  // the first entry, increment the static subpriority level.
  Sub_Priority dynamic_subpriority_level = 0;
  Sub_Priority static_subpriority_level = 0;
  u_int dynamic_subpriority_elements = 1;
  u_int static_subpriority_elements = 1;
  dispatches [0]->dynamic_subpriority (dynamic_subpriority_level);
  dispatches [0]->static_subpriority (static_subpriority_level);

  // Advance the static subpriority level.
  static_subpriority_level++;

  u_int i,j;
  // Traverse ordered dispatch entry array, assigning priority
  // (array is sorted from highest to lowest priority).
  for (i = 1; i < count; ++i)
  {
    switch (strategy_.priority_comp (*(dispatches [i-1]),
                                     *(dispatches [i])))
    {
      case -1:  // The current entry is at lower priority than the previous.
      {
        // Fill in the high to low dynamic subpriority values by subtracting
        // the previously assigned subpriority value of each of element in the
        // current priority level from the value of last subpriority level.
        for (j = 1; j <= dynamic_subpriority_elements; ++j)
        {
          dispatches [i - j]->
                          dynamic_subpriority (dynamic_subpriority_level -
                                   dispatches [i - j]-> dynamic_subpriority ());
        }
        for (j = 1; j <= static_subpriority_elements; ++j)
        {
          dispatches [i - j]->
                          static_subpriority (static_subpriority_level -
                                  dispatches [i - j]-> static_subpriority () - 1);
        }

        // Reset the subpriority counters, set these values in the
        // current entry, and increment the static subpriority counter.
        dynamic_subpriority_elements = 1;
        static_subpriority_elements = 1;
        dynamic_subpriority_level = 0;
        static_subpriority_level = 0;
        dispatches [i]->dynamic_subpriority (dynamic_subpriority_level);
        dispatches [i]->static_subpriority (static_subpriority_level);

        // Advance the static subpriority level.
        static_subpriority_level++;

        break;
      }

      case 0:  // Still at the same priority level.

        // Compare the dynamic subpriorities.
        switch (strategy_.dynamic_subpriority_comp (*(dispatches[i-1]),
                                                    *(dispatches[i])))
        {
          case -1:  // The current entry is at lower dynamic subpriority.

            // Increment the dynamic subpriority level.
            ++dynamic_subpriority_level;

            // Update the static subpriority as well: this avoids problems
            // with non-determinism if due to run-time conditions, two
            // dispatches line up with identical dynamic subpriority that
            // were considered different with respect to the critical instant.
            dispatches [i]->static_subpriority (static_subpriority_level);
            static_subpriority_level++;
            static_subpriority_elements++;

            break;

          case 0:  // Still at the same dynamic subpriority level.
          {
            switch (strategy_.static_subpriority_comp (*(dispatches[i-1]),
                                                       *(dispatches[i])))
            {
              case -1:
              case  0:

                // Assign and then increment the static subpriority: even if
                // still at the same dynamic or static subpriority level as
                // far as the scheduling strategy is concerned, assign a new
                // one anyway, to give a completely deterministic schedule
                // even if the dynamic subpriorities happen to align due to
                // run-time variation.
                dispatches [i]->static_subpriority (static_subpriority_level);
                static_subpriority_level++;
                static_subpriority_elements++;
                break;

              default: // We should never reach here: something *bad* has happened.

                ACE_ERROR ((
                  LM_ERROR,
                  "Static subpriority assignment failure: tasks"
                  " \"%s\" and \"%s\" are out of order.\n",
                  dispatches [i-1]->task_entry ().rt_info ()->entry_point.in (),
                  dispatches [i]->task_entry ().rt_info ()->entry_point.in ()));

                  status = ST_INVALID_PRIORITY_ORDERING;

                  // Log the anomaly.
                  anomaly =
                    create_anomaly (ST_INVALID_PRIORITY_ORDERING);
                  if (anomaly)
                    {
                      anomaly_set.insert (anomaly);
                    }
                  else
                    {
                      return ST_VIRTUAL_MEMORY_EXHAUSTED;
                    }
            }
            break;
          }

          default: // We should never reach here: something *bad* has happened.

            ACE_ERROR ((
              LM_ERROR,
              "Dynamic subpriority assignment failure: tasks"
              " \"%s\" and \"%s\" are out of order.\n",
              dispatches [i-1]->task_entry ().rt_info ()->entry_point.in (),
              dispatches [i]->task_entry ().rt_info ()->entry_point.in ()));

              status = ACE_DynScheduler::ST_INVALID_PRIORITY_ORDERING;

              // Log the anomaly.
              anomaly =
                create_anomaly (ST_INVALID_PRIORITY_ORDERING);
              if (anomaly)
                {
                  anomaly_set.insert (anomaly);
                }
              else
                {
                  return ST_VIRTUAL_MEMORY_EXHAUSTED;
                }
        }

        dispatches [i]->dynamic_subpriority (dynamic_subpriority_level);
        dynamic_subpriority_elements++;

        break;

      default: // We should never reach here: something *bad* has happened.

        ACE_ERROR ((
          LM_ERROR,
          "Priority assignment failure: tasks"
          " \"%s\" and \"%s\" are out of order.\n",
          dispatches [i-1]->task_entry ().rt_info ()->entry_point.in (),
          dispatches [i]->task_entry ().rt_info ()->entry_point.in ()));

        status = ACE_DynScheduler::ST_INVALID_PRIORITY_ORDERING;

        // Log the anomaly.
        anomaly =
          create_anomaly (ST_INVALID_PRIORITY_ORDERING);
        if (anomaly)
          {
            anomaly_set.insert (anomaly);
          }
        else
          {
            return ST_VIRTUAL_MEMORY_EXHAUSTED;
          }
    }
  }

  // Fill in the high to low subpriority values for the last priority
  // level by subtracting the previously assigned subpriorities from
  // the total number of subpriorities.
  for (j = 1; j <= dynamic_subpriority_elements; ++j)
  {
    dispatches [i - j]->
      dynamic_subpriority (dynamic_subpriority_level -
                           dispatches [i - j]->dynamic_subpriority ());
  }
  for (j = 1; j <= static_subpriority_elements; ++j)
  {
    dispatches [i - j]->
      static_subpriority (static_subpriority_level -
                          dispatches [i - j]->static_subpriority () - 1);
  }

  return status;
}

// = Determine the minimum critical priority number.

ACE_DynScheduler::Preemption_Priority
ACE_Strategy_Scheduler::minimum_critical_priority ()
{
  return strategy_.minimum_critical_priority ();
}


// = Schedules a dispatch entry into the timeline being created.

ACE_DynScheduler::status_t
ACE_Strategy_Scheduler::schedule_timeline_entry (
  Dispatch_Entry &dispatch_entry,
  ACE_Unbounded_Queue <Dispatch_Entry *> &reschedule_queue)
{
  status_t status = SUCCEEDED;

  // Timeline entries cover the execution time of the dispatch.
  Time remaining_time =
    dispatch_entry.task_entry().rt_info ()->worst_case_execution_time;

  // Initialize last stop time to arrival time of the dispatch.
  Time last_stop = dispatch_entry.arrival ();

  TimeLine_Entry *last_entry = 0;
  TimeLine_Entry *current_entry = 0;
  ACE_Ordered_MultiSet_Iterator <TimeLine_Entry_Link> iter (*timeline_);
  for (iter.first (); (remaining_time > 0U) && (iter.done () == 0);
       iter.advance ())
  {
    TimeLine_Entry_Link *link;
    if ((iter.next (link) == 0) || (! link))
    {
      return ST_BAD_INTERNAL_POINTER;
    }

    // For each entry already in the timeline that is the first one for a
    // dispatch, and has lower dynamic subpriority and does not have greater
    // static priority, and starts in the period in which the new entry would
    // execute, advance the iterator to the next timeline entry
    // having a different dispatch entry (if there is such), add its dispatch
    // entry to the reschedule set, remove all TimeLine_Entry_Links that
    // correspond to that dispatch entry, and delete all its TimeLine_Entry
    // objects as well.  NOTE: 0 is highest priority, 1 next, etc.
    while ((iter.done () == 0) &&
           (link->entry ().start() < last_stop + remaining_time) &&
           (link->entry ().start() >= last_stop) &&
           (link->entry ().prev () == 0) &&
           (link->entry ().dispatch_entry().priority () >=
            dispatch_entry.priority ()) &&
           (strategy_.dynamic_subpriority (dispatch_entry, link->entry ().start ()) >
            strategy_.dynamic_subpriority (link->entry ().dispatch_entry (),
                                           link->entry ().start ())))
    {
      // Point to the dispatch entry whose timeline entries will be removed and