Changeset 2019 for XIOS/dev/dev_trunk_graph/src/context_client.cpp

Timestamp:

01/22/21 12:00:29 (3 years ago)

Author:

yushan

Message:

Graph intermedia commit to a tmp branch

File:

• XIOS/dev/dev_trunk_graph/src/context_client.cpp (modified) (18 diffs)

XIOS/dev/dev_trunk_graph/src/context_client.cpp

@@ -22 +22 @@
     */
     CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
-     : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4)
-    {
+     : mapBufferSize_(), parentServer(cxtSer), maxBufferedEvents(4), associatedServer_(nullptr)
+    {
+      
       context = parent;
       intraComm = intraComm_;
@@ -32 +33 @@
       int flag;
       MPI_Comm_test_inter(interComm, &flag);
+      if (flag) isAttached_=false ;
+      else  isAttached_=true ;
+
+      pureOneSided=CXios::getin<bool>("pure_one_sided",false); // pure one sided communication (for test)
+      if (isAttachedModeEnabled()) pureOneSided=false ; // no one sided in attach mode
+      
+
+
       if (flag) MPI_Comm_remote_size(interComm, &serverSize);
       else  MPI_Comm_size(interComm, &serverSize);
@@ -37 +46 @@
       computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
 
-      timeLine = 0;
+      if (flag) MPI_Intercomm_merge(interComm_,false,&interCommMerged) ;
+      
+      if (!isAttachedModeEnabled())
+      {  
+        windows.resize(serverSize) ;
+        MPI_Comm winComm ;
+        for(int rank=0; rank<serverSize; rank++)
+        {
+          windows[rank].resize(2) ;
+          MPI_Comm_split(interCommMerged, rank, clientRank, &winComm);
+          int myRank ;
+          MPI_Comm_rank(winComm,&myRank);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][0]);
+          MPI_Win_create_dynamic(MPI_INFO_NULL, winComm, &windows[rank][1]);
+          MPI_Comm_free(&winComm) ;
+        }
+      }
+
+      MPI_Comm_split(intraComm_,clientRank,clientRank, &commSelf) ;
+
+      timeLine = 1;
     }
 
@@ -116 +145 @@
         list<int> sizes = event.getSizes();
 
-        // We force the getBuffers call to be non-blocking on classical servers
+         // We force the getBuffers call to be non-blocking on classical servers
         list<CBufferOut*> buffList;
-        bool couldBuffer = getBuffers(ranks, sizes, buffList, (!CXios::isClient && (CServer::serverLevel == 0) ));
-//        bool couldBuffer = getBuffers(ranks, sizes, buffList, CXios::isServer );
-
-        if (couldBuffer)
-        {
-          event.send(timeLine, sizes, buffList);
-          info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
-
-          checkBuffers(ranks);
-
-          if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
-          {
-            waitEvent(ranks);
-            CContext::setCurrent(context->getId());
-          }
-        }
-        else
-        {
-          tmpBufferedEvent.ranks = ranks;
-          tmpBufferedEvent.sizes = sizes;
-
-          for (list<int>::const_iterator it = sizes.begin(); it != sizes.end(); it++)
-            tmpBufferedEvent.buffers.push_back(new CBufferOut(*it));
-          info(100)<<"DEBUG : temporaly event created : timeline "<<timeLine<<endl ;
-          event.send(timeLine, tmpBufferedEvent.sizes, tmpBufferedEvent.buffers);
-          info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
-        }
-      }
-
+        getBuffers(timeLine, ranks, sizes, buffList) ;
+
+        event.send(timeLine, sizes, buffList);
+       
+        //for (auto itRank = ranks.begin(); itRank != ranks.end(); itRank++) buffers[*itRank]->infoBuffer() ;
+
+        unlockBuffers(ranks) ;
+        info(100)<<"Event "<<timeLine<<" of context "<<context->getId()<<"  sent"<<endl ;
+          
+        checkBuffers(ranks);
+      }
+      
+      if (isAttachedModeEnabled()) // couldBuffer is always true in attached mode
+      {
+        waitEvent(ranks);
+        CContext::setCurrent(context->getId());
+      }
+      
       timeLine++;
