Used astyle --style=kr formatted source codes.
authorChao Mei <chaomei2@illinois.edu>
Sat, 7 Apr 2012 06:12:22 +0000 (01:12 -0500)
committerChao Mei <chaomei2@illinois.edu>
Sat, 7 Apr 2012 06:12:22 +0000 (01:12 -0500)
NodeHelper.C
NodeHelper.h
NodeHelperAPI.h

index 62b728a708accfe9696c8f762b94162ab7b21868..16f15b487cdb993f24a46d867a5d5886645e38cd 100644 (file)
@@ -25,91 +25,91 @@ static volatile int exitFlag = 0;
 #include <sys/syscall.h>
 #endif
 
-static int HelperOnCore(){
+static int HelperOnCore() {
 #if CMK_OS_IS_LINUX
-       char fname[64];
-       sprintf(fname, "/proc/%d/task/%ld/stat", getpid(), syscall(SYS_gettid));
-       FILE *ifp = fopen(fname, "r");
-       if(ifp == NULL) return -1;
-       fseek(ifp, 0, SEEK_SET);
-       char str[128];
-       for(int i=0; i<39; i++) fscanf(ifp, "%s", str);
-       fclose(ifp);
-       return atoi(str);
+    char fname[64];
+    sprintf(fname, "/proc/%d/task/%ld/stat", getpid(), syscall(SYS_gettid));
+    FILE *ifp = fopen(fname, "r");
+    if (ifp == NULL) return -1;
+    fseek(ifp, 0, SEEK_SET);
+    char str[128];
+    for (int i=0; i<39; i++) fscanf(ifp, "%s", str);
+    fclose(ifp);
+    return atoi(str);
 #else
-       return -1;
+    return -1;
 #endif
 }
 
 static void *ndhThreadWork(void *id) {
-       size_t myId = (size_t) id;
-
-       //further improvement of this affinity setting!!
-       int myPhyRank = (myId+mainHelperPhyRank)%numPhysicalPEs;
-       //printf("thread[%d]: affixed to rank %d\n", myId, myPhyRank);
-       myPhyRank = myId;
-       CmiSetCPUAffinity(myPhyRank);
-
-       pthread_mutex_init(&thdLock, NULL);
-       pthread_cond_init(&thdCondition, NULL);
-
-       allLocks[myId-1] = &thdLock;
-       allConds[myId-1] = &thdCondition;
-
-       __sync_add_and_fetch(&gCrtCnt, 1);
-
-       while(1){
-               //printf("thread[%ld]: on core %d with main %d\n", myId, HelperOnCore(), mainHelperPhyRank);
-               if(exitFlag) break;
-               pthread_mutex_lock(&thdLock);
-               pthread_cond_wait(&thdCondition, &thdLock);
-               pthread_mutex_unlock(&thdLock);
-               /* kids ID range: [1 ~ numHelpers-1] */
-               if(mainHelper->needTreeBcast()){
-                       //notify my children
-                       int myKid = myId*TREE_BCAST_BRANCH+1;
-                       for(int i=0; i<TREE_BCAST_BRANCH; i++, myKid++){
-                               if(myKid >= mainHelper->getNumHelpers()) break;
-                               //all locks and conditions exclude the main thread, so index needs to be subtracted by one
-                               pthread_mutex_lock(allLocks[myKid-1]);
-                               pthread_cond_signal(allConds[myKid-1]);
-                               pthread_mutex_unlock(allLocks[myKid-1]);
-                       }
-               }
-               pthdLoop->stealWork();
-       }
+    size_t myId = (size_t) id;
+
+    //further improvement of this affinity setting!!
+    int myPhyRank = (myId+mainHelperPhyRank)%numPhysicalPEs;
+    //printf("thread[%d]: affixed to rank %d\n", myId, myPhyRank);
+    myPhyRank = myId;
+    CmiSetCPUAffinity(myPhyRank);
+
+    pthread_mutex_init(&thdLock, NULL);
+    pthread_cond_init(&thdCondition, NULL);
+
+    allLocks[myId-1] = &thdLock;
+    allConds[myId-1] = &thdCondition;
+
+    __sync_add_and_fetch(&gCrtCnt, 1);
+
+    while (1) {
+        //printf("thread[%ld]: on core %d with main %d\n", myId, HelperOnCore(), mainHelperPhyRank);
+        if (exitFlag) break;
+        pthread_mutex_lock(&thdLock);
+        pthread_cond_wait(&thdCondition, &thdLock);
+        pthread_mutex_unlock(&thdLock);
+        /* kids ID range: [1 ~ numHelpers-1] */
+        if (mainHelper->needTreeBcast()) {
+            //notify my children
+            int myKid = myId*TREE_BCAST_BRANCH+1;
+            for (int i=0; i<TREE_BCAST_BRANCH; i++, myKid++) {
+                if (myKid >= mainHelper->getNumHelpers()) break;
+                //all locks and conditions exclude the main thread, so index needs to be subtracted by one
+                pthread_mutex_lock(allLocks[myKid-1]);
+                pthread_cond_signal(allConds[myKid-1]);
+                pthread_mutex_unlock(allLocks[myKid-1]);
+            }
+        }
+        pthdLoop->stealWork();
+    }
 }
 
 void FuncNodeHelper::createPThreads() {
-       int numThreads = numHelpers - 1;
-       allLocks = (pthread_mutex_t **)malloc(sizeof(void *)*numThreads);
-       allConds = (pthread_cond_t **)malloc(sizeof(void *)*numThreads);
-       memset(allLocks, 0, sizeof(void *)*numThreads);
-       memset(allConds, 0, sizeof(void *)*numThreads);
-
-       pthread_attr_t attr;
-       pthread_attr_init(&attr);
-       pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
-       ndhThreads = new pthread_t[numThreads];
-       mainHelperPhyRank = CmiOnCore();
-       numPhysicalPEs = CmiNumCores();
-       if(mainHelperPhyRank == -1) mainHelperPhyRank = 0;
-       for(int i=1; i<=numThreads; i++){
-               pthread_create(ndhThreads+i, &attr, ndhThreadWork, (void *)i);
-       }
-       while(gCrtCnt != numThreads); //wait for all threads to finish creation
+    int numThreads = numHelpers - 1;
+    allLocks = (pthread_mutex_t **)malloc(sizeof(void *)*numThreads);
+    allConds = (pthread_cond_t **)malloc(sizeof(void *)*numThreads);
+    memset(allLocks, 0, sizeof(void *)*numThreads);
+    memset(allConds, 0, sizeof(void *)*numThreads);
+
+    pthread_attr_t attr;
+    pthread_attr_init(&attr);
+    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+    ndhThreads = new pthread_t[numThreads];
+    mainHelperPhyRank = CmiOnCore();
+    numPhysicalPEs = CmiNumCores();
+    if (mainHelperPhyRank == -1) mainHelperPhyRank = 0;
+    for (int i=1; i<=numThreads; i++) {
+        pthread_create(ndhThreads+i, &attr, ndhThreadWork, (void *)i);
+    }
+    while (gCrtCnt != numThreads); //wait for all threads to finish creation
 }
 
-void FuncNodeHelper::exit(){
-       if(mode == NODEHELPER_PTHREAD){
-               exitFlag = 1;
-               for(int i=0; i<numHelpers-1; i++)
-                       pthread_join(ndhThreads[i], NULL);
-               delete [] ndhThreads;
-               free(allLocks);
-               free(allConds);
-               delete pthdLoop;
-       }
+void FuncNodeHelper::exit() {
+    if (mode == NODEHELPER_PTHREAD) {
+        exitFlag = 1;
+        for (int i=0; i<numHelpers-1; i++)
+            pthread_join(ndhThreads[i], NULL);
+        delete [] ndhThreads;
+        free(allLocks);
+        free(allConds);
+        delete pthdLoop;
+    }
 }
 
 /*====End of pthread-related variables and impelementation====*/
@@ -121,8 +121,7 @@ void FuncNodeHelper::exit(){
 
 static FuncNodeHelper *globalNodeHelper = NULL;
 
-FuncNodeHelper::FuncNodeHelper(int mode_, int numThreads_)
-{
+FuncNodeHelper::FuncNodeHelper(int mode_, int numThreads_) {
     traceRegisterUserEvent("nodehelper total work",NDH_TOTAL_WORK_EVENTID);
     traceRegisterUserEvent("nodehlelper finish signal",NDH_FINISH_SIGNAL_EVENTID);
 
@@ -131,7 +130,7 @@ FuncNodeHelper::FuncNodeHelper(int mode_, int numThreads_)
     CmiAssert(globalNodeHelper==NULL);
     globalNodeHelper = this;
 
-    if(mode == NODEHELPER_USECHARM){
+    if (mode == NODEHELPER_USECHARM) {
         //CkPrintf("FuncNodeHelper created on node %d\n", CkMyNode());
         numHelpers = CkMyNodeSize();
         helperPtr = new FuncSingleHelper *[numHelpers];
@@ -143,13 +142,13 @@ FuncNodeHelper::FuncNodeHelper(int mode_, int numThreads_)
             CkChareID helper;
             CProxy_FuncSingleHelper::ckNew(numHelpers, &helper, pestart+i);
         }
-    }else if(mode == NODEHELPER_PTHREAD){
-               helperPtr = NULL;
+    } else if (mode == NODEHELPER_PTHREAD) {
+        helperPtr = NULL;
 
-               numHelpers = numThreads_;
-               useTreeBcast = (numHelpers >= USE_TREE_BROADCAST_THRESHOLD);
-               pthdLoop = new CurLoopInfo(FuncNodeHelper::MAX_CHUNKS);
-               mainHelper = this;
+        numHelpers = numThreads_;
+        useTreeBcast = (numHelpers >= USE_TREE_BROADCAST_THRESHOLD);
+        pthdLoop = new CurLoopInfo(FuncNodeHelper::MAX_CHUNKS);
+        mainHelper = this;
         createPThreads();
     }
 }
@@ -166,13 +165,13 @@ int FuncNodeHelper::MAX_CHUNKS = 64;
 
 #define ALLOW_MULTIPLE_UNSYNC 1
 void FuncNodeHelper::parallelizeFunc(HelperFn func, int paramNum, void * param,
-                                    int numChunks, int lowerRange,
-                                                   int upperRange, int sync,
-                                    void *redResult, REDUCTION_TYPE type) {
+                                     int numChunks, int lowerRange,
+                                     int upperRange, int sync,
+                                     void *redResult, REDUCTION_TYPE type) {
 
     double _start; //may be used for tracing
 
-    if(numChunks > MAX_CHUNKS){
+    if (numChunks > MAX_CHUNKS) {
         CkPrintf("NodeHelper[%d]: WARNING! chunk is set to MAX_CHUNKS=%d\n", CmiMyPe(), MAX_CHUNKS);
         numChunks = MAX_CHUNKS;
     }
@@ -181,101 +180,101 @@ void FuncNodeHelper::parallelizeFunc(HelperFn func, int paramNum, void * param,
      * for chunk indexed at 0 to remainder-1, stride is "unit+1";
      * for chunk indexed at remainder to numChunks-1, stride is "unit"
      */
-     int stride;
-        CurLoopInfo *curLoop = NULL;
+    int stride;
+    CurLoopInfo *curLoop = NULL;
 
     //for using nodequeue
-       TRACE_START(NDH_TOTAL_WORK_EVENTID);
-       if(mode == NODEHELPER_USECHARM){
-               FuncSingleHelper *thisHelper = helperPtr[CkMyRank()];
+    TRACE_START(NDH_TOTAL_WORK_EVENTID);
+    if (mode == NODEHELPER_USECHARM) {
+        FuncSingleHelper *thisHelper = helperPtr[CkMyRank()];
 #if USE_CONVERSE_NOTIFICATION
-       #if ALLOW_MULTIPLE_UNSYNC
-               ConverseNotifyMsg *notifyMsg = thisHelper->getNotifyMsg();
-       #else
-               ConverseNotifyMsg *notifyMsg = thisHelper->notifyMsg;
-       #endif
-               curLoop = (CurLoopInfo *)(notifyMsg->ptr);
-               curLoop->set(numChunks, func, lowerRange, upperRange, paramNum, param);
-               if(useTreeBcast){
-                       int loopTimes = TREE_BCAST_BRANCH>(CmiMyNodeSize()-1)?CmiMyNodeSize()-1:TREE_BCAST_BRANCH;
-                       //just implicit binary tree
-                       int pe = CmiMyRank()+1;
-                       for(int i=0; i<loopTimes; i++, pe++){
-                               if(pe >= CmiMyNodeSize()) pe -= CmiMyNodeSize();
-                               CmiPushPE(pe, (void *)(notifyMsg));
-                       }
-               }else{
-                       for (int i=CmiMyRank()+1; i<numHelpers; i++) {
-                               CmiPushPE(i, (void *)(notifyMsg));
-                       }
-                       for (int i=0; i<CmiMyRank(); i++) {
-                               CmiPushPE(i, (void *)(notifyMsg));
-                       }
-               }
+#if ALLOW_MULTIPLE_UNSYNC
+        ConverseNotifyMsg *notifyMsg = thisHelper->getNotifyMsg();
+#else
+        ConverseNotifyMsg *notifyMsg = thisHelper->notifyMsg;
+#endif
+        curLoop = (CurLoopInfo *)(notifyMsg->ptr);
+        curLoop->set(numChunks, func, lowerRange, upperRange, paramNum, param);
+        if (useTreeBcast) {
+            int loopTimes = TREE_BCAST_BRANCH>(CmiMyNodeSize()-1)?CmiMyNodeSize()-1:TREE_BCAST_BRANCH;
+            //just implicit binary tree
+            int pe = CmiMyRank()+1;
+            for (int i=0; i<loopTimes; i++, pe++) {
+                if (pe >= CmiMyNodeSize()) pe -= CmiMyNodeSize();
+                CmiPushPE(pe, (void *)(notifyMsg));
+            }
+        } else {
+            for (int i=CmiMyRank()+1; i<numHelpers; i++) {
+                CmiPushPE(i, (void *)(notifyMsg));
+            }
+            for (int i=0; i<CmiMyRank(); i++) {
+                CmiPushPE(i, (void *)(notifyMsg));
+            }
+        }
 #else
-    #if ALLOW_MULTIPLE_UNSYNC
+#if ALLOW_MULTIPLE_UNSYNC
         curLoop = thisHelper->getNewTask();
-    #else
+#else
         curLoop = thisHelper->taskBuffer[0];
-    #endif
+#endif
         curLoop->set(numChunks, func, lowerRange, upperRange, paramNum, param);
-       CpvAccess(_qd)->create(numHelpers-1);
-        if(useTreeBcast){
-                       int loopTimes = TREE_BCAST_BRANCH>(CmiMyNodeSize()-1)?CmiMyNodeSize()-1:TREE_BCAST_BRANCH;
-                       //just implicit binary tree
-                       int pe = CmiMyRank()+1;
-                       for(int i=0; i<loopTimes; i++, pe++){
-                               if(pe >= CmiMyNodeSize()) pe -= CmiMyNodeSize();
+        CpvAccess(_qd)->create(numHelpers-1);
+        if (useTreeBcast) {
+            int loopTimes = TREE_BCAST_BRANCH>(CmiMyNodeSize()-1)?CmiMyNodeSize()-1:TREE_BCAST_BRANCH;
+            //just implicit binary tree
+            int pe = CmiMyRank()+1;
+            for (int i=0; i<loopTimes; i++, pe++) {
+                if (pe >= CmiMyNodeSize()) pe -= CmiMyNodeSize();
                 CharmNotifyMsg *one = thisHelper->getNotifyMsg();
                 one->ptr = (void *)curLoop;
                 envelope *env = UsrToEnv(one);
                 env->setObjPtr(thisHelper->ckGetChareID().objPtr);
-                               CmiPushPE(pe, (void *)(env));
-                       }
-               }else{
-                       for (int i=CmiMyRank()+1; i<numHelpers; i++) {
+                CmiPushPE(pe, (void *)(env));
+            }
+        } else {
+            for (int i=CmiMyRank()+1; i<numHelpers; i++) {
                 CharmNotifyMsg *one = thisHelper->getNotifyMsg();
                 one->ptr = (void *)curLoop;
                 envelope *env = UsrToEnv(one);
                 env->setObjPtr(thisHelper->ckGetChareID().objPtr);
                 //printf("[%d] sending a msg %p (env=%p) to [%d]\n", CmiMyRank(), one, env, i);
-                               CmiPushPE(i, (void *)(env));
-                       }
-                       for (int i=0; i<CmiMyRank(); i++) {
+                CmiPushPE(i, (void *)(env));
+            }
+            for (int i=0; i<CmiMyRank(); i++) {
                 CharmNotifyMsg *one = thisHelper->getNotifyMsg();
                 one->ptr = (void *)curLoop;
                 envelope *env = UsrToEnv(one);
                 env->setObjPtr(thisHelper->ckGetChareID().objPtr);
                 //printf("[%d] sending a msg %p (env=%p) to [%d]\n", CmiMyRank(), one, env, i);
-                               CmiPushPE(i, (void *)(env));
-                       }
-               }
+                CmiPushPE(i, (void *)(env));
+            }
+        }
 #endif
-       }else if(mode == NODEHELPER_PTHREAD){
-               int numThreads = numHelpers-1;
-               curLoop = pthdLoop;
-               curLoop->set(numChunks, func, lowerRange, upperRange, paramNum, param);
-               int numNotices = numThreads;
-               if(useTreeBcast){
-                       numNotices = TREE_BCAST_BRANCH>=numThreads?numThreads:TREE_BCAST_BRANCH;
-               }
-               for(int i=0; i<numNotices; i++){
-                       pthread_mutex_lock(allLocks[i]);
-                       pthread_cond_signal(allConds[i]);
-                       pthread_mutex_unlock(allLocks[i]);
-               }
-               //in this mode, it's always synced
-               sync = 1;
-       }
-
-       curLoop->stealWork();
-       TRACE_BRACKET(NDH_TOTAL_WORK_EVENTID);
+    } else if (mode == NODEHELPER_PTHREAD) {
+        int numThreads = numHelpers-1;
+        curLoop = pthdLoop;
+        curLoop->set(numChunks, func, lowerRange, upperRange, paramNum, param);
+        int numNotices = numThreads;
+        if (useTreeBcast) {
+            numNotices = TREE_BCAST_BRANCH>=numThreads?numThreads:TREE_BCAST_BRANCH;
+        }
+        for (int i=0; i<numNotices; i++) {
+            pthread_mutex_lock(allLocks[i]);
+            pthread_cond_signal(allConds[i]);
+            pthread_mutex_unlock(allLocks[i]);
+        }
+        //in this mode, it's always synced
+        sync = 1;
+    }
+
+    curLoop->stealWork();
+    TRACE_BRACKET(NDH_TOTAL_WORK_EVENTID);
 
     //printf("[%d]: parallelize func %p with [%d ~ %d] divided into %d chunks using loop=%p\n", CkMyPe(), func, lowerRange, upperRange, numChunks, curLoop);
 
-       TRACE_START(NDH_FINISH_SIGNAL_EVENTID);
-       curLoop->waitLoopDone(sync);
-       TRACE_BRACKET(NDH_FINISH_SIGNAL_EVENTID);
+    TRACE_START(NDH_FINISH_SIGNAL_EVENTID);
+    curLoop->waitLoopDone(sync);
+    TRACE_BRACKET(NDH_FINISH_SIGNAL_EVENTID);
 
     //printf("[%d]: finished parallelize func %p with [%d ~ %d] divided into %d chunks using loop=%p\n", CkMyPe(), func, lowerRange, upperRange, numChunks, curLoop);
 
@@ -292,35 +291,32 @@ void FuncNodeHelper::parallelizeFunc(HelperFn func, int paramNum, void * param,
 }
 
 void FuncNodeHelper::reduce(void **redBufs, void *redBuf, REDUCTION_TYPE type, int numChunks) {
-    switch(type){
-        case NODEHELPER_INT_SUM:
-        {
-            int result=0;
-            COMPUTE_REDUCTION(int)
-            *((int *)redBuf) = result;
-            break;
-        }
-        case NODEHELPER_FLOAT_SUM:
-        {
-            float result=0;
-            COMPUTE_REDUCTION(float)
-            *((float *)redBuf) = result;
-            break;
-        }
-        case NODEHELPER_DOUBLE_SUM:
-        {
-            double result=0;
-            COMPUTE_REDUCTION(double)
-            *((double *)redBuf) = result;
-            break;
-        }
-        default:
+    switch (type) {
+    case NODEHELPER_INT_SUM: {
+        int result=0;
+        COMPUTE_REDUCTION(int)
+        *((int *)redBuf) = result;
+        break;
+    }
+    case NODEHELPER_FLOAT_SUM: {
+        float result=0;
+        COMPUTE_REDUCTION(float)
+        *((float *)redBuf) = result;
+        break;
+    }
+    case NODEHELPER_DOUBLE_SUM: {
+        double result=0;
+        COMPUTE_REDUCTION(double)
+        *((double *)redBuf) = result;
+        break;
+    }
+    default:
         break;
     }
 }
 
 CpvStaticDeclare(int, NdhStealWorkHandler);
-static void RegisterNodeHelperHdlrs(){
+static void RegisterNodeHelperHdlrs() {
     CpvInitialize(int, NdhStealWorkHandler);
     CpvAccess(NdhStealWorkHandler) = CmiRegisterHandler((CmiHandler)SingleHelperStealWork);
 }
@@ -337,14 +333,14 @@ FuncSingleHelper::FuncSingleHelper(int numHelpers) {
     CmiAssert(globalNodeHelper!=NULL);
     thisNodeHelper = globalNodeHelper;
 
-       nextFreeNotifyMsg = 0;
+    nextFreeNotifyMsg = 0;
 #if USE_CONVERSE_NOTIFICATION
     notifyMsg = (ConverseNotifyMsg *)malloc(sizeof(ConverseNotifyMsg)*notifyMsgBufSize);
-    for(int i=0; i<notifyMsgBufSize; i++){
+    for (int i=0; i<notifyMsgBufSize; i++) {
         ConverseNotifyMsg *tmp = notifyMsg+i;
-        if(thisNodeHelper->useTreeBcast){
+        if (thisNodeHelper->useTreeBcast) {
             tmp->srcRank = CmiMyRank();
-        }else{
+        } else {
             tmp->srcRank = -1;
         }
         tmp->ptr = (void *)(new CurLoopInfo(FuncNodeHelper::MAX_CHUNKS));
@@ -353,12 +349,12 @@ FuncSingleHelper::FuncSingleHelper(int numHelpers) {
 #else
     nextFreeTaskBuffer = 0;
     notifyMsg = (CharmNotifyMsg **)malloc(sizeof(CharmNotifyMsg *)*notifyMsgBufSize);
-    for(int i=0; i<notifyMsgBufSize; i++){
+    for (int i=0; i<notifyMsgBufSize; i++) {
         CharmNotifyMsg *tmp = new(sizeof(int)*8)CharmNotifyMsg; //allow msg priority bits
         notifyMsg[i] = tmp;
-        if(thisNodeHelper->useTreeBcast){
+        if (thisNodeHelper->useTreeBcast) {
             tmp->srcRank = CmiMyRank();
-        }else{
+        } else {
             tmp->srcRank = -1;
         }
         tmp->ptr = NULL;
@@ -370,66 +366,66 @@ FuncSingleHelper::FuncSingleHelper(int numHelpers) {
         //env->setObjPtr has to be called when a notification msg is sent
     }
     taskBuffer = (CurLoopInfo **)malloc(sizeof(CurLoopInfo *)*TASK_BUFFER_SIZE);
-    for(int i=0; i<TASK_BUFFER_SIZE; i++){
+    for (int i=0; i<TASK_BUFFER_SIZE; i++) {
         taskBuffer[i] = new CurLoopInfo(FuncNodeHelper::MAX_CHUNKS);
     }
 #endif
     globalNodeHelper->helperPtr[CkMyRank()] = this;
 }
 
-void FuncSingleHelper::stealWork(CharmNotifyMsg *msg){
+void FuncSingleHelper::stealWork(CharmNotifyMsg *msg) {
 #if !USE_CONVERSE_NOTIFICATION
     int srcRank = msg->srcRank;
     CurLoopInfo *loop = (CurLoopInfo *)msg->ptr;
-    if(srcRank >= 0){
+    if (srcRank >= 0) {
         //means using tree-broadcast to send the notification msg
         int relPE = CmiMyRank()-msg->srcRank;
-               if(relPE<0) relPE += CmiMyNodeSize();
-
-               //CmiPrintf("Rank[%d]: got msg from src %d with relPE %d\n", CmiMyRank(), msg->srcRank, relPE);
-               relPE=relPE*TREE_BCAST_BRANCH+1;
-               for(int i=0; i<TREE_BCAST_BRANCH; i++, relPE++){
-                       if(relPE >= CmiMyNodeSize()) break;
-                       int pe = (relPE + msg->srcRank)%CmiMyNodeSize();
-                       //CmiPrintf("Rank[%d]: send msg to dst %d (relPE: %d) from src %d\n", CmiMyRank(), pe, relPE, msg->srcRank);
+        if (relPE<0) relPE += CmiMyNodeSize();
+
+        //CmiPrintf("Rank[%d]: got msg from src %d with relPE %d\n", CmiMyRank(), msg->srcRank, relPE);
+        relPE=relPE*TREE_BCAST_BRANCH+1;
+        for (int i=0; i<TREE_BCAST_BRANCH; i++, relPE++) {
+            if (relPE >= CmiMyNodeSize()) break;
+            int pe = (relPE + msg->srcRank)%CmiMyNodeSize();
+            //CmiPrintf("Rank[%d]: send msg to dst %d (relPE: %d) from src %d\n", CmiMyRank(), pe, relPE, msg->srcRank);
             CharmNotifyMsg *newone = getNotifyMsg();
             newone->ptr = (void *)loop;
             envelope *env = UsrToEnv(newone);
             env->setObjPtr(thisNodeHelper->helperPtr[pe]->ckGetChareID().objPtr);
-                       CmiPushPE(pe, (void *)env);
-               }
+            CmiPushPE(pe, (void *)env);
+        }
     }
     loop->stealWork();
 #endif
 }
 
-void SingleHelperStealWork(ConverseNotifyMsg *msg){
-       int srcRank = msg->srcRank;
-
-       if(srcRank >= 0){
-               //means using tree-broadcast to send the notification msg
-
-               //int numHelpers = CmiMyNodeSize(); //the value of "numHelpers" should be obtained somewhere else
-               int relPE = CmiMyRank()-msg->srcRank;
-               if(relPE<0) relPE += CmiMyNodeSize();
-
-               //CmiPrintf("Rank[%d]: got msg from src %d with relPE %d\n", CmiMyRank(), msg->srcRank, relPE);
-               relPE=relPE*TREE_BCAST_BRANCH+1;
-               for(int i=0; i<TREE_BCAST_BRANCH; i++, relPE++){
-                       if(relPE >= CmiMyNodeSize()) break;
-                       int pe = (relPE + msg->srcRank)%CmiMyNodeSize();
-                       //CmiPrintf("Rank[%d]: send msg to dst %d (relPE: %d) from src %d\n", CmiMyRank(), pe, relPE, msg->srcRank);
-                       CmiPushPE(pe, (void *)msg);
-               }
-       }
+void SingleHelperStealWork(ConverseNotifyMsg *msg) {
+    int srcRank = msg->srcRank;
+
+    if (srcRank >= 0) {
+        //means using tree-broadcast to send the notification msg
+
+        //int numHelpers = CmiMyNodeSize(); //the value of "numHelpers" should be obtained somewhere else
+        int relPE = CmiMyRank()-msg->srcRank;
+        if (relPE<0) relPE += CmiMyNodeSize();
+
+        //CmiPrintf("Rank[%d]: got msg from src %d with relPE %d\n", CmiMyRank(), msg->srcRank, relPE);
+        relPE=relPE*TREE_BCAST_BRANCH+1;
+        for (int i=0; i<TREE_BCAST_BRANCH; i++, relPE++) {
+            if (relPE >= CmiMyNodeSize()) break;
+            int pe = (relPE + msg->srcRank)%CmiMyNodeSize();
+            //CmiPrintf("Rank[%d]: send msg to dst %d (relPE: %d) from src %d\n", CmiMyRank(), pe, relPE, msg->srcRank);
+            CmiPushPE(pe, (void *)msg);
+        }
+    }
     CurLoopInfo *loop = (CurLoopInfo *)msg->ptr;
     loop->stealWork();
 }
 
-void CurLoopInfo::stealWork(){
+void CurLoopInfo::stealWork() {
     //indicate the current work hasn't been initialized
     //or the old work has finished.
-    if(inited == 0) return;
+    if (inited == 0) return;
 
     int first, last;
     int unit = (upperIndex-lowerIndex+1)/numChunks;
@@ -438,11 +434,11 @@ void CurLoopInfo::stealWork(){
 
     int nextChunkId = getNextChunkIdx();
     int execTimes = 0;
-    while(nextChunkId < numChunks){
-        if(nextChunkId < remainder){
+    while (nextChunkId < numChunks) {
+        if (nextChunkId < remainder) {
             first = (unit+1)*nextChunkId;
             last = first+unit;
-        }else{
+        } else {
             first = (nextChunkId - remainder)*unit + markIdx;
             last = first+unit-1;
         }
@@ -458,39 +454,37 @@ void CurLoopInfo::stealWork(){
 //   End of functions related with FuncSingleHelper                     //
 //======================================================================//
 
-CProxy_FuncNodeHelper NodeHelper_Init(int mode, int numThreads){
-    if(mode == NODEHELPER_USECHARM){
+CProxy_FuncNodeHelper NodeHelper_Init(int mode, int numThreads) {
+    if (mode == NODEHELPER_USECHARM) {
 #if USE_CONVERSE_NOTIFICATION
         CkPrintf("NodeHelperLib is used in SMP with a simple dynamic scheduling (converse-level notification) but not using node-level queue\n");
 #else
         CkPrintf("NodeHelperLib is used in SMP with a simple dynamic scheduling (charm-level notifiation) but not using node-level queue\n");
 #endif
-    }else if(mode==NODEHELPER_PTHREAD){
+    } else if (mode==NODEHELPER_PTHREAD) {
         CkPrintf("NodeHelperLib is used with extra %d pthreads via a simple dynamic scheduling\n", numThreads);
         CmiAssert(numThreads>0);
     }
     return CProxy_FuncNodeHelper::ckNew(mode, numThreads);
 }
 
-void NodeHelper_Exit(CProxy_FuncNodeHelper nodeHelper){
-       nodeHelper.exit();
+void NodeHelper_Exit(CProxy_FuncNodeHelper nodeHelper) {
+    nodeHelper.exit();
 }
 
 void NodeHelper_Parallelize(CProxy_FuncNodeHelper nodeHelper, HelperFn func,
-                        int paramNum, void * param,
-                        int numChunks, int lowerRange, int upperRange,
-                                   int sync,
-                        void *redResult, REDUCTION_TYPE type)
-{
+                            int paramNum, void * param,
+                            int numChunks, int lowerRange, int upperRange,
+                            int sync,
+                            void *redResult, REDUCTION_TYPE type) {
     nodeHelper[CkMyNode()].ckLocalBranch()->parallelizeFunc(func, paramNum, param, numChunks, lowerRange, upperRange, sync, redResult, type);
 }
 
 void NodeHelper_Parallelize(HelperFn func,
-                        int paramNum, void * param,
-                        int numChunks, int lowerRange, int upperRange,
-                                   int sync,
-                        void *redResult, REDUCTION_TYPE type)
-{
+                            int paramNum, void * param,
+                            int numChunks, int lowerRange, int upperRange,
+                            int sync,
+                            void *redResult, REDUCTION_TYPE type) {
     globalNodeHelper->parallelizeFunc(func, paramNum, param, numChunks, lowerRange, upperRange, sync, redResult, type);
 }
 #include "NodeHelper.def.h"
index 15b9449de7c63cf605e7abc6c8889f55e6e00ea0..22706580925d25977da5a0a572b72647c8b07858 100644 (file)
 /* 1. Using converse-level msg, then the msg is always of highest priority.
  * And the notification msg comes from the singlehelper where the loop parallelization
  * is initiated.
- * 
+ *
  * 2. Using charm-level msg, then the msg could be set with different priorities.
  * However, the notification msg comes from the singlehelper where the parallelized
  * loop is executed.
- * 
+ *
  * */
 #define USE_CONVERSE_NOTIFICATION 1
 
 class FuncSingleHelper;
 
-class CurLoopInfo{
+class CurLoopInfo {
     friend class FuncSingleHelper;
-    
+
 private:
     volatile int curChunkIdx;
     int numChunks;
@@ -34,32 +34,31 @@ private:
     void *param;
     //limitation: only allow single variable reduction of size numChunks!!!
     void **redBufs;
-       char *bufSpace;
+    char *bufSpace;
 
     volatile int finishFlag;
-    
+
     //a tag to indicate whether the task for this new loop has been inited
     //this tag is needed to prevent other helpers to run the old task
     int inited;
-    
-public:    
-    CurLoopInfo(int maxChunks):numChunks(0),fnPtr(NULL), lowerIndex(-1), upperIndex(0), 
-    paramNum(0), param(NULL), curChunkIdx(-1), finishFlag(0), redBufs(NULL), bufSpace(NULL), inited(0) 
-       {
-               redBufs = new void *[maxChunks];
-               bufSpace = new char[maxChunks * CACHE_LINE_SIZE];
-        for(int i=0; i<maxChunks; i++) redBufs[i] = (void *)(bufSpace+i*CACHE_LINE_SIZE);
-       }
-    
-    ~CurLoopInfo() { 
-               delete [] redBufs; 
-               delete [] bufSpace;
-       }
-    
-    void set(int nc, HelperFn f, int lIdx, int uIdx, int numParams, void *p){        /*
+
+public:
+    CurLoopInfo(int maxChunks):numChunks(0),fnPtr(NULL), lowerIndex(-1), upperIndex(0),
+            paramNum(0), param(NULL), curChunkIdx(-1), finishFlag(0), redBufs(NULL), bufSpace(NULL), inited(0) {
+        redBufs = new void *[maxChunks];
+        bufSpace = new char[maxChunks * CACHE_LINE_SIZE];
+        for (int i=0; i<maxChunks; i++) redBufs[i] = (void *)(bufSpace+i*CACHE_LINE_SIZE);
+    }
+
+    ~CurLoopInfo() {
+        delete [] redBufs;
+        delete [] bufSpace;
+    }
+
+    void set(int nc, HelperFn f, int lIdx, int uIdx, int numParams, void *p) {        /*
       * WARNING: there's a rare data-racing case here. The current loop is
-      * about to finish (just before setting inited to 0; A helper (say B) 
-      * just enters the stealWork and passes the inited check. The helper 
+      * about to finish (just before setting inited to 0; A helper (say B)
+      * just enters the stealWork and passes the inited check. The helper
       * (say A) is very fast, and starts the next loop, and happens enter
       * into the middle of this function. Then helper B will face corrupted
       * task info as it is trying to execute the old loop task!
@@ -76,37 +75,41 @@ public:
         //needs to be set last
         inited = 1;
     }
-      
-    void waitLoopDone(int sync){
+
+    void waitLoopDone(int sync) {
         //while(!__sync_bool_compare_and_swap(&finishFlag, numChunks, 0));
-       if(sync) while(finishFlag!=numChunks);
-       //finishFlag = 0;
+        if (sync) while (finishFlag!=numChunks);
+        //finishFlag = 0;
         inited = 0;
     }
-    int getNextChunkIdx(){
+    int getNextChunkIdx() {
         return __sync_add_and_fetch(&curChunkIdx, 1);
     }
-    void reportFinished(int counter){
-       if(counter==0) return;
+    void reportFinished(int counter) {
+        if (counter==0) return;
         __sync_add_and_fetch(&finishFlag, counter);
     }
-    
-    int isFree() { return finishFlag == numChunks; }
-    
-       void **getRedBufs() { return redBufs; }
-       
+
+    int isFree() {
+        return finishFlag == numChunks;
+    }
+
+    void **getRedBufs() {
+        return redBufs;
+    }
+
     void stealWork();
 };
 
 /* FuncNodeHelper is a nodegroup object */
 
-typedef struct converseNotifyMsg{
+typedef struct converseNotifyMsg {
     char core[CmiMsgHeaderSizeBytes];
     int srcRank;
     void *ptr;
-}ConverseNotifyMsg;
+} ConverseNotifyMsg;
 
-class CharmNotifyMsg: public CMessage_CharmNotifyMsg{
+class CharmNotifyMsg: public CMessage_CharmNotifyMsg {
 public:
     int srcRank;
     void *ptr; //the loop info
@@ -114,34 +117,38 @@ public:
 
 class FuncNodeHelper : public CBase_FuncNodeHelper {
     friend class FuncSingleHelper;
-       
+
 public:
     static int MAX_CHUNKS;
 private:
     int mode;
-        
-    int numHelpers; //in pthread mode, the counter includes itself    
+
+    int numHelpers; //in pthread mode, the counter includes itself
     FuncSingleHelper **helperPtr; /* ptrs to the FuncSingleHelpers it manages */
-       int useTreeBcast;
-    
+    int useTreeBcast;
+
 public:
-       FuncNodeHelper(int mode_, int numThreads_);
+    FuncNodeHelper(int mode_, int numThreads_);
     ~FuncNodeHelper() {
         delete [] helperPtr;
     }
-       
-       void createPThreads();
-       void exit();
-       
-       int getNumHelpers() { return numHelpers; }
-       int needTreeBcast() { return useTreeBcast; }
-    
+
+    void createPThreads();
+    void exit();
+
+    int getNumHelpers() {
+        return numHelpers;
+    }
+    int needTreeBcast() {
+        return useTreeBcast;
+    }
+
     void parallelizeFunc(HelperFn func, /* the function that finishes a partial work on another thread */
-                        int paramNum, void * param, /* the input parameters for the above func */
-                        int numChunks, /* number of chunks to be partitioned */
-                        int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */                        
-                        int sync=1, /* whether the flow will continue until all chunks have finished */
-                        void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
+                         int paramNum, void * param, /* the input parameters for the above func */
+                         int numChunks, /* number of chunks to be partitioned */
+                         int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */
+                         int sync=1, /* whether the flow will continue until all chunks have finished */
+                         void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
                         );
     void reduce(void **redBufs, void *redBuf, REDUCTION_TYPE type, int numChunks);
 };
@@ -152,18 +159,18 @@ void SingleHelperStealWork(ConverseNotifyMsg *msg);
 //allowing arbitrary combination of sync and unsync parallelizd loops
 #define TASK_BUFFER_SIZE (3)
 class FuncSingleHelper: public CBase_FuncSingleHelper {
-       friend class FuncNodeHelper;
+    friend class FuncNodeHelper;
 private:
     int totalHelpers;
     int notifyMsgBufSize;
-    
+
     FuncNodeHelper *thisNodeHelper;
 #if USE_CONVERSE_NOTIFICATION
     //this msg is shared across all SingleHelpers
     ConverseNotifyMsg *notifyMsg;
 #else
     //acted as a msg buffer for charm-level notification msgs sent to other
-    //SingleHelpers. At each sending, 
+    //SingleHelpers. At each sending,
     //1. the msg destination chare (SingleHelper) has to be set.
     //2. the associated loop info has to be set.
     CharmNotifyMsg **notifyMsg;
@@ -171,58 +178,58 @@ private:
     int nextFreeTaskBuffer;
 #endif
     int nextFreeNotifyMsg;
-    
+
 public:
     FuncSingleHelper(int numHelpers);
 
     ~FuncSingleHelper() {
-    #if USE_CONVERSE_NOTIFICATION
-        for(int i=0; i<notifyMsgBufSize; i++){
+#if USE_CONVERSE_NOTIFICATION
+        for (int i=0; i<notifyMsgBufSize; i++) {
             ConverseNotifyMsg *tmp = notifyMsg+i;
             CurLoopInfo *loop = (CurLoopInfo *)(tmp->ptr);
             delete loop;
         }
         free(notifyMsg);
-    #else
-        for(int i=0; i<notifyMsgBufSize; i++) delete notifyMsg[i];
+#else
+        for (int i=0; i<notifyMsgBufSize; i++) delete notifyMsg[i];
         free(notifyMsg);
-        for(int i=0; i<TASK_BUFFER_SIZE; i++) delete taskBuffer[i];
+        for (int i=0; i<TASK_BUFFER_SIZE; i++) delete taskBuffer[i];
         free(taskBuffer);
-    #endif
+#endif
     }
-#if USE_CONVERSE_NOTIFICATION    
-    ConverseNotifyMsg *getNotifyMsg(){
-        while(1){
+#if USE_CONVERSE_NOTIFICATION
+    ConverseNotifyMsg *getNotifyMsg() {
+        while (1) {
             ConverseNotifyMsg *cur = notifyMsg+nextFreeNotifyMsg;
             CurLoopInfo *loop = (CurLoopInfo *)(cur->ptr);
             nextFreeNotifyMsg = (nextFreeNotifyMsg+1)%notifyMsgBufSize;
-            if(loop->isFree()) return cur;
+            if (loop->isFree()) return cur;
         }
         return NULL;
     }
 #else
-    CharmNotifyMsg *getNotifyMsg(){
-        while(1){
+    CharmNotifyMsg *getNotifyMsg() {
+        while (1) {
             CharmNotifyMsg *cur = notifyMsg[nextFreeNotifyMsg];
             CurLoopInfo *loop = (CurLoopInfo *)(cur->ptr);
             nextFreeNotifyMsg = (nextFreeNotifyMsg+1)%notifyMsgBufSize;
-            if(loop==NULL || loop->isFree()) return cur;
+            if (loop==NULL || loop->isFree()) return cur;
         }
         return NULL;
     }
-    CurLoopInfo *getNewTask(){
-        while(1){
+    CurLoopInfo *getNewTask() {
+        while (1) {
             CurLoopInfo *cur = taskBuffer[nextFreeTaskBuffer];
             nextFreeTaskBuffer = (nextFreeTaskBuffer+1)%TASK_BUFFER_SIZE;
-            if(cur->isFree()) return cur;
+            if (cur->isFree()) return cur;
         }
         return NULL;
     }
-#endif    
-    
+#endif
+
     void stealWork(CharmNotifyMsg *msg);
-    
-    FuncSingleHelper(CkMigrateMessage *m) {}           
+
+    FuncSingleHelper(CkMigrateMessage *m) {}
 };
 
 #endif
index 7808e79d5119867d1129d3cd6239a5571cb5e32a..348f69ea0b4aaccd683605a1c6e50eab5c80d2db 100644 (file)
@@ -6,41 +6,41 @@
 /* "result" is the buffer for reduction result on a single simple-type variable */
 typedef void (*HelperFn)(int first,int last, void *result, int paramNum, void *param);
 
-typedef enum REDUCTION_TYPE{
+typedef enum REDUCTION_TYPE {
     NODEHELPER_NONE=0,
     NODEHELPER_INT_SUM,
     NODEHELPER_FLOAT_SUM,
     NODEHELPER_DOUBLE_SUM
-}REDUCTION_TYPE;
+} REDUCTION_TYPE;
 
 #define NODEHELPER_USECHARM 1
 #define NODEHELPER_PTHREAD 2
 
 class CProxy_FuncNodeHelper;
-/* 
- * The default mode is intended to be used in SMP mode 
+/*
+ * The default mode is intended to be used in SMP mode
  * The next mode that uses pthread is intended to be used in a restricted mode where
  * a node just have one charm PE!
  **/
 extern CProxy_FuncNodeHelper NodeHelper_Init(int mode=NODEHELPER_USECHARM, int numThreads=0);
 
-extern void NodeHelper_Exit(CProxy_FuncNodeHelper nodeHelper); /* used to free resources if using pthread mode. It should be called on just one PE, say PE 0 */                                                                                
+extern void NodeHelper_Exit(CProxy_FuncNodeHelper nodeHelper); /* used to free resources if using pthread mode. It should be called on just one PE, say PE 0 */
 
 extern void NodeHelper_Parallelize(
-                                               CProxy_FuncNodeHelper nodeHelper, /* the proxy to the FuncNodeHelper instance */
-                                               HelperFn func, /* the function that finishes a partial work on another thread */
-                        int paramNum, void * param, /* the input parameters for the above func */
-                        int numChunks, /* number of chunks to be partitioned */
-                        int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */                        
-                        int sync=1, /* whether the flow will continue unless all chunks have finished */ 
-                        void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
-                        );
+    CProxy_FuncNodeHelper nodeHelper, /* the proxy to the FuncNodeHelper instance */
+    HelperFn func, /* the function that finishes a partial work on another thread */
+    int paramNum, void * param, /* the input parameters for the above func */
+    int numChunks, /* number of chunks to be partitioned */
+    int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */
+    int sync=1, /* whether the flow will continue unless all chunks have finished */
+    void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
+);
 extern void NodeHelper_Parallelize(
-                                               HelperFn func, /* the function that finishes a partial work on another thread */
-                        int paramNum, void * param, /* the input parameters for the above func */
-                        int numChunks, /* number of chunks to be partitioned */
-                        int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */                        
-                        int sync=1, /* whether the flow will continue unless all chunks have finished */ 
-                        void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
-                        );
+    HelperFn func, /* the function that finishes a partial work on another thread */
+    int paramNum, void * param, /* the input parameters for the above func */
+    int numChunks, /* number of chunks to be partitioned */
+    int lowerRange, int upperRange, /* the loop-like parallelization happens in [lowerRange, upperRange] */
+    int sync=1, /* whether the flow will continue unless all chunks have finished */
+    void *redResult=NULL, REDUCTION_TYPE type=NODEHELPER_NONE /* the reduction result, ONLY SUPPORT SINGLE VAR of TYPE int/float/double */
+);
 #endif