Changeset 1460 for XIOS/dev/branch_openmp/src/context_client.cpp

Timestamp:

03/22/18 10:43:20 (6 years ago)

Author:

yushan

Message:

branch_openmp merged with XIOS_DEV_CMIP6@1459

File:

• XIOS/dev/branch_openmp/src/context_client.cpp (modified) (13 diffs)

XIOS/dev/branch_openmp/src/context_client.cpp

@@ -11 +11 @@
 #include "timer.hpp"
 #include "cxios.hpp"
+#include "server.hpp"
 using namespace ep_lib;
 
@@ -19 +20 @@
     \param [in] intraComm_ communicator of group client
     \param [in] interComm_ communicator of group server
-    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used on case of attached mode)
+    \cxtSer [in] cxtSer Pointer to context of server side. (It is only used in case of attached mode).
     */
     CContextClient::CContextClient(CContext* parent, MPI_Comm intraComm_, MPI_Comm interComm_, CContext* cxtSer)
@@ -40 +41 @@
       else  MPI_Comm_size(interComm, &serverSize);
 
+      computeLeader(clientRank, clientSize, serverSize, ranksServerLeader, ranksServerNotLeader);
+
+      timeLine = 0;
+    }
+
+    void CContextClient::computeLeader(int clientRank, int clientSize, int serverSize,
+                                       std::list<int>& rankRecvLeader,
+                                       std::list<int>& rankRecvNotLeader)
+    {
+      if ((0 == clientSize) || (0 == serverSize)) return;
+
       if (clientSize < serverSize)
       {
@@ -55 +67 @@
 
         for (int i = 0; i < serverByClient; i++)
-          ranksServerLeader.push_back(rankStart + i);
-
-        ranksServerNotLeader.resize(0);
+          rankRecvLeader.push_back(rankStart + i);
+
+        rankRecvNotLeader.resize(0);
       }
       else
@@ -67 +79 @@
         {
           if (clientRank % (clientByServer + 1) == 0)
-            ranksServerLeader.push_back(clientRank / (clientByServer + 1));
+            rankRecvLeader.push_back(clientRank / (clientByServer + 1));
           else
-            ranksServerNotLeader.push_back(clientRank / (clientByServer + 1));
+            rankRecvNotLeader.push_back(clientRank / (clientByServer + 1));
         }
         else
@@ -75 +87 @@
           int rank = clientRank - (clientByServer + 1) * remain;
           if (rank % clientByServer == 0)
-            ranksServerLeader.push_back(remain + rank / clientByServer);
+            rankRecvLeader.push_back(remain + rank / clientByServer);
           else
-            ranksServerNotLeader.push_back(remain + rank / clientByServer);
-        }        
-      }
-
-      timeLine = 0;
+            rankRecvNotLeader.push_back(remain + rank / clientByServer);
+        }
+      }
     }
 
@@ -92 +102 @@
       list<int> ranks = event.getRanks();
 
+      if (CXios::checkEventSync)
+      {
+        int typeId, classId, typeId_in, classId_in, timeLine_out;
+        typeId_in=event.getTypeId() ;
+        classId_in=event.getClassId() ;
+        MPI_Allreduce(&timeLine,&timeLine_out, 1, MPI_UINT64_T, MPI_SUM, intraComm) ;
+        MPI_Allreduce(&typeId_in,&typeId, 1, MPI_INT, MPI_SUM, intraComm) ;
+        MPI_Allreduce(&classId_in,&classId, 1, MPI_INT, MPI_SUM, intraComm) ;
+        if (typeId/clientSize!=event.getTypeId() || classId/clientSize!=event.getClassId() || timeLine_out/clientSize!=timeLine)
+        {
+           ERROR("void CContextClient::sendEvent(CEventClient& event)",
+               << "Event are not coherent between client.");
+        }
+      }
+
       if (!event.isEmpty())
       {
         list<int> sizes = event.getSizes();
 
-        // We force the getBuffers call to be non-blocking on the servers
+        // We force the getBuffers call to be non-blocking on classical servers
         list<CBufferOut*> buffList;
-        bool couldBuffer = getBuffers(ranks, sizes, buffList, !CXios::isClient);
+        bool couldBuffer = getBuffers(ranks, sizes, buffList, (!CXios::isClient && (CServer::serverLevel == 0) ));
+//        bool couldBuffer = getBuffers(ranks, sizes, buffList, CXios::isServer );
 
         if (couldBuffer)
@@ -119 +145 @@
           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);
         }
