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){
#if !CMK_USE_STL_MSGQ
    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);

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

void CqsIncreasePriorityForMemCriticalEntries(Queue q){
#if !CMK_USE_STL_MSGQ
    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);

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

static bool checkAndRemoveMatching(void *&msgPtr, const int *entryMethod, const int numEntryMethods, envelope *env, void** &p) {
    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
        *p = NULL;
        msgPtr = env;
        return true;
        }
    }
    return false;
}

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 (checkAndRemoveMatching(msgPtr, entryMethod, numEntryMethods, env, iter))
      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 (checkAndRemoveMatching(msgPtr, entryMethod, numEntryMethods, env, head))
          return 1;

        // Advance head to the next item in the circular queue
        head++;
            if(head == (pe->data).end)
                head = (pe->data).bgn;
        }
    }
    return 0;
}

---