-    }
-
-    /*!
-     * Send the temporarily buffered event (if any).
-     *
-     * \return true if a temporarily buffered event could be sent, false otherwise 
-     */
-    bool CContextClient::sendTemporarilyBufferedEvent()
-    {
-      bool couldSendTmpBufferedEvent = false;
-
-      if (hasTemporarilyBufferedEvent())
-      {
-        list<CBufferOut*> buffList;
-        if (getBuffers(tmpBufferedEvent.ranks, tmpBufferedEvent.sizes, buffList, true)) // Non-blocking call
-        {
-          list<CBufferOut*>::iterator it, itBuffer;
-
-          for (it = tmpBufferedEvent.buffers.begin(), itBuffer = buffList.begin(); it != tmpBufferedEvent.buffers.end(); it++, itBuffer++)
-            (*itBuffer)->put((char*)(*it)->start(), (*it)->count());
-
-          info(100)<<"DEBUG : temporaly event sent "<<endl ;
-          checkBuffers(tmpBufferedEvent.ranks);
-
-          tmpBufferedEvent.clear();
-
-          couldSendTmpBufferedEvent = true;
-        }
-      }
-
-      return couldSendTmpBufferedEvent;
     }
 
@@ -188 +175 @@
     void CContextClient::waitEvent(list<int>& ranks)
     {
+      while (checkBuffers(ranks))
+      {
+        CXios::getDaemonsManager()->eventLoop() ;
+      }
+
+      MPI_Request req ;
+      MPI_Status status ;
+
+      MPI_Ibarrier(intraComm,&req) ;
+      int flag=false ;
+
+      do  
+      {
+        CXios::getDaemonsManager()->eventLoop() ;
+        MPI_Test(&req,&flag,&status) ;
+      } while (!flag) ;
+
+
+    }
+
+
+    void CContextClient::waitEvent_old(list<int>& ranks)
+    {
       parentServer->server->setPendingEvent();
       while (checkBuffers(ranks))
@@ -205 +215 @@
      * it is explicitly requested to be non-blocking.
      *
+     *
+     * \param [in] timeLine time line of the event which will be sent to servers
      * \param [in] serverList list of rank of connected server
      * \param [in] sizeList size of message corresponding to each connection
@@ -211 +223 @@
      * \return whether the already allocated buffers could be used
     */
-    bool CContextClient::getBuffers(const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
+    bool CContextClient::getBuffers(const size_t timeLine, const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
                                     bool nonBlocking /*= false*/)
     {
@@ -236 +248 @@
         areBuffersFree = true;
         for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
+        {
           areBuffersFree &= (*itBuffer)->isBufferFree(*itSize);
+        }
 
         if (!areBuffersFree)
         {
+          for (itBuffer = bufferList.begin(); itBuffer != bufferList.end(); itBuffer++) (*itBuffer)->unlockBuffer();
           checkBuffers();
-          if (CServer::serverLevel == 0)
-            context->server->listen();
-
-          else if (CServer::serverLevel == 1)
+          
+          context->server->listen();
+
+          if (context->serverPrimServer.size()>0)
           {
-            context->server->listen();
-            for (int i = 0; i < context->serverPrimServer.size(); ++i)
-              context->serverPrimServer[i]->listen();
-            CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
+            for (int i = 0; i < context->serverPrimServer.size(); ++i)  context->serverPrimServer[i]->listen();
+ //ym           CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
+            context->globalEventLoop() ;
           }
 
-          else if (CServer::serverLevel == 2)
-            context->server->listen();
-
         }
       } while (!areBuffersFree && !nonBlocking);
-
       CTimer::get("Blocking time").suspend();
 
@@ -263 +273 @@
       {
         for (itBuffer = bufferList.begin(), itSize = sizeList.begin(); itBuffer != bufferList.end(); itBuffer++, itSize++)
-          retBuffers.push_back((*itBuffer)->getBuffer(*itSize));
-      }
-
+          retBuffers.push_back((*itBuffer)->getBuffer(timeLine, *itSize));
+      }
       return areBuffersFree;
    }
