3 Routines for modifying the Charm++ prioritized message queue
4 @ingroup CharmScheduler
6 @addtogroup CharmScheduler
12 #include "queueing.h" // For access to scheduler data structures
13 #include "conv-trace.h"
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 );
21 /** Search Queue for messages associated with a specified entry method */
22 void CqsIncreasePriorityForEntryMethod(Queue q, const int entrymethod){
27 entryMethods[0] = entrymethod;
29 numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, entryMethods, 1 );
31 numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, entryMethods, 1 );
33 numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, entryMethods, 1 );
36 CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
38 CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);
42 sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
43 traceUserSuppliedNote(traceStr);
48 #ifdef ADAPT_SCHED_MEM
49 /** Search Queue for messages associated with memory-critical entry methods */
50 void CqsIncreasePriorityForMemCriticalEntries(Queue q){
54 numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
56 numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
58 numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
61 CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
63 CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);
67 sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
68 traceUserSuppliedNote(traceStr);
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)
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]){
88 // Remove the entry from the queue
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.
99 @return number of entries that were replaced with NULL
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.
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
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))
115 // Advance head to the next item in the circular queue
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.
128 @return number of entries that were replaced with NULL
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.
135 int CqsFindRemoveSpecificPrioq(_prioq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods ){
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;
146 // *head contains a pointer to a message
147 envelope *env = (envelope*)*head;
148 if (checkAndRemoveMatching(msgPtr, entryMethod, numEntryMethods, env, head))
151 // Advance head to the next item in the circular queue
153 if(head == (pe->data).end)
154 head = (pe->data).bgn;