@@ -146 +172 @@
             (*itBuffer)->put((char*)(*it)->start(), (*it)->count());
 
+          info(100)<<"DEBUG : temporaly event sent "<<endl ;
           checkBuffers(tmpBufferedEvent.ranks);
 
@@ -187 +214 @@
      * \return whether the already allocated buffers could be used
     */
-    bool CContextClient::getBuffers(const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers, bool nonBlocking /*= false*/)
+    bool CContextClient::getBuffers(const list<int>& serverList, const list<int>& sizeList, list<CBufferOut*>& retBuffers,
+                                    bool nonBlocking /*= false*/)
     {
       list<int>::const_iterator itServer, itSize;
@@ -216 +244 @@
         {
           checkBuffers();
-          context->server->listen();
+          if (CServer::serverLevel == 0)
+            context->server->listen();
+
+          else if (CServer::serverLevel == 1)
+          {
+            context->server->listen();
+            for (int i = 0; i < context->serverPrimServer.size(); ++i)
+              context->serverPrimServer[i]->listen();
+            CServer::contextEventLoop(false) ; // avoid dead-lock at finalize...
+          }
+
+          else if (CServer::serverLevel == 2)
+            context->server->listen();
+
         }
       } while (!areBuffersFree && !nonBlocking);
+
       CTimer::get("Blocking time").suspend();
 
@@ -257 +299 @@
       map<int,CClientBuffer*>::iterator itBuff;
       bool pending = false;
-      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) pending |= itBuff->second->checkBuffer();
+      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
+        pending |= itBuff->second->checkBuffer();
       return pending;
    }
 
    //! Release all buffers
-   void CContextClient::releaseBuffers(void)
+   void CContextClient::releaseBuffers()
    {
       map<int,CClientBuffer*>::iterator itBuff;
-      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++) delete itBuff->second;
+      for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
+      {
+          delete itBuff->second;
+      }
+      buffers.clear();
    }
 
@@ -362 +409 @@
 
    /*!
-   Finalize context client and do some reports
-   */
-   void CContextClient::finalize(void)
-   {
-     map<int,CClientBuffer*>::iterator itBuff;
-     bool stop = false;
-
-     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)
-         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();
-     }
-     CTimer::get("Blocking time").suspend();
-
-     std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
-                                           iteMap = mapBufferSize_.end(), itMap;
-     StdSize totalBuf = 0;
-     for (itMap = itbMap; itMap != iteMap; ++itMap)
-     {
-       //report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
-       //           << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
-       totalBuf += itMap->second;
-     }
-     //report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
-
-     releaseBuffers();
-   }
+   * Finalize context client and do some reports. Function is non-blocking.
+   */
+  void CContextClient::finalize(void)
+  {
+    map<int,CClientBuffer*>::iterator itBuff;
+    bool stop = false;
+
+    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)
+      {
+        #pragma omp critical (_output)
+        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();
+    }
+    CTimer::get("Blocking time").suspend();
+
+    std::map<int,StdSize>::const_iterator itbMap = mapBufferSize_.begin(),
+                                          iteMap = mapBufferSize_.end(), itMap;
+
+    StdSize totalBuf = 0;
+    for (itMap = itbMap; itMap != iteMap; ++itMap)
+    {
+      #pragma omp critical (_output)
+      report(10) << " Memory report : Context <" << context->getId() << "> : client side : memory used for buffer of each connection to server" << endl
+                 << "  +) To server with rank " << itMap->first << " : " << itMap->second << " bytes " << endl;
+      totalBuf += itMap->second;
+    }
+    #pragma omp critical (_output)
+    report(0) << " Memory report : Context <" << context->getId() << "> : client side : total memory used for buffer " << totalBuf << " bytes" << endl;
+
+    //releaseBuffers(); // moved to CContext::finalize()
+  }
+
+
+  /*!
+  */
+  bool CContextClient::havePendingRequests(void)
+  {
+    bool pending = false;
+    map<int,CClientBuffer*>::iterator itBuff;
+    for (itBuff = buffers.begin(); itBuff != buffers.end(); itBuff++)
+      pending |= itBuff->second->hasPendingRequest();
+    return pending;
+  }
+
+
 }