src/ck-core/modifyScheduler.C

   1 /**
   2     @file
   3     Routines for modifying the Charm++ prioritized message queue
   4     @ingroup CharmScheduler
   5
   6     @addtogroup CharmScheduler
   7     @{
   8  */
   9
  10
  11 #include "charm++.h"
  12 #include "queueing.h" // For access to scheduler data structures
  13 #include "conv-trace.h"
  14
  15
  16 // Predeclarations:
  17 int CqsFindRemoveSpecificPrioq(_prioq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods );
  18 int CqsFindRemoveSpecificDeq(_deq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods );
  19
  20
  21 /** Search Queue for messages associated with a specified entry method */
  22 void CqsIncreasePriorityForEntryMethod(Queue q, const int entrymethod){
  23     void *removedMsgPtr;
  24     int numRemoved;
  25
  26     int entryMethods[1];
  27     entryMethods[0] = entrymethod;
  28
  29     numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, entryMethods, 1 );
  30     if(numRemoved == 0)
  31         numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, entryMethods, 1 );
  32     if(numRemoved == 0)
  33         numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, entryMethods, 1 );
  34
  35     if(numRemoved > 0){
  36         CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
  37         int prio = -1000000;
  38         CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);
  39
  40 #if CMK_TRACE_ENABLED
  41         char traceStr[64];
  42         sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
  43         traceUserSuppliedNote(traceStr);
  44 #endif
  45     }
  46 }
  47
  48 #ifdef ADAPT_SCHED_MEM
  49 /** Search Queue for messages associated with memory-critical entry methods */
  50 void CqsIncreasePriorityForMemCriticalEntries(Queue q){
  51     void *removedMsgPtr;
  52     int numRemoved;
  53
  54     numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
  55     if(numRemoved == 0)
  56         numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
  57     if(numRemoved == 0)
  58         numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
  59
  60     if(numRemoved > 0){
  61         CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
  62         int prio = -1000000;
  63         CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);
  64
  65 #if CMK_TRACE_ENABLED
  66         char traceStr[64];
  67         sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
  68         traceUserSuppliedNote(traceStr);
  69 #endif
  70     }
  71 }
  72 #endif
  73
  74 static bool checkAndRemoveMatching(void *&msgPtr, const int *entryMethod, const int numEntryMethods, envelope *env, void** &p) {
  75         if(env != NULL && (env->getMsgtype() == ForArrayEltMsg ||
  76                        env->getMsgtype() == ForChareMsg)
  77       ){
  78             const int ep = env->getsetArrayEp();
  79             bool foundMatch = false;
  80             // Search for ep by linearly searching through entryMethod
  81             for(int i=0;i<numEntryMethods;++i){
  82                 if(ep==entryMethod[i]){
  83                     foundMatch=true;
  84                     break;
  85                 }
  86             }
  87             if(foundMatch){
  88                 // Remove the entry from the queue
  89                 *p = NULL;
  90                 msgPtr = env;
  91                 return true;
  92             }
  93         }
  94 }
  95
  96 /** Find and remove the first 1 occurences of messages that matches a specified entry method index.
  97     The size of the deq will not change, it will just contain an entry for a NULL pointer.
  98
  99     @return number of entries that were replaced with NULL
 100
 101     @param [in] q A circular double ended queue
 102     @param [out] msgPtr returns the message that was removed from the prioq
 103     @param [in] entryMethod An array of entry method ids that should be considered for removal
 104     @param [in] numEntryMethods The number of the values in the entryMethod array.
 105 */
 106 int CqsFindRemoveSpecificDeq(_deq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods ){
 107     void **iter = q->head; ///< An iterator used to iterate through the circular queue
 108
 109     while(iter != q->tail){
 110         // *iter is contains a pointer to a message
 111         envelope *env = (envelope*)*iter;
 112         if (checkAndRemoveMatching(msgPtr, entryMethod, numEntryMethods, env, iter))
 113           return 1;
 114
 115         // Advance head to the next item in the circular queue
 116         iter++;
 117         if(iter == q->end)
 118             iter = q->bgn;
 119     }
 120     return 0;
 121 }
 122
 123
 124
 125 /** Find and remove the first 1 occurences of messages that matches a specified entry method index.
 126     The size of the prioq will not change, it will just contain an entry for a NULL pointer.
 127
 128     @return number of entries that were replaced with NULL
 129
 130     @param [in] q A priority queue
 131     @param [out] msgPtr returns the message that was removed from the prioq
 132     @param [in] entryMethod An array of entry method ids that should be considered for removal
 133     @param [in] numEntryMethods The number of the values in the entryMethod array.
 134 */
 135 int CqsFindRemoveSpecificPrioq(_prioq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods ){
 136
 137     // A priority queue is a heap of circular queues
 138     for(int i = 1; i < q->heapnext; i++){
 139         // For each of the circular queues:
 140         _prioqelt pe = (q->heap)[i];
 141         void **head; ///< An iterator used to iterate through a circular queue
 142         void **tail; ///< The end of the circular queue
 143         head = pe->data.head;
 144         tail = pe->data.tail;
 145         while(head != tail){
 146             // *head contains a pointer to a message
 147             envelope *env = (envelope*)*head;
 148             if (checkAndRemoveMatching(msgPtr, entryMethod, numEntryMethods, env, head))
 149               return 1;
 150
 151             // Advance head to the next item in the circular queue
 152             head++;
 153             if(head == (pe->data).end)
 154                 head = (pe->data).bgn;
 155         }
 156     }
 157     return 0;
 158 }
 159
 160
 161
 162
 163
 164 /** @} */