Home Research Papers Posters Manuals Talks Download People Help Internal

ck-core/modifyScheduler.C

Go to the documentation of this file.

#include "charm++.h"
#include "queueing.h" // For access to scheduler data structures
#include "conv-trace.h"


// Predeclarations:
int CqsFindRemoveSpecificPrioq(_prioq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods );
int CqsFindRemoveSpecificDeq(_deq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods );


void CqsIncreasePriorityForEntryMethod(Queue q, const int entrymethod){
    void *removedMsgPtr;
    int numRemoved;
    
    int entryMethods[1];
    entryMethods[0] = entrymethod;

    numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, entryMethods, 1 );
    if(numRemoved == 0)
        numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, entryMethods, 1 );
    if(numRemoved == 0)
        numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, entryMethods, 1 );
    
    if(numRemoved > 0){
        CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
        int prio = -1000000; 
        CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);

#ifndef CMK_OPTIMIZE 
        char traceStr[64];
        sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
        traceUserSuppliedNote(traceStr);
#endif
    }
}
 
#ifdef ADAPT_SCHED_MEM

void CqsIncreasePriorityForMemCriticalEntries(Queue q){
    void *removedMsgPtr;
    int numRemoved;

    numRemoved = CqsFindRemoveSpecificPrioq(&(q->negprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
    if(numRemoved == 0)
        numRemoved = CqsFindRemoveSpecificDeq(&(q->zeroprio), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
    if(numRemoved == 0)
        numRemoved = CqsFindRemoveSpecificPrioq(&(q->posprioq), removedMsgPtr, memCriticalEntries, numMemCriticalEntries);
    
    if(numRemoved > 0){
        CkAssert(numRemoved==1); // We need to reenqueue all removed messages, but we currently only handle one
        int prio = -1000000; 
        CqsEnqueueGeneral(q, removedMsgPtr, CQS_QUEUEING_IFIFO, 0, (unsigned int*)&prio);

#ifndef CMK_OPTIMIZE 
        char traceStr[64];
        sprintf(traceStr, "Replacing %p in message queue with NULL", removedMsgPtr);
        traceUserSuppliedNote(traceStr);
#endif
    }
}
#endif

int CqsFindRemoveSpecificDeq(_deq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods ){
    void **iter = q->head; 

    while(iter != q->tail){
        // *iter is contains a pointer to a message
        envelope *env = (envelope*)*iter;
        if(env != NULL && (env->getMsgtype() == ForArrayEltMsg || env->getMsgtype() == ForChareMsg)){
            const int ep = env->getsetArrayEp();
            bool foundMatch = false;
            // Search for ep by linearly searching through entryMethod
            for(int i=0;i<numEntryMethods;++i){
                if(ep==entryMethod[i]){
                    foundMatch=true;
                    break;
                }
            }
            if(foundMatch){
                // Remove the entry from the queue
                *iter = NULL;
                msgPtr = env;
                return 1;
            }
        }
        // Advance head to the next item in the circular queue
        iter++;
        if(iter == q->end)
            iter = q->bgn;
    }
    return 0;
}

 

int CqsFindRemoveSpecificPrioq(_prioq q, void *&msgPtr, const int *entryMethod, const int numEntryMethods ){

    // A priority queue is a heap of circular queues
    for(int i = 1; i < q->heapnext; i++){
        // For each of the circular queues:
        _prioqelt pe = (q->heap)[i];
        void **head; 
        void **tail; 
        head = pe->data.head;
        tail = pe->data.tail;
        while(head != tail){
            // *head contains a pointer to a message
            envelope *env = (envelope*)*head;
            if(env != NULL && (env->getMsgtype() == ForArrayEltMsg || env->getMsgtype() == ForChareMsg)){
                const int ep = env->getsetArrayEp();
                bool foundMatch = false;
                // Search for ep by linearly searching through entryMethod
                for(int i=0;i<numEntryMethods;++i){
                    if(ep==entryMethod[i]){
                        foundMatch=true;
//                      break;
                    }
                }
                if(foundMatch){
                    // Remove the entry from the queue
                    *head = NULL;
                    msgPtr = env;
                    return 1;
                }
            }
            // Advance head to the next item in the circular queue
            head++;
            if(head == (pe->data).end)
                head = (pe->data).bgn;
        }
    }
    return 0;
}

---