@@ -281 +290 @@
         maxEventSizes[rank] = CXios::minBufferSize;
       }
-      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, rank, mapBufferSize_[rank], maxEventSizes[rank], maxBufferedEvents);
+      
+      vector<MPI_Win> Wins(2,MPI_WIN_NULL) ;
+      if (!isAttachedModeEnabled()) Wins=windows[rank] ;
+  
+      CClientBuffer* buffer = buffers[rank] = new CClientBuffer(interComm, Wins, clientRank, rank, mapBufferSize_[rank], maxEventSizes[rank]);
       // Notify the server
-      CBufferOut* bufOut = buffer->getBuffer(sizeof(StdSize));
-      bufOut->put(mapBufferSize_[rank]); // Stupid C++
-      buffer->checkBuffer();
+      CBufferOut* bufOut = buffer->getBuffer(0, 3*sizeof(MPI_Aint));
+      MPI_Aint sendBuff[3] ;
+      sendBuff[0]=mapBufferSize_[rank]; // Stupid C++
+      sendBuff[1]=buffers[rank]->getWinAddress(0); 
+      sendBuff[2]=buffers[rank]->getWinAddress(1); 
+      info(100)<<"CContextClient::newBuffer : rank "<<rank<<" winAdress[0] "<<buffers[rank]->getWinAddress(0)<<" winAdress[1] "<<buffers[rank]->getWinAddress(1)<<endl;
+      bufOut->put(sendBuff, 3); // Stupid C++
+      buffer->checkBuffer(true);
+
    }
 
@@ -297 +316 @@
       bool pending = false;
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
-        pending |= itBuff->second->checkBuffer();
+        pending |= itBuff->second->checkBuffer(!pureOneSided);
       return pending;
    }
@@ -307 +326 @@
       for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
       {
-          delete itBuff->second;
+         delete itBuff->second;
       }
       buffers.clear();
