自学教程：C++ CmiMyPe函数代码示例

int print_thread_affinity() {  unsigned long mask;  size_t len = sizeof(mask);#if  CMK_HAS_PTHREAD_SETAFFINITY  int j;  cpu_set_t cpuset;  pthread_t thread;  char str[256], pe[16];  thread = pthread_self();    if (errno = pthread_getaffinity_np(thread, sizeof(cpu_set_t), &cpuset)) {    perror("pthread_getaffinity");    return -1;  }  sprintf(str, "[%d] %s affinity is: ", CmiMyPe(), CmiMyPe()>=CmiNumPes()?"communication pthread":"pthread");  for (j = 0; j < CPU_SETSIZE; j++)        if (CPU_ISSET(j, &cpuset)) {            sprintf(pe, " %d ", j);            strcat(str, pe);        }  CmiPrintf("%s/n", str);#endif  return 0;}

extern "C" void req_fw_handler(char *msg){  int offset = CmiReservedHeaderSize + sizeof(CcsImplHeader);  CcsImplHeader *hdr = (CcsImplHeader *)(msg+CmiReservedHeaderSize);  int destPE = (int)ChMessageInt(hdr->pe);  if (CmiMyPe() == 0 && destPE == -1) {    /* Broadcast message to all other processors */    int len=CmiReservedHeaderSize+sizeof(CcsImplHeader)+ChMessageInt(hdr->len);    CmiSyncBroadcast(len, msg);  }  else if (destPE < -1) {    /* Multicast the message to your children */    int len=CmiReservedHeaderSize+sizeof(CcsImplHeader)+ChMessageInt(hdr->len)-destPE*sizeof(ChMessageInt_t);    int index, child, i;    int *pes = (int*)(msg+CmiReservedHeaderSize+sizeof(CcsImplHeader));    ChMessageInt_t *pes_nbo = (ChMessageInt_t *)pes;    offset -= destPE * sizeof(ChMessageInt_t);    if (ChMessageInt(pes_nbo[0]) == CmiMyPe()) {      for (index=0; index<-destPE; ++index) pes[index] = ChMessageInt(pes_nbo[index]);    }    for (index=0; index<-destPE; ++index) {      if (pes[index] == CmiMyPe()) break;    }    child = (index << 2) + 1;    for (i=0; i<4; ++i) {      if (child+i < -destPE) {        CmiSyncSend(pes[child+i], len, msg);      }    }  }  CcsHandleRequest(hdr, msg+offset);  CmiFree(msg);}

static void CpdDebugCallAllocationTree(char *msg){  int numNodes;  int forPE;  void *tree;  if (CpdDebugGetAllocationTree == NULL) {    CmiPrintf("Error> Invoked CpdDebugCalloAllocationTree but no function initialized./nDid you forget to link in memory charmdebug?/n");    CcsSendReply(0, NULL);    return;  }  sscanf(msg+CmiMsgHeaderSizeBytes, "%d", &forPE);  if (CmiMyPe() == forPE) CpvAccess(allocationTreeDelayedReply) = CcsDelayReply();  if (forPE == -1 && CmiMyPe()==0) {    CpvAccess(allocationTreeDelayedReply) = CcsDelayReply();    CmiSetXHandler(msg, CpvAccess(CpdDebugCallAllocationTree_Index));    CmiSetHandler(msg, _debugHandlerIdx);    CmiSyncBroadcast(CmiMsgHeaderSizeBytes+strlen(msg+CmiMsgHeaderSizeBytes)+1, msg);  }  tree = CpdDebugGetAllocationTree(&numNodes);  if (forPE == CmiMyPe()) CpdDebugReturnAllocationTree(tree);  else if (forPE == -1) CmiReduceStruct(tree, CpdDebug_pupAllocationPoint, CpdDebug_MergeAllocationTree,                                CpdDebugReturnAllocationTree, CpdDebug_deleteAllocationPoint);  else CmiAbort("Received allocationTree request for another PE!");  CmiFree(msg);}

// open file, if phase is -1, don't add the phase stringFILE* CountLogPool::openFile(int phase) {  // technically, the sprintf into pestr & phasestr are unnecessary,  // can just make a limit and check for that  DEBUGF(("%d CountLogPool::openFile(%d)/n", CmiMyPe(), phase));  const static int strSize = 10;  char pestr[strSize+1];  char phasestr[strSize+1];  snprintf(pestr, strSize, "%d", CmiMyPe());  pestr[strSize] = '/0';  int len = strlen(CpvAccess(_logName)) + strlen("phase.count.") + 2*strSize + 1;  char* fname = new char[len+1];  _MEMCHECK(fname);  if (phase >= 0) {     snprintf(phasestr, strSize, "%d", phase);    phasestr[strSize] = '/0';    sprintf(fname, "%s.phase%s.%s.count", CpvAccess(_logName), phasestr, pestr);   }  else { sprintf(fname, "%s.%s.count", CpvAccess(_logName), pestr); }  FILE* fp = NULL;  DEBUGF(("%d/%d DEBUG: TRACE: %s:%d/n", CmiMyPe(), CmiNumPes(), fname, errno));  do {    fp = fopen(fname, "w+");  } while (!fp && errno == EINTR);  delete[] fname;  if(!fp) {    CmiAbort("Cannot open Summary Trace File for writing.../n");  }  return fp;}

static void CmiNotifyStillIdle(CmiIdleState *s){#if CMK_SHARED_VARS_UNAVAILABLE  /*No comm. thread-- listen on sockets for incoming messages*/  MACHSTATE(1,"idle commserver {")  CommunicationServer(Cmi_idlepoll?0:10, COMM_SERVER_FROM_SMP);  MACHSTATE(1,"} idle commserver")#else#if CMK_SHARED_VARS_POSIX_THREADS_SMP  if(_Cmi_sleepOnIdle ){#endif    int nSpins=20; /*Number of times to spin before sleeping*/    s->nIdles++;    if (s->nIdles>nSpins) { /*Start giving some time back to the OS*/      s->sleepMs+=2;      if (s->sleepMs>10) s->sleepMs=10;    }    /*Comm. thread will listen on sockets-- just sleep*/    if (s->sleepMs>0) {      MACHSTATE1(3,"idle lock(%d) {",CmiMyPe())      CmiIdleLock_sleep(&s->cs->idle,s->sleepMs);      CsdResetPeriodic();		/* check ccd callbacks when I am awakened */      MACHSTATE1(3,"} idle lock(%d)",CmiMyPe())    }#if CMK_SHARED_VARS_POSIX_THREADS_SMP  }

static void setupMessage(Message *msg){    Message *mm;    int     i, size;    int nextSize =  msg->idx;     size = sizeof(Message)+sizes[nextSize].size;     buffer_msgs = (char*)malloc((sizes[nextSize].numiter) * sizeof(Message*));    for(i=0; i<sizes[nextSize].numiter; i++)    {        mm = (Message*)CmiAlloc(size);        mm->srcpe = CmiMyPe();        CmiSetHandler(mm, pva(iterHandler));        //mm->idx = pva(nextSize);        //*((Message**)buffer_msgs+i*sizeof(char*)) = mm;        ((Message**)buffer_msgs)[i] = mm;    }    mm = (Message *) CmiAlloc(size);    mm->srcpe = CmiMyPe();    mm->idx = nextSize;    CmiSetHandler(mm, pva(startHandler));    //fillMessage(mm);//CmiAssert(msg->srcpe == 0);    CmiSyncSendAndFree(msg->srcpe, size, mm);    CmiFree(msg);}

static void startMessage(Message *msg){#if 0    EmptyMsg m;       CmiFree(msg);    CmiSetHandler(&m, pva(iterHandler));    //fillMessage(mm);    pva(starttime) = CmiWallTimer();    CmiSyncSend(CmiMyPe(), sizeof(EmptyMsg), &m);#else   Message *mm;   int size;   size = sizeof(Message)+sizes[pva(nextSize)].size;   mm = (Message *) CmiAlloc(size);   mm->srcpe = CmiMyPe();   mm->idx = pva(nextSize);   CmiSetHandler(mm, pva(iterHandler));   //fillMessage(mm);   pva(starttime) = CmiWallTimer();   CmiSyncSendAndFree(mm->srcpe, size, mm);   CmiFree(msg);#endif  }

static void startNextSize(EmptyMsg *msg){  EmptyMsg m;  Message *mm;  int i; //CmiAssert(CmiMyPe()==0);  pva(nextSize)++;  if(pva(nextSize) == pva(numSizes)) {    pva(nextSize) = -1;    CmiSetHandler(&m, pva(nbrHandler));    CmiSyncSend(CmiMyPe(), sizeof(EmptyMsg), &m);  } else {    int size = sizeof(Message)+sizes[pva(nextSize)].size;        buffer_msgs = (char*)malloc((sizes[pva(nextSize)].numiter) * sizeof(Message*));    for(i=0; i<sizes[pva(nextSize)].numiter; i++)    {        mm = CmiAlloc(size);        mm->srcpe = CmiMyPe();        mm->idx = pva(nextSize);        CmiSetHandler(mm, pva(bounceHandler));        //*((Message**)(buffer_msgs+i*sizeof(char*))) = mm;        ((Message**)buffer_msgs)[i] = mm;    }    mm = (Message *) CmiAlloc(size);    mm->srcpe = CmiMyPe();    mm->idx = pva(nextSize);    CmiSetHandler(mm, pva(setupHandler));    //fillMessage(mm);    CmiSyncSendAndFree(pva(nextNbr), size, mm);  }  CmiFree(msg);}

static void startNextNbr(EmptyMsg *msg){  EmptyMsg m;  TimeMessage *tm;  int i, size;//CmiAssert(CmiMyPe()==0);  CmiFree(msg);  pva(nextNbr)++;  if(pva(nextNbr) == CmiMyNode()) {    CmiSetHandler(&m, pva(nbrHandler));    CmiSyncSend(CmiMyPe(), sizeof(EmptyMsg), &m);    return;  }  if(pva(nextNbr) == CmiNumNodes()) {    pva(nextNbr) = -1;    CmiSetHandler(&m, pva(nodeHandler));    CmiSyncSend(CmiMyPe(), sizeof(EmptyMsg), &m);    size = sizeof(TimeMessage)+pva(numSizes)*CmiNumNodes()*sizeof(double);    tm = (TimeMessage *) CmiAlloc(size);    for(i=0;i<CmiNumNodes();i++)      memcpy(tm->data+i*pva(numSizes),pva(times)[i],             sizeof(double)*pva(numSizes));    tm->srcNode = CmiMyNode();    CmiSetHandler(tm, pva(timeHandler));    CmiSyncSendAndFree(0, size, tm);  } else {    CmiSetHandler(&m, pva(sizeHandler));    CmiSyncSend(CmiMyPe(), sizeof(EmptyMsg), &m);  }}

// constructorStatTable::StatTable():  stats_(NULL), numStats_(0){  DEBUGF(("%d/%d DEBUG: StatTable::StatTable %08x size %d/n",           CmiMyPe(), CmiNumPes(), this));  DEBUGF(("%d/%d registering reductionID/n", CmiMyPe(), CmiNumPes()));  CpvAccess(_reductionID) = CmiRegisterHandler((CmiHandler)StatTableReduction); }

CpmInvokable fibobj_calc(int n, int ppe, fibobj_chare ppos){  if (n<2) Cpm_fibobj_result(CpmSend(ppe), n, ppos);  else {    fibobj_chare c = (fibobj_chare)malloc(sizeof(struct fibobj_chare));    c->ppe = ppe, c->ppos = ppos; c->count=c->total=0;    Cpm_fibobj_calc(CpmLDB(), n-1, CmiMyPe(), c);    Cpm_fibobj_calc(CpmLDB(), n-2, CmiMyPe(), c);  }}

CmiStartFn mymain(int argc, char** argv){    if(CmiMyRank() == CmiMyNodeSize()) return 0;    CpvInitialize(int,msgSize);    CpvInitialize(int,cycleNum);    CpvInitialize(int,sizeNum);    CpvAccess(sizeNum) = 1;    CpvAccess(msgSize)= CmiMsgHeaderSizeBytes + 8;        CpvInitialize(int,exitHandler);    CpvAccess(exitHandler) = CmiRegisterHandler((CmiHandler) exitHandlerFunc);    CpvInitialize(int,node0Handler);    CpvAccess(node0Handler) = CmiRegisterHandler((CmiHandler) node0HandlerFunc);    CpvInitialize(int,node1Handler);    CpvAccess(node1Handler) = CmiRegisterHandler((CmiHandler) node1HandlerFunc);    CpvInitialize(int,ackHandler);    CpvAccess(ackHandler) = CmiRegisterHandler((CmiHandler) ackHandlerFunc);        CpvInitialize(double,startTime);    CpvInitialize(double,endTime);        CpvInitialize(double, IdleStartTime);    CpvInitialize(double, IdleTime);    CpvInitialize(int,ackCount);    CpvAccess(ackCount) = 0;    CpvInitialize(int,twoway);    CpvAccess(twoway) = 0;    CcdCallOnConditionKeep(CcdPROCESSOR_BEGIN_IDLE, ApplIdleStart, NULL);    CcdCallOnConditionKeep(CcdPROCESSOR_END_IDLE, ApplIdleEnd, NULL);        if(argc > 1)        CpvAccess(twoway) = atoi(argv[1]);    if(CmiMyPe() == 0) {      if(!CpvAccess(twoway))        CmiPrintf("Starting Pingpong with oneway traffic /n");      else        CmiPrintf("Starting Pingpong with twoway traffic/n");    }    if ((CmiMyPe() < CmiNumPes()/2) || CpvAccess(twoway))      startPingpong();    return 0;}

//! add the argument parameters to the linked list of args choicesvoid TraceCounter::registerArg(CounterArg* arg){  if (firstArg_ == NULL) {    firstArg_ = lastArg_ = arg;    argStrSize_ = strlen(arg->arg);  }  else {     // check to see if any redundancy     CounterArg* check = firstArg_;    while (check != NULL) {      if (strcmp(check->arg, arg->arg)==0 || check->code == arg->code) {	if (CmiMyPe()==0) { 	  CmiPrintf("Two args with same name %s or code %d/n", 		    arg->arg, arg->code); 	}	CmiAbort("TraceCounter::registerArg()/n");      }      check = check->next;    }    lastArg_->next = arg;    lastArg_ = arg;    int len = strlen(arg->arg);    if (len > argStrSize_) { argStrSize_ = len; }  }}

/* Functions regarding broadcat op that sends to every one else except me */void CmiSyncBroadcastFn1(int size, char *msg) {    int i, mype;    CQdCreate(CpvAccess(cQdState), CmiNumPes()-1);    /*record the rank to avoid re-sending the msg in  spanning tree or hypercube*/    CMI_DEST_RANK(msg) = CmiMyRank();#if CMK_BROADCAST_SPANNING_TREE    CMI_SET_BROADCAST_ROOT(msg, CmiMyNode()+1);    SendSpanningChildrenProc(size, msg);#elif CMK_BROADCAST_HYPERCUBE    CMI_SET_BROADCAST_ROOT(msg, CmiMyNode()+1);    SendHyperCubeProc(size, msg);#else    mype = CmiMyPe();    #if CMK_SMP    /* In SMP, this function may be called from comm thread with a larger pe */    if(mype >= _Cmi_numpes){	for(i=0; i<_Cmi_numpes; i++)		CmiSyncSendFn(i, size, msg);	return;    }    #endif	    for ( i=mype+1; i<_Cmi_numpes; i++ )        CmiSyncSendFn(i, size, msg) ;	    for ( i=0; i<mype; i++ )        CmiSyncSendFn(i, size, msg) ;#endif    /*CmiPrintf("In  SyncBroadcast broadcast/n");*/}

// called on every processorvoid CentralLB::SendStats(){#if CMK_LBDB_ON  CmiAssert(statsMsg != NULL);  reduction_started = 0;#if USE_LDB_SPANNING_TREE  if(CkNumPes()>1024)  {    if (CkMyPe() == cur_ld_balancer)      thisProxy[CkMyPe()].ReceiveStats(statsMsg);    else      thisProxy[CkMyPe()].ReceiveStatsViaTree(statsMsg);  }  else#endif  {    DEBUGF(("[%d] calling ReceiveStats on step %d /n",CmiMyPe(),step()));    thisProxy[cur_ld_balancer].ReceiveStats(statsMsg);  }  statsMsg = NULL;#ifdef __BIGSIM__  BgEndStreaming();#endif  {  // enfore the barrier to wait until centralLB says no  LDOMHandle h;  h.id.id.idx = 0;  theLbdb->getLBDB()->RegisteringObjects(h);  }#endif}

CountLogPool::CountLogPool():  stats_     (),  lastPhase_ (-1){  DEBUGF(("%d/%d DEBUG: CountLogPool::CountLogPool() %08x/n",           CmiMyPe(), CmiNumPes(), this));}

CmiCommHandle CmiAsyncBroadcastFn(int size, char *msg) {#if ENSURE_MSG_PAIRORDER    /* Not sure how to add the msg seq no for async broadcast messages --Chao Mei */    /* so abort here ! */    CmiAssert(0);    return 0;#else    int i, rank;    int mype = CmiMyPe();#if ENABLE_CONVERSE_QD    CQdCreate(CpvAccess(cQdState), CmiNumPes()-1);#endif    MACHSTATE1(3,"[%d] Sending async broadcast message from {",CmiMyNode());    CMI_BROADCAST_ROOT(msg) = 0;    void *handle = malloc(sizeof(int));    *((int *)handle) = CmiNumPes()-1;    for (i=mype+1; i<CmiNumPes(); i++) {        CMI_DEST_RANK(msg) = CmiRankOf(i);        lapiSendFn(CmiNodeOf(i), size, msg, DeliveredMsg, handle);    }    for (i=0; i<mype; i++) {        CMI_DEST_RANK(msg) = CmiRankOf(i);        lapiSendFn(CmiNodeOf(i), size, msg, DeliveredMsg, handle);    }    MACHSTATE(3,"} Sending async broadcast message end");    return handle;#endif}

static void *meta_valloc(size_t size){  void *ret=mm_valloc(size);  if (memInit) CmiPrintf("CMI_MEMORY(%d)> valloc(%d) => %p/n",			 CmiMyPe(),size,ret);    return ret;}

static int getNUMANidByRank(int coreid) {    int i;    /*int totalCores = CmiNumCores();*/    int totalNUMANodes = CmiNumNUMANodes();    /*The core id array is viewed as 2D array but in form of 1D array*/    /*int *coreids=(int *)malloc(sizeof(int)*totalCores);*/    /*Assume each NUMA node has the same number of cores*/    /*int nCoresPerNode = totalCores/totalNUMANodes;*/    /*CmiAssert(totalCores%totalNUMANodes==0);*/    char command[256];    for (i=0; i<totalNUMANodes; i++) {        FILE *cmd;        int cpuid;        sprintf(command, "ls -1d /sys/devices/system/node/node%d/cpu[0-9]* | cut -d'u' -f2", i);        cmd = popen(command, "r");        while (1) {            int ret=fscanf(cmd, "%d/n", &cpuid);            if (ret==EOF) break;            if (cpuid == coreid) {                pclose(cmd);                /*free(coreids);*/                return i;            }        }        pclose(cmd);    }    /*free(coreids);*/    CmiPrintf("%d: the corresponding NUMA node for cpu id %d is not found!/n", CmiMyPe(), coreid);    CmiAssert(0);    return -1;}

static CmiCommHandle MachineSendFuncForLAPI(int destNode, int size, char *msg, int mode) {    scompl_hndlr_t *shdlr = NULL;    void *sinfo = NULL;    if (mode==P2P_SYNC) {        shdlr = ReleaseMsg;        sinfo = (void *)msg;    } else if (mode==P2P_ASYNC) {        shdlr = DeliveredMsg;        sinfo = malloc(sizeof(int));        *((int *)sinfo) = 1;    }    CMI_MSG_SIZE(msg) = size;#if ENSURE_MSG_PAIRORDER#if CMK_NODE_QUEUE_AVAILABLE    if (CMI_DEST_RANK(msg) == DGRAM_NODEMESSAGE) {        lapiSendFn(destNode, size, msg, shdlr, sinfo);        return sinfo;    }#endif    int destPE = CmiNodeFirst(destNode)+CMI_DEST_RANK(msg);    CMI_MSG_SRCPE(msg) = CmiMyPe();    /* Note: This could be executed on comm threads, where CmiMyPe() >= CmiNumPes() */    CMI_MSG_SEQNO(msg) = getNextMsgSeqNo(CpvAccess(p2pMsgSeqInfo).nextMsgSeqNo, destPE);    setNextMsgSeqNo(CpvAccess(p2pMsgSeqInfo).nextMsgSeqNo, destPE, CMI_MSG_SEQNO(msg));#endif    lapiSendFn(destNode, size, msg, shdlr, sinfo);    return sinfo;}

void BTest::sendall() {  //ComlibManager* myman = CProxy_ComlibManager(cinst1.getComlibManagerID()).ckLocalBranch();  cinst8.beginIteration();  //totalTime = 0;  if (thisIndex==0) CkPrintf("[%2d] Broadcast with length %d/n",CmiMyPe(),lengths[length]);  send();}

static void meta_free(void *mem){  if (memInit) CmiPrintf("CMI_MEMORY(%d)> free(%p)/n",			 CmiMyPe(),mem);  if (memInit>1) {int memBack=memInit; memInit=0; CmiPrintStackTrace(0); memInit=memBack;}  mm_free(mem);}

CpmInvokable vars_set_cpv_and_csv(vars_chare c){  CpvAccess(cpv1) = CmiMyPe();  if (CmiMyRank() == 0)    CsvAccess(csv1) = 0x12345678;  Cpm_vars_ack(CpmSend(0), c);}