Charm++: Charm Source Code Documentation

00001 /*************************************************************
00002  * File: ampiOneSided.C
00003  *       This file contains one-sided communication functions
00004  *       win_obj class definition and AMPI_* implementations.
00005  *************************************************************/
00006 
00007 #include "ampiimpl.h"
00008 
00009 /*************************************************************
00010  * Local flags used for Win_obj class: 
00011  *     WIN_ERROR -- the operation fails
00012  *     WIN_SUCCESS -- the operation succeeds
00013  *************************************************************/
00014 #define WIN_SUCCESS     0
00015 #define WIN_ERROR       (-1)
00016 
00017 win_obj::win_obj() {
00018   winName = NULL;
00019   winNameLeng = 0;
00020   baseAddr = NULL;
00021   comm = MPI_COMM_NULL; 
00022   initflag = 0;
00023 }
00024 
00025 win_obj::win_obj(char *name, void *base, MPI_Aint size, int disp_unit, 
00026                          MPI_Comm comm) {
00027   create(name, base, size, disp_unit, comm);
00028   //ampi *ptr = getAmpiInstance(comm);
00029   owner = -1;  // the lock is not owned by anyone yet
00030 }
00031 
00032 void 
00033 win_obj::setName(const char *src,int len) {
00034   if(winName==NULL) winName=new char[MPI_MAX_OBJECT_NAME];
00035   winNameLeng = len;
00036   memcpy(winName,src,len);
00037   winName[len] = '\0';
00038 }
00039 
00040 void 
00041 win_obj::getName(char *name, int *len) {
00042   if(winName==NULL){
00043     name = NULL;
00044     *len = 0;
00045     return;
00046   }
00047   *len = winNameLeng; 
00048   memcpy(name, winName, *len+1);
00049 }
00050 
00051 win_obj::~win_obj() {
00052   free();  
00053 }
00054 
00055 // Note that this is supposed to be used for migration. 
00056 // We should not hava a remote methos which has to pack the win data --- Inefficient
00057 void
00058 win_obj::pup(PUP::er &p) {
00059 #if 0
00060   p|winSize;
00061   p|disp_unit;
00062   p|comm;
00063   p|initflag; 
00064   
00065   int len = 0;
00066   if(winName) len = strlen(winName)+1;
00067   p|len;
00068   if(p.isUnpacking()) winName = new char[len+1];
00069   p(winName, len);
00070 
00071   int size = 0;
00072   if(baseAddr) size = winSize;
00073   p|size;
00074   if(p.isUnpacking()) baseAddr = new char[size+1];
00075   p(baseAddr, size);
00076 #endif
00077 }
00078 
00079 int win_obj::create(char *name, void *base, MPI_Aint size, int disp_unit, MPI_Comm comm){
00080   winName = NULL;
00081   baseAddr = base;
00082   winSize = size*disp_unit;
00083   this->disp_unit = disp_unit;
00084   this->comm = comm;
00085   // assume : memory pointed by base has been allocated 
00086   initflag = 1;
00087   return WIN_SUCCESS;
00088 }
00089 
00090 int win_obj::free(){
00091   delete[] winName;
00092   winName = NULL;
00093   // Assume : memory will be deallocated by user
00094   initflag = 0;
00095   return WIN_SUCCESS;
00096 }
00097 
00098 // This is a local function. 
00099 // AMPI_Win_put will act as a wrapper: pack the input parameters, copy the 
00100 //   remote data to local, and call this function of the involved WIN object
00101 int win_obj::put(void *orgaddr, int orgcnt, int orgunit, 
00102                  MPI_Aint targdisp, int targcnt, int targunit) {
00103   if(initflag == 0) {
00104     CkAbort("Put to non-existing MPI_Win\n");
00105     return WIN_ERROR;
00106   }
00107   int totalsize = targdisp+targcnt*targunit;
00108   if(totalsize > (winSize)){
00109     CkAbort("Put size exceeds MPI_Win size\n");
00110     return WIN_ERROR;
00111   }
00112   
00113   return WIN_SUCCESS;
00114 }
00115 
00116 int win_obj::get(void *orgaddr, int orgcnt, int orgunit, 
00117                  MPI_Aint targdisp, int targcnt, int targunit){
00118   if(initflag == 0) {
00119     CkAbort("Get from non-existing MPI_Win\n");
00120     return WIN_ERROR;
00121   }
00122   int totalsize = targdisp+targcnt*targunit;
00123   if(totalsize > (winSize)){
00124     CkAbort("Get size exceeds MPI_Win size\n");
00125     return WIN_ERROR;
00126   }
00127   // Call the RMA operation here!!!     
00128 
00129   return WIN_SUCCESS;
00130 }
00131 
00132 int win_obj::iget(int orgcnt, int orgunit,
00133                  MPI_Aint targdisp, int targcnt, int targunit){
00134   if(initflag == 0) {
00135     CkAbort("Get from non-existing MPI_Win\n");
00136     return WIN_ERROR;
00137   }
00138 
00139   if((targdisp+targcnt*targunit) > (winSize)){
00140     CkAbort("Get size exceeds MPI_Win size\n");
00141     return WIN_ERROR;
00142   }
00143   // Call the RMA operation here!!!
00144 
00145   return WIN_SUCCESS;
00146 }
00147 
00148 int 
00149 win_obj::accumulate(void *orgaddr, int orgcnt, MPI_Datatype orgtype, 
00150                     MPI_Aint targdisp, int targcnt, 
00151                     MPI_Datatype targtype, MPI_Op op){
00152   applyOp(targtype, op, targcnt, (void*)((int*)baseAddr+targdisp) , (void*)orgaddr);
00153   return WIN_SUCCESS;
00154 }
00155 
00156 int win_obj::fence(){
00157   return WIN_SUCCESS;
00158 }
00159 
00160 int win_obj::lock(int requestRank, int pe_src, int ftHandle, int lock_type){
00161   owner = requestRank;
00162 
00163   int tmp = 0;
00164   AmpiMsg *msg = new (tmp, 0) AmpiMsg(-1, -1, -1, 0, 0, comm);
00165   CkSendToFutureID(ftHandle, (void *)msg, pe_src);
00166   
00167   return WIN_SUCCESS;
00168 }
00169 
00170 
00171 int win_obj::unlock(int requestRank, int pe_src, int ftHandle){
00172   if (owner != requestRank){
00173     CkPrintf("    ERROR: Can't unlock a lock which you don't own.\n");
00174     return WIN_ERROR;
00175   }  
00176   owner = -1;
00177 
00178   // dequeue from queue itself
00179   dequeue();
00180  
00181   int tmp = 0;
00182   AmpiMsg *msg = new (tmp, 0) AmpiMsg(-1, -1, -1, 0, 0, comm);
00183   CkSendToFutureID(ftHandle, (void *)msg, pe_src); 
00184   
00185   return WIN_SUCCESS;
00186 }
00187 
00188 void win_obj::dequeue() {
00189   lockQueueEntry *lq = lockQueue.deq();
00190   delete lq;
00191 }
00192 
00193 void win_obj::enqueue(int requestRank, int pe_src, int ftHandle, int lock_type) {
00194   lockQueueEntry *lq = new lockQueueEntry(requestRank, pe_src, ftHandle, lock_type);
00195   lockQueue.enq(lq);
00196 }
00197 
00198 bool win_obj::emptyQueue() {
00199   return (lockQueue.length()==0) ; 
00200 }
00201 
00202 void win_obj::lockTopQueue() {
00203   lockQueueEntry *lq = lockQueue.deq();
00204   lock(lq->requestRank, lq->pe_src, lq->ftHandle, lq->lock_type);
00205   lockQueue.insert(0, lq);
00206 }
00207 
00208 /* these four functions are yet to implement */
00209 int win_obj::wait(){
00210   return -1;
00211 }
00212 int win_obj::post(){
00213   return -1;
00214 }
00215 int win_obj::start(){
00216   return -1;
00217 }
00218 int win_obj::complete(){
00219   return -1;
00220 }
00221 
00222 int ampiParent::addWinStruct(WinStruct* win) { 
00223         winStructList.push_back(win);
00224         return winStructList.size()-1;
00225 }
00226 
00227 WinStruct ampiParent::getWinStruct(MPI_Win win) {
00228         return *(winStructList[(int)win]);
00229 }
00230    
00231 void ampiParent::removeWinStruct(WinStruct win) {/*winStructList.remove(win);*/}
00232 
00233 int
00234 ampi::winPut(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank, 
00235              MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, WinStruct win){
00236   // Create a Future object 
00237   AmpiMsg *msg = new AmpiMsg();
00238   CkFutureID ftHandle = CkCreateAttachedFuture((void*)msg);
00239   AMPI_DEBUG("    Created Future with handle %d\n", ftHandle);
00240   
00241   CProxy_ampi pa(thisArrayID);
00242 
00243   CkDDT_DataType *ddt = getDDT()->getType(orgtype);
00244   int orgtotalsize = ddt->getSize(orgcnt);
00245   char* sorgaddr = (char*)new char[orgtotalsize]; 
00246   ddt->serialize((char*)orgaddr, (char*)sorgaddr, orgcnt, 1);
00247    
00248   pa[rank].winRemotePut(orgtotalsize, (char*)sorgaddr, orgcnt, orgtype, targdisp, targcnt, targtype, 
00249                         win.index, ftHandle, CkMyPe());
00250 
00251   // Wait on the Future object 
00252   AMPI_DEBUG("    Future [%d] waiting\n", ftHandle);
00253   msg = (AmpiMsg*)CkWaitFutureID(ftHandle);
00254   AMPI_DEBUG("    Future [%d] awaken\n", ftHandle);
00255   
00256   delete [] sorgaddr;
00257   return MPI_SUCCESS;
00258 }
00259 
00260 void
00261 ampi::winRemotePut(int orgtotalsize, char* sorgaddr, int orgcnt, MPI_Datatype orgtype, 
00262                    MPI_Aint targdisp, int targcnt, MPI_Datatype targtype,
00263                    int winIndex, CkFutureID ftHandle, int pe_src) {
00264   win_obj *winobj = winObjects[winIndex];
00265   int orgunit, targunit;
00266   CkDDT_DataType *oddt = getDDT()->getType(orgtype); 
00267   CkDDT_DataType *tddt = getDDT()->getType(targtype);
00268   orgunit = oddt->getSize(1);
00269   targunit = tddt->getSize(1);
00270   winobj->put(sorgaddr, orgcnt, orgunit, targdisp, targcnt, targunit);
00271   char* targaddr = ((char*)(winobj->baseAddr)) + tddt->getSize(targdisp);
00272   tddt->serialize(targaddr, (char*)sorgaddr, targcnt, (-1));
00273 
00274   int tmp = 0;
00275   AmpiMsg *msg = new (tmp, 0) AmpiMsg(-1, -1, -1, thisIndex, 0,myComm.getComm());
00276   AMPI_DEBUG("    Rank[%d] Send to Future [%d] \n", thisIndex, ftHandle);
00277   CkSendToFutureID(ftHandle, (void *)msg, pe_src);
00278 }
00279 
00280 int 
00281 ampi::winGet(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank, 
00282              MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, WinStruct win){
00283   // Create a Future object 
00284   AmpiMsg *msg = new AmpiMsg();
00285   CkFutureID ftHandle = CkCreateAttachedFuture((void*)msg);
00286   AMPI_DEBUG("    Created Future with handle %d\n", ftHandle);
00287 
00288   // Send the request to data and handle of Future to remote side 
00289   CProxy_ampi pa(thisArrayID);
00290   AMPI_DEBUG("    Rank[%d:%d] invoke Remote get at [%d]\n", thisIndex, myRank, rank);
00291   CkDDT_DataType *ddt = getDDT()->getType(orgtype);
00292   int orgtotalsize = ddt->getSize(orgcnt);
00293   char* sorgaddr = (char*)new char[orgtotalsize];
00294 
00295   pa[rank].winRemoteGet(orgcnt, orgtype, targdisp, targcnt, targtype, win.index, ftHandle, CkMyPe());
00296 
00297   // Wait on the Future object 
00298   AMPI_DEBUG("    Future [%d] waiting\n", ftHandle);
00299   msg = (AmpiMsg*)CkWaitFutureID(ftHandle);
00300   AMPI_DEBUG("    Future [%d] awaken\n", ftHandle);
00301   
00302   // Process the reply message and copy the data into desired memory position 
00303   memcpy(sorgaddr, msg->data, orgtotalsize);  
00304   ddt->serialize((char*)orgaddr, (char*)sorgaddr, orgcnt, (-1));
00305   AMPI_DEBUG("    Rank[%d] got win  [%d] \n", thisIndex, *(int*)msg->data);
00306   AMPI_DEBUG("    Rank[%d] got win  [%d] , size %d\n", thisIndex, *(int*)orgaddr, orgcnt);
00307   delete [] sorgaddr;
00308   return MPI_SUCCESS;
00309 }
00310 
00311 void
00312 ampi::winRemoteGet(int orgcnt, MPI_Datatype orgtype, MPI_Aint targdisp, int targcnt, 
00313                    MPI_Datatype targtype, int winIndex, CkFutureID ftHandle, 
00314                    int pe_src) {
00315   AMPI_DEBUG("    RemoteGet invoked at Rank[%d:%d]\n", thisIndex, myRank);  
00316 
00317   int orgunit, targunit;
00318   CkDDT_DataType *oddt = getDDT()->getType(orgtype);
00319   CkDDT_DataType *tddt = getDDT()->getType(targtype);
00320   orgunit = oddt->getSize(1);
00321   targunit = tddt->getSize(1);
00322   int stargtotalsize = tddt->getSize(targcnt);
00323   char* stargaddr = (char*)new char[stargtotalsize];
00324  
00325   win_obj *winobj = winObjects[winIndex];
00326   winobj->get(stargaddr, orgcnt, orgunit, targdisp, targcnt, targunit);
00327 
00328   AMPI_DEBUG("    Rank[%d] get win  [%d] \n", thisIndex, *(int*)stargaddr);
00329   
00330   AmpiMsg *msg = new (stargtotalsize, 0) AmpiMsg(-1, -1, -1, thisIndex, stargtotalsize,myComm.getComm());
00331 
00332   char* targaddr = (char*)(winobj->baseAddr) + targunit*targdisp;
00333   tddt->serialize(targaddr, (char*)stargaddr, targcnt, 1);
00334   memcpy(msg->data, stargaddr, stargtotalsize);
00335   AMPI_DEBUG("    Rank[%d] copy win  [%d] \n", thisIndex, *(int*)msg->data);
00336 
00337   AMPI_DEBUG("    Rank[%d] Send to Future [%d] \n", thisIndex, ftHandle);
00338   CkSendToFutureID(ftHandle, (void *)msg, pe_src);
00339   delete [] stargaddr;
00340  } 
00341 
00342 
00343 
00344 int
00345 ampi::winIGet(MPI_Aint orgdisp, int orgcnt, MPI_Datatype orgtype, int rank,
00346              MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, WinStruct win, 
00347              MPI_Request *req){
00348   // Send the request to data and handle of Future to remote side
00349   CProxy_ampi pa(thisArrayID);
00350   AMPI_DEBUG("    Rank[%d:%d] request Remote iget at [%d]\n", thisIndex, myRank, rank);
00351   *req = pa[rank].winRemoteIGet(orgdisp, orgcnt, orgtype, targdisp, targcnt, targtype, win.index);
00352   return MPI_SUCCESS;
00353 }
00354 
00355 
00356 AmpiMsg*
00357 ampi::winRemoteIGet(int orgdisp, int orgcnt, MPI_Datatype orgtype, MPI_Aint targdisp, int targcnt,
00358                    MPI_Datatype targtype, int winIndex) {
00359   AMPI_DEBUG("    RemoteIGet invoked at Rank[%d:%d]\n", thisIndex, myRank);
00360 // FIX: no need for stargaddr??
00361 // FIX: what is targaddr pointing??
00362   int orgunit, targunit;
00363   CkDDT_DataType *oddt = getDDT()->getType(orgtype);
00364   CkDDT_DataType *tddt = getDDT()->getType(targtype);
00365   orgunit = oddt->getSize(1);
00366   targunit = tddt->getSize(1);
00367   int stargtotalsize = targunit*targcnt;
00368   char* stargaddr = (char*)new char[stargtotalsize];
00369 
00370   win_obj *winobj = winObjects[winIndex];
00371   winobj->iget(orgcnt, orgunit, targdisp, targcnt, targunit);
00372 
00373   AMPI_DEBUG("    Rank[%d] iget win  [%d] \n", thisIndex, *(int*)stargaddr);
00374 
00375   AmpiMsg *msg = new (stargtotalsize, 0) AmpiMsg(-1, -1, -1, thisIndex,
00376 stargtotalsize,myComm.getComm());
00377 
00378   char* targaddr = (char*)(winobj->baseAddr) + targdisp*targunit;
00379   tddt->serialize(targaddr, (char*)stargaddr, targcnt, 1);
00380   memcpy(msg->data, stargaddr, stargtotalsize);
00381   AMPI_DEBUG("    Rank[%d] copy win  [%d] \n", thisIndex, *(int*)msg->data);
00382   delete [] stargaddr;
00383   return msg;
00384  }
00385 
00386 int
00387 ampi::winIGetWait(MPI_Request *request, MPI_Status *status) {
00388   // Wait on the Future object
00389   AMPI_DEBUG("    [%d] IGet Waiting\n", thisIndex, *request);
00390   status->msg = (AmpiMsg*)CkWaitReleaseFuture(*request);
00391   AMPI_DEBUG("    [%d] IGet Waiting [%d] awaken\n", thisIndex, *request);
00392   return MPI_SUCCESS;
00393 }
00394 
00395 int
00396 ampi::winIGetFree(MPI_Request *request, MPI_Status *status) {
00397   AMPI_DEBUG("    [%d] : IGet [%d] frees buffer\n", thisIndex, *request);
00398 
00399   
00400   if(!(MPI_IGet_Data(*status))) {
00401     AMPI_DEBUG("    [%d] IGet [%d] attempt to free NULL buffer \n", thisIndex, *request);
00402     return MPI_ERR_BUFFER;
00403   }
00404   else {
00405     delete (status->msg);       
00406     return MPI_SUCCESS; 
00407   }
00408 }
00409 
00410 
00411  
00412 int 
00413 ampi::winAccumulate(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank,
00414                     MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, 
00415                     MPI_Op op, WinStruct win) {
00416   // Create a Future object 
00417   AmpiMsg *msg = new AmpiMsg();
00418   CkFutureID ftHandle = CkCreateAttachedFuture((void*)msg);
00419   AMPI_DEBUG("    Created Future with handle %d\n", ftHandle);
00420 
00421   // Send the request to data and handle of Future to remote side 
00422   CProxy_ampi pa(thisArrayID);
00423 
00424   CkDDT_DataType *ddt = getDDT()->getType(orgtype);
00425   int orgtotalsize = ddt->getSize(orgcnt);
00426   char* sorgaddr = (char*)new char[orgtotalsize];
00427   ddt->serialize((char*)orgaddr, (char*)sorgaddr, orgcnt, 1);
00428 
00429   AMPI_DEBUG("    Rank[%d:%d] invoke Remote accumulate at [%d]\n", thisIndex, myRank, rank);
00430   pa[rank].winRemoteAccumulate(orgtotalsize, (char*)orgaddr, orgcnt, orgtype, targdisp, targcnt, targtype,  op, win.index, ftHandle, CkMyPe());
00431             
00432   // Wait on the Future object 
00433   AMPI_DEBUG("    Future [%d] waiting\n", ftHandle);
00434   msg = (AmpiMsg*)CkWaitFutureID(ftHandle);
00435   AMPI_DEBUG("    Future [%d] awaken\n", ftHandle);
00436   
00437   return MPI_SUCCESS; 
00438 }
00439 
00440 void
00441 ampi::winRemoteAccumulate(int orgtotalsize, char* sorgaddr, int orgcnt, MPI_Datatype orgtype, MPI_Aint targdisp, 
00442                           int targcnt, MPI_Datatype targtype, 
00443                           MPI_Op op, int winIndex, CkFutureID ftHandle, 
00444                           int pe_src) {
00445   win_obj *winobj = winObjects[winIndex];
00446 
00447   CkDDT_DataType *ddt = getDDT()->getType(targtype);
00448   char* getdata = (char*) new char[orgtotalsize];
00449   ddt->serialize(getdata, (char*)sorgaddr, targcnt, (-1));
00450 
00451   winobj->accumulate(getdata, targcnt, targtype, targdisp, targcnt, targtype, op);  
00452     
00453   int tmp = 0;
00454   AmpiMsg *msg = new (tmp, 0) AmpiMsg(-1, -1, -1, thisIndex, 0,myComm.getComm());
00455 
00456   AMPI_DEBUG("    Rank[%d] Send to Future [%d] \n", thisIndex, ftHandle);
00457   CkSendToFutureID(ftHandle, (void *)msg, pe_src);
00458 }
00459 
00460 int 
00461 ampi::winLock(int lock_type, int rank, WinStruct win) {
00462   // Create a Future object 
00463   AmpiMsg *msg = new AmpiMsg();
00464   CkFutureID ftHandle = CkCreateAttachedFuture((void*)msg);
00465   AMPI_DEBUG("    [%d] Lock: Created Future with handle %d\n", thisIndex, ftHandle);
00466 
00467   // Send the request to data and handle of Future to remote side 
00468   CProxy_ampi pa(thisArrayID);
00469   AMPI_DEBUG("    [%d] Lock: invoke Remote lock at [%d]\n", thisIndex, rank);
00470   pa[rank].winRemoteLock(lock_type, win.index, ftHandle, CkMyPe(), thisIndex);
00471 
00472   // Wait on the Future object 
00473   AMPI_DEBUG("    [%d] Lock: Future [%d] waiting\n", thisIndex, ftHandle);
00474   msg = (AmpiMsg*)CkWaitFutureID(ftHandle);
00475   AMPI_DEBUG("    [%d] Lock: Future [%d] awaken\n", thisIndex, ftHandle);
00476   
00477   return MPI_SUCCESS;
00478 }
00479 
00480 void 
00481 ampi::winRemoteLock(int lock_type, int winIndex, CkFutureID ftHandle, int pe_src, int requestRank) {
00482   AMPI_DEBUG("    [%d] RemoteLock: invoked \n", thisIndex);
00483   win_obj *winobj = winObjects[winIndex]; 
00484 
00485   // check if any one else waiting in the queue 
00486   if(winobj->owner > -1 && !(winobj->emptyQueue()))  {
00487   // queue it if queue non-empty
00488     winobj->enqueue(requestRank, pe_src, ftHandle, lock_type);  
00489     AMPI_DEBUG("    [%d] RemoteLock: queue lock from [%d] \n", thisIndex, requestRank); 
00490   }  
00491   // if queue empty, get semaphore and queue it
00492   else {
00493     winobj->lock(requestRank, pe_src, ftHandle, lock_type);
00494     winobj->enqueue(requestRank, pe_src, ftHandle, lock_type);  
00495     AMPI_DEBUG("    [%d] RemoteLock: give lock to [%d] \n", thisIndex, requestRank);
00496   }
00497 }
00498 
00499 int 
00500 ampi::winUnlock(int rank, WinStruct win) {
00501   // Create a Future object 
00502   AmpiMsg *msg = new AmpiMsg();
00503   CkFutureID ftHandle = CkCreateAttachedFuture((void*)msg);
00504   AMPI_DEBUG("    [%d] Unlock: Created Future with handle %d\n", thisIndex, ftHandle);
00505   
00506   // Send the request to data and handle of Future to remote side 
00507   CProxy_ampi pa(thisArrayID);
00508   AMPI_DEBUG("    [%d] Unlock: invoke Remote lock at [%d]\n", thisIndex, rank);
00509   pa[rank].winRemoteUnlock(win.index, ftHandle, CkMyPe(), thisIndex);
00510 
00511   // Wait on the Future object 
00512   AMPI_DEBUG("    [%d] Unlock: Future [%d] waiting\n", thisIndex, ftHandle);
00513   msg = (AmpiMsg*)CkWaitFutureID(ftHandle);
00514   AMPI_DEBUG("    [%d] Unlock: Future [%d] awaken\n", thisIndex, ftHandle);
00515   
00516   return MPI_SUCCESS;
00517 }
00518 
00519 void 
00520 ampi::winRemoteUnlock(int winIndex, CkFutureID ftHandle, int pe_src, int requestRank) {
00521   AMPI_DEBUG("    [%d] RemoteUnlock: invoked \n", thisIndex);
00522   win_obj *winobj = winObjects[winIndex];
00523   winobj->unlock(requestRank, pe_src, ftHandle);  
00524   AMPI_DEBUG("    [%d] RemoteUnlock: [%d] release lock\n", thisIndex, requestRank);
00525 
00526   // if queue non-empty, get lock for the first waiting one and reply
00527   if(!(winobj->emptyQueue())) {
00528     AMPI_DEBUG("    [%d] RemoteUnlock: queue non-empty, give lock to \n", thisIndex );
00529     winobj->lockTopQueue(); 
00530   }
00531 }
00532 
00533 MPI_Win 
00534 ampi::createWinInstance(void *base, MPI_Aint size, int disp_unit, MPI_Info info) {
00535   AMPI_DEBUG("     Creating win obj {%d, %p}\n ", myComm.getComm(), base);
00536   win_obj *newobj = new win_obj((char*)(NULL), base, size, disp_unit, myComm.getComm());
00537   winObjects.push_back(newobj);
00538   WinStruct *newwin = new WinStruct(myComm.getComm(),winObjects.size()-1);
00539   AMPI_DEBUG("     Creating MPI_WIN at (%p) with {%d, %d}\n", &newwin, myComm.getComm(), winObjects.size()-1);
00540   return (parent->addWinStruct(newwin));
00541 }
00542 
00543 int 
00544 ampi::deleteWinInstance(MPI_Win win) {
00545   WinStruct winStruct = parent->getWinStruct(win);
00546   win_obj *winobj = winObjects[winStruct.index];
00547   parent->removeWinStruct(winStruct); // really it does nothing at all
00548   winobj->free();
00549   return MPI_SUCCESS;
00550 }
00551 
00552 int 
00553 ampi::winGetGroup(WinStruct win, MPI_Group *group){
00554    *group = parent->comm2group(win.comm);
00555    return MPI_SUCCESS;
00556 }
00557 
00558 void 
00559 ampi::winSetName(WinStruct win, char *name) {
00560   win_obj *winobj = winObjects[win.index];
00561   winobj->setName((const char*)name, strlen(name));
00562 }
00563 
00564 void 
00565 ampi::winGetName(WinStruct win, char *name, int *length) {
00566   win_obj *winobj = winObjects[win.index];
00567   winobj->getName(name, length);
00568 }
00569 
00570 win_obj*
00571 ampi::getWinObjInstance(WinStruct win) {
00572   return winObjects[win.index];
00573 }
00574 
00575 
00576 /*
00577  * int AMPI_Win_create(void *base, MPI_Aint size, int disp_unit,
00578  *             MPI_Info info, MPI_Comm comm, MPI_Win *newwin)
00579  *   Creates the window object and returns the pointer for *win
00580  *
00581  *   ---Assumption: memory location at *base is pre-allocated 
00582  *   ---by a MPI_Alloc_mem call
00583  *
00584  *   Inputs:
00585  *     void *base : pointer specifying the memory area to create the window
00586  *     MPI_Aint size : size of target memory area (in bytes)
00587  *     int disp_unit : number of bytes for one datatype
00588  *     MPI_Info info : MPI_Info object, provides hints for optimization
00589  *     MPI_Comm comm : communicator 
00590  *     MPI_Win *newwin : stores the handle to the created MPI_Win object on return
00591  *  
00592  *   Returns int: MPI_SUCCESS or MPI_ERR_WIN
00593  */
00594 // A collective call over all processes in the communicator
00595 // MPI_Win object created LOCALLY on all processes when the call returns
00596 CDECL
00597 int AMPI_Win_create(void *base, MPI_Aint size, int disp_unit,
00598                MPI_Info info, MPI_Comm comm, MPI_Win *newwin){
00599   AMPIAPI("AMPI_Win_create");
00600   ampi *ptr = getAmpiInstance(comm);
00601   *newwin = ptr->createWinInstance(base, size, disp_unit, info);
00602   /* need to reduction here: to make sure every processor participates */
00603   AMPI_Barrier(comm);
00604   return MPI_SUCCESS;
00605 }
00606 
00607 /*
00608  * int AMPI_Win_free(MPI_Win *win):
00609  *   Frees the window object and returns a null pointer for *win
00610  */
00611 // A collective call over all processes in the communicator
00612 // MPI_Win object deleted LOCALLY on all processes when the call returns
00613 CDECL
00614 int AMPI_Win_free(MPI_Win *win){
00615   AMPIAPI("AMPI_Win_free");
00616   if(win==NULL) { return MPI_ERR_WIN; }
00617 
00618   WinStruct winStruct = getAmpiParent()->getWinStruct(*win);
00619   ampi *ptr = getAmpiInstance(winStruct.comm);
00620   ptr->deleteWinInstance(*win);
00621   /* Need a barrier here: to ensure that every process participates */
00622   AMPI_Barrier(winStruct.comm);
00623   return MPI_SUCCESS;
00624 }
00625 
00626 CDECL
00627 int AMPI_Win_delete_attr(MPI_Win win, int key){
00628   AMPIAPI("AMPI_Win_delete_attr");
00629   return MPI_SUCCESS;
00630 }
00631 
00632 /*
00633  * ---Note : No atomicity for overlapping Puts. 
00634  * ---sync calls should be made on this window to ensure the 
00635  * ---correctness of the operation
00636  */     
00637 CDECL
00638 int AMPI_Put(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank, 
00639         MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, MPI_Win win){
00640   AMPIAPI("AMPI_Put");
00641   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00642   ampi *ptr = getAmpiInstance(winStruct.comm);
00643   return ptr->winPut(orgaddr, orgcnt, orgtype, rank, targdisp, targcnt, targtype, winStruct);
00644 }
00645 
00646 /*
00647  * ---Note : No atomicity for overlapping Gets. 
00648  * ---sync calls should be made on this window to ensure the 
00649  * ---correctness of the operation
00650  */
00651 CDECL
00652 int AMPI_Get(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank, 
00653         MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, MPI_Win win){
00654   AMPIAPI("AMPI_Get");
00655   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00656   ampi *ptr = getAmpiInstance(winStruct.comm);
00657   // winGet is a local function which will call the remote method on #rank processor 
00658   return  ptr->winGet(orgaddr, orgcnt, orgtype, rank, targdisp, targcnt, targtype, winStruct);
00659 }
00660 
00661 
00662 
00663 /*
00664  * int AMPI_Accumulate(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank,
00665  *                 MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, 
00666  *                 MPI_Op op, MPI_Win win)
00667  *   Accumulates the contents from the origin buffer to the target area using
00668  *   the predefined op operation.
00669  *
00670  * ---Accumulate call is ATOMIC: no sync is needed 
00671  * ---Many accumulate can be made from many origins to one target
00672  */
00673 CDECL
00674 int AMPI_Accumulate(void *orgaddr, int orgcnt, MPI_Datatype orgtype, int rank,
00675                    MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, 
00676                    MPI_Op op, MPI_Win win) {
00677   AMPIAPI("AMPI_Accumulate");
00678   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00679   ampi *ptr = getAmpiInstance(winStruct.comm);                     
00680   return  ptr->winAccumulate(orgaddr, orgcnt, orgtype, rank, 
00681                          targdisp, targcnt, targtype, op, winStruct);   
00682 }
00683 
00684 /*
00685  * int AMPI_Win_fence(int assertion, MPI_Win win)
00686  *   Synchronizes all one-sided communication calls on this MPI_Win.   
00687  *   (Synchronized RMA operations on the specified window)
00688  *
00689  *   Inputs: 
00690  *     int assertion : program assertion, used to provide optimization hints
00691  *   Returns int : MPI_SUCCESS or MPI_ERR_WIN
00692  */
00693 CDECL
00694 int AMPI_Win_fence(int assertion, MPI_Win win){
00695   AMPIAPI("AMPI_Win_fence");
00696   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00697   MPI_Comm comm = winStruct.comm;
00698   ampi *ptr = getAmpiInstance(comm);
00699 
00700   // Wait until everyone reaches the fence
00701   AMPI_Barrier(comm);
00702 
00703   // Complete all outstanding one-sided comm requests
00704   // no need to do this for the pseudo-implementation
00705   return MPI_SUCCESS;
00706 }
00707 
00708 /*
00709  * int AMPI_Win_lock(int lock_type, int rank, int assertion, MPI_Win win)
00710  *   Locks access to this MPI_Win object.   
00711  *   Input:
00712  *     int lock_type : MPI_LOCK_EXCLUSIVE or MPI_LOCK_SHARED
00713  *     int rank : rank of locked window
00714  *     int assertion : program assertion, used to provide optimization hints
00715  *   Returns int : MPI_SUCCESS or MPI_ERR_WIN
00716  */
00717 CDECL
00718 int AMPI_Win_lock(int lock_type, int rank, int assertion, MPI_Win win){
00719   AMPIAPI("AMPI_Win_lock");
00720   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00721   ampi *ptr = getAmpiInstance(winStruct.comm);
00722 
00723   // process assertion here:   
00724   // end of assertion
00725   ptr->winLock(lock_type, rank, winStruct);
00726   return MPI_SUCCESS;
00727 }
00728 
00729 /*
00730  * int AMPI_Win_unlock(int rank, MPI_Win win)
00731  *   Unlocks access to this MPI_Win object.   
00732  *   Input:
00733  *     int rank : rank of locked window
00734  *   Returns int : MPI_SUCCESS or MPI_ERR_WIN
00735  */
00736 // The RMA call is completed both locally and remotely after unlock. 
00737   // process assertion here: HOW???  
00738 int AMPI_Win_unlock(int rank, MPI_Win win){
00739   AMPIAPI("AMPI_Win_unlock");
00740   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00741   ampi *ptr = getAmpiInstance(winStruct.comm);
00742 
00743   // process assertion here: HOW???  
00744   // end of assertion
00745   ptr->winUnlock(rank, winStruct);
00746   return MPI_SUCCESS;
00747 }
00748 
00749 /* the following four functions are yet to implement */
00750 /*
00751  * int AMPI_Win_post(MPI_Group group, int assertion, MPI_Win win)
00752  *   Opens a RMA access epoch for local window win.
00753  *   Only processes in group can access this window with RMA calls.
00754  *   Each process must issue a matching MPI_Win_start to start the 
00755  *     access epoch.
00756  *   Post is non-blocking while start could be blocking.
00757  *   Input:
00758  *     MPI_Group group : a group of processes 
00759  *   Returns int : MPI_SUCCESS or MPI_ERR_WIN
00760  */
00761 CDECL
00762 int AMPI_Win_post(MPI_Group group, int assertion, MPI_Win win){
00763   AMPIAPI("AMPI_Win_post");
00764   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00765   ampi *ptr = getAmpiInstance(winStruct.comm);
00766   //  ptr->winPost(group, winStruct);
00767   return MPI_SUCCESS;
00768 }
00769 
00770 CDECL
00771 int AMPI_Win_wait(MPI_Win win){
00772   AMPIAPI("AMPI_Win_wait");
00773   return MPI_SUCCESS;
00774 }
00775 
00776 CDECL
00777 int AMPI_Win_start(MPI_Group group, int assertion, MPI_Win win){
00778   AMPIAPI("AMPI_Win_start");
00779   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00780   ampi *ptr = getAmpiInstance(winStruct.comm);
00781   //  ptr->winStart(group, winStruct);
00782   return MPI_SUCCESS;
00783 }
00784 
00785 CDECL
00786 int AMPI_Win_complete(MPI_Win win){
00787   AMPIAPI("AMPI_Win_complete");
00788   return MPI_SUCCESS;
00789 }
00790 
00791 // FIX PLACE II
00792 CDECL
00793 int AMPI_IGet(MPI_Aint orgdisp, int orgcnt, MPI_Datatype orgtype, int rank,
00794         MPI_Aint targdisp, int targcnt, MPI_Datatype targtype, MPI_Win win,
00795         MPI_Request *request) {
00796   AMPIAPI("AMPI_IGet");
00797   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00798   ampi *ptr = getAmpiInstance(winStruct.comm);
00799   // winGet is a local function which will call the remote method on #rank processor
00800   return  ptr->winIGet(orgdisp, orgcnt, orgtype, rank, targdisp, targcnt, targtype, winStruct,
00801                        request);
00802 }
00803 
00804 CDECL
00805 int AMPI_IGet_Wait(MPI_Request *request, MPI_Status *status, MPI_Win win) {
00806   AMPIAPI("AMPI_IGet_Wait");
00807   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00808   ampi *ptr = getAmpiInstance(winStruct.comm);
00809   // winGet is a local function which will call the remote method on #rank processor
00810   return  ptr->winIGetWait(request,status);
00811 }
00812 
00813 CDECL
00814 int AMPI_IGet_Free(MPI_Request *request, MPI_Status *status, MPI_Win win) {
00815   AMPIAPI("AMPI_IGet_Free");
00816   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00817   ampi *ptr = getAmpiInstance(winStruct.comm);
00818   // winGet is a local function which will call the remote method on #rank processor
00819   return  ptr->winIGetFree(request, status);
00820 }
00821 
00822 char* 
00823 MPI_IGet_Data(MPI_Status status) 
00824 { return (char*)((AmpiMsg*)status.msg)->data; }
00825 
00826 /*
00827  * int AMPI_Alloc_mem(MPI_Aint size, MPI_Info info, void *baseptr) 
00828  *   A simple wrapper around 'malloc' call. Used to allocate memory
00829  *   for MPI functions
00830  *
00831  *   Inputs: 
00832  *     MPI_Aint size : size of target memory area
00833  *     MPI_Info info : 
00834  *     void *base : stores the base pointer of target memory area on return
00835  *   Return: 
00836  *     void* : address of the allocated memory
00837  */
00838 CDECL
00839 int AMPI_Alloc_mem(MPI_Aint size, MPI_Info info, void *baseptr){
00840   AMPIAPI("AMPI_Alloc_mem");
00841   *(void **)baseptr = malloc(size);
00842   return MPI_SUCCESS;
00843 }
00844 
00845 
00846 /*
00847  * int AMPI_Free_mem(void *base)
00848  *   Frees memory that was previous allocated by MPI_Alloc_mem call
00849  */
00850 CDECL
00851 int AMPI_Free_mem(void *baseptr){
00852   AMPIAPI("AMPI_Free_mem");
00853   free(baseptr);
00854   return MPI_SUCCESS;
00855 }
00856 
00857 CDECL
00858 int AMPI_Win_get_group(MPI_Win win, MPI_Group *group) {
00859   AMPIAPI("AMPI_Win_get_group");
00860   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00861   ampi *ptr = getAmpiInstance(winStruct.comm);
00862   ptr->winGetGroup(winStruct, group);
00863   return MPI_SUCCESS;
00864 }
00865 
00866 CDECL
00867 int AMPI_Win_set_attr(MPI_Win win, int key, void* value) {
00868   AMPIAPI("AMPI_Win_set_attr");
00869   return MPI_SUCCESS;
00870 }
00871 
00872 CDECL
00873 int AMPI_Win_set_name(MPI_Win win, char *name) {
00874   AMPIAPI("AMPI_Win_set_name");
00875   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00876   ampi *ptr = getAmpiInstance(winStruct.comm);
00877   ptr->winSetName(winStruct, name);
00878   return MPI_SUCCESS;
00879 }
00880 
00881 CDECL
00882 int AMPI_Win_get_name(MPI_Win win, char *name, int *length) {
00883   AMPIAPI("AMPI_Win_get_name");
00884   WinStruct winStruct = getAmpiParent()->getWinStruct(win);
00885   ampi *ptr = getAmpiInstance(winStruct.comm);
00886   ptr->winGetName(winStruct, name, length);
00887   return MPI_SUCCESS;
00888 }
00889
libs/ck-libs/ampi/ampiOneSided.C