-   }
-
+
+/* don't know when release windows
+
+      if (!isAttachedModeEnabled())
+      {  
+        for(int rank=0; rank<serverSize; rank++)
+        {
+          MPI_Win_free(&windows[rank][0]);
+          MPI_Win_free(&windows[rank][1]);
+        }
+      } 
+*/
+   }
+
+      
+  /*!
+   Lock the buffers for one sided communications
+   \param [in] ranks list rank of server to which client connects to
+   */
+   void CContextClient::lockBuffers(list<int>& ranks)
+   {
+      list<int>::iterator it;
+      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->lockBuffer();
+   }
+
+  /*!
+   Unlock the buffers for one sided communications
+   \param [in] ranks list rank of server to which client connects to
+   */
+   void CContextClient::unlockBuffers(list<int>& ranks)
+   {
+      list<int>::iterator it;
+      for (it = ranks.begin(); it != ranks.end(); it++) buffers[*it]->unlockBuffer();
+   }
+      
    /*!
    Verify state of buffers corresponding to a connection
@@ -321 +373 @@
       list<int>::iterator it;
       bool pending = false;
-      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer();
+      for (it = ranks.begin(); it != ranks.end(); it++) pending |= buffers[*it]->checkBuffer(!pureOneSided);
       return pending;
    }
@@ -335 +387 @@
      mapBufferSize_ = mapSize;
      maxEventSizes = maxEventSize;
-
-     // Compute the maximum number of events that can be safely buffered.
-     double minBufferSizeEventSizeRatio = std::numeric_limits<double>::max();
-     for (std::map<int,StdSize>::const_iterator it = mapSize.begin(), ite = mapSize.end(); it != ite; ++it)
-     {
-       double ratio = double(it->second) / maxEventSizes[it->first];
-       if (ratio < minBufferSizeEventSizeRatio) minBufferSizeEventSizeRatio = ratio;
-     }
-     MPI_Allreduce(MPI_IN_PLACE, &minBufferSizeEventSizeRatio, 1, MPI_DOUBLE, MPI_MIN, intraComm);
-
-     if (minBufferSizeEventSizeRatio < 1.0)
-     {
-       ERROR("void CContextClient::setBufferSize(const std::map<int,StdSize>& mapSize, const std::map<int,StdSize>& maxEventSize)",
-             << "The buffer sizes and the maximum events sizes are incoherent.");
-     }
-     else if (minBufferSizeEventSizeRatio == std::numeric_limits<double>::max())
-       minBufferSizeEventSizeRatio = 1.0; // In this case, maxBufferedEvents will never be used but we want to avoid any floating point exception
-
-     maxBufferedEvents = size_t(2 * minBufferSizeEventSizeRatio) // there is room for two local buffers on the server
-                          + size_t(minBufferSizeEventSizeRatio)  // one local buffer can always be fully used
-                          + 1;                                   // the other local buffer might contain only one event
    }
 
@@ -394 +425 @@
   }
 
-  /*!
-   * Check if the attached mode is used.
-   *
-   * \return true if and only if attached mode is used
-   */
-  bool CContextClient::isAttachedModeEnabled() const
-  {
-    return (parentServer != 0);
-  }
-
    /*!
    * Finalize context client and do some reports. Function is non-blocking.
@@ -410 +431 @@
   {
     map<int,CClientBuffer*>::iterator itBuff;
+    std::list<int>::iterator ItServerLeader; 
+    
     bool stop = false;
 
+    int* nbServerConnectionLocal  = new int[serverSize] ;
+    int* nbServerConnectionGlobal  = new int[serverSize] ;
+    for(int i=0;i<serverSize;++i) nbServerConnectionLocal[i]=0 ;
+    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)  nbServerConnectionLocal[itBuff->first]=1 ;
+    for (ItServerLeader = ranksServerLeader.begin(); ItServerLeader != ranksServerLeader.end(); ItServerLeader++)  nbServerConnectionLocal[*ItServerLeader]=1 ;
+    
+    MPI_Allreduce(nbServerConnectionLocal, nbServerConnectionGlobal, serverSize, MPI_INT, MPI_SUM, intraComm);
+    
+    CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
+    CMessage msg;
+
+    for (int i=0;i<serverSize;++i) if (nbServerConnectionLocal[i]==1) event.push(i, nbServerConnectionGlobal[i], msg) ;
+    sendEvent(event);
+
+    delete[] nbServerConnectionLocal ;
+    delete[] nbServerConnectionGlobal ;
+
+
     CTimer::get("Blocking time").resume();
-    while (hasTemporarilyBufferedEvent())
-    {
-      checkBuffers();
-      sendTemporarilyBufferedEvent();
-    }
-    CTimer::get("Blocking time").suspend();
-
-    CEventClient event(CContext::GetType(), CContext::EVENT_ID_CONTEXT_FINALIZE);
-    if (isServerLeader())
-    {
-      CMessage msg;
-      const std::list<int>& ranks = getRanksServerLeader();
-      for (std::list<int>::const_iterator itRank = ranks.begin(), itRankEnd = ranks.end(); itRank != itRankEnd; ++itRank)
-      {
-        info(100)<<"DEBUG : Sent context Finalize event to rank "<<*itRank<<endl ;
-        event.push(*itRank, 1, msg);
-      }
-      sendEvent(event);
-    }
-    else sendEvent(event);
-
-    CTimer::get("Blocking time").resume();
-//    while (!stop)
-    {
-      checkBuffers();
-      if (hasTemporarilyBufferedEvent())
-        sendTemporarilyBufferedEvent();
-
-      stop = true;
-//      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) stop &= !itBuff->second->hasPendingRequest();
-    }
+    checkBuffers();
     CTimer::get("Blocking time").suspend();
 
@@ -458 +469 @@
     report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
 
-    //releaseBuffers(); // moved to CContext::finalize()
   }
 
@@ -472 +482 @@
     return pending;
   }
-
+  
+  bool CContextClient::isNotifiedFinalized(void)
+  {
+    if (isAttachedModeEnabled()) return true ;
+
+    bool finalized = true;
+    map<int,CClientBuffer*>::iterator itBuff;
+    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
+      finalized &= itBuff->second->isNotifiedFinalized();
+    return finalized;
+  }
 
 }