Changeset 1243

Timestamp:

08/28/17 12:15:49 (7 years ago)

Author:

oabramkina

Message:

Modifications allowing secondary-server pools of different size. It's controlled by a new parameter number_pools_server2.
Tests: complete, toy, ipsl.

This functionality is to be used in future. For now it is disabled and XIOS works as before with one process assigned per pool.

Location:

XIOS/dev/XIOS_DEV_CMIP6/src

Files:

• client.cpp (modified) (2 diffs)
• client.hpp (modified) (1 diff)
• cxios.cpp (modified) (4 diffs)
• cxios.hpp (modified) (1 diff)
• node/grid.hpp (modified) (4 diffs)
• server.cpp (modified) (7 diffs)
• server.hpp (modified) (2 diffs)

XIOS/dev/XIOS_DEV_CMIP6/src/client.cpp

@@ -96 +96 @@
           }
 
-//          myColor=colors[hashClient];
-          myColor=leaders[hashClient] ;
+          myColor=colors[hashClient];
           MPI_Comm_split(CXios::globalComm,myColor,rank_,&intraComm) ;
 
@@ -275 +274 @@
 
     /*!
-    * Return rank in model intraComm
+    * Return global rank without oasis and current rank in model intraComm in case of oasis
     */
    int CClient::getRank()

XIOS/dev/XIOS_DEV_CMIP6/src/client.hpp

@@ -22 +22 @@
         static MPI_Comm& getInterComm();
 
-        //! Get rank of the current process
+        //! Get global rank without oasis and current rank in model intraComm in case of oasis
         static int getRank();
 

XIOS/dev/XIOS_DEV_CMIP6/src/cxios.cpp

@@ -28 +28 @@
   bool CXios::usingServer2 = false;
   int CXios::ratioServer2 = 50;
+  int CXios::nbPoolsServer2 = 1;
   double CXios::bufferSizeFactor = 1.0;
   const double CXios::defaultBufferSizeFactor = 1.0;
@@ -55 +56 @@
     usingServer2=getin<bool>("using_server2",false) ;
     ratioServer2=getin<int>("ratio_server2",50);
+    nbPoolsServer2=getin<int>("number_pools_server2",1);
     info.setLevel(getin<int>("info_level",0)) ;
     report.setLevel(getin<int>("info_level",50));
@@ -197 +199 @@
     else
     {
-      // If using two server levels, first merge globalRegistry of each server pool into one registry on the first pool
+      // If using two server levels:
+      // (1) merge registries on each pool
+      // (2) send merged registries to the first pool
+      // (3) merge received registries on the first pool
       if (CServer::serverLevel == 2)
       {
@@ -205 +210 @@
         MPI_Comm_rank(globalComm, &rankGlobal);
 
+        // Merge registries defined on each pools
+        CRegistry globalRegistrySndServers (CServer::intraComm);
+
         // All pools (except the first): send globalRegistry to the first pool
-        if (rankGlobal != firstPoolGlobalRank)
+        for (int i=1; i<secondaryServerGlobalRanks.size(); i++)
         {
-          int registrySize = globalRegistry->size();
-          MPI_Send(&registrySize,1,MPI_LONG,firstPoolGlobalRank,15,CXios::globalComm) ;
-          CBufferOut buffer(registrySize) ;
-          globalRegistry->toBuffer(buffer) ;
-          MPI_Send(buffer.start(),registrySize,MPI_CHAR,firstPoolGlobalRank,15,CXios::globalComm) ;
+          if (rankGlobal == secondaryServerGlobalRanks[i])
+          {
+            globalRegistrySndServers.mergeRegistry(*globalRegistry) ;
+            int registrySize = globalRegistrySndServers.size();
+            MPI_Send(&registrySize,1,MPI_LONG,firstPoolGlobalRank,15,CXios::globalComm) ;
+            CBufferOut buffer(registrySize) ;
+            globalRegistrySndServers.toBuffer(buffer) ;
+            MPI_Send(buffer.start(),registrySize,MPI_CHAR,firstPoolGlobalRank,15,CXios::globalComm) ;
+          }
         }
 
         // First pool: receive globalRegistry of all secondary server pools, merge and write the resultant registry
-        else
+        if (rankGlobal == firstPoolGlobalRank)
         {
           MPI_Status status;
           char* recvBuff;
 
-          CRegistry globalRegistrySndServers (CServer::intraComm);
           globalRegistrySndServers.mergeRegistry(*globalRegistry) ;
 

XIOS/dev/XIOS_DEV_CMIP6/src/cxios.hpp

@@ -40 +40 @@
 
      static bool printLogs2Files; //!< Printing out logs into files
-     static bool usingOasis ; //!< Using Oasis
-     static bool usingServer ; //!< Using server (server mode)
-     static bool usingServer2 ; //!< Using secondary server (server mode). IMPORTANT: Use this variable ONLY in CServer::initialize().
-     static int ratioServer2 ;  //!< Percentage of server processors dedicated to secondary server
+     static bool usingOasis ;     //!< Using Oasis
+     static bool usingServer ;    //!< Using server (server mode)
+     static bool usingServer2 ;   //!< Using secondary server (server mode). IMPORTANT: Use this variable ONLY in CServer::initialize().
+     static int ratioServer2 ;    //!< Percentage of server processors dedicated to secondary server
+     static int nbPoolsServer2 ;  //!< Number of pools created on the secondary server
      static double bufferSizeFactor; //!< Factor used to tune the buffer size
      static const double defaultBufferSizeFactor; //!< Default factor value

XIOS/dev/XIOS_DEV_CMIP6/src/node/grid.hpp

@@ -203 +203 @@
          CArray<int, 1> storeIndex_client;
 
-         std::map<CContextClient*, map<int, CArray<int, 1> > > storeIndex_toSrv; // Same grid but can be sent to several pools
+/** Map containing indexes that will be sent in sendIndex(). In future: change the key to pair<distrType, serverSize> (?) */
+         std::map<CContextClient*, map<int, CArray<int, 1> > > storeIndex_toSrv;
+
          map<int, CArray<int, 1> > storeIndex_fromSrv; // Support, for now, reading with level-1 server
+
+
          std::map<CContextClient*, std::map<int,int> > nbSenders, nbReadSenders;
 
@@ -215 +219 @@
          map<int, CArray<size_t, 1> > outLocalIndexStoreOnClient; 
 
-         // If a client wants to write out data, it'll use this index.
-         // A client receives global index of data from other clients (via recvIndex),
-         // then does remapping these index into local index to WRITE into a file
-         CArray<size_t,1> localIndexToWriteOnClient, localIndexToWriteOnServer;
+/** Indexes calculated based on server distribution (serverDistribution_). They are used for writing data into a file. */
+         CArray<size_t,1> localIndexToWriteOnServer;
+
+/** Indexes calculated based on client distribution (clientDistribution_). They are not used at all.
+    They should be the same as localIndexToWriteOnServer and potentially can be used as an additional check.*/
+         CArray<size_t,1> localIndexToWriteOnClient;
 
          CArray<size_t,1> indexFromClients;
@@ -286 +292 @@
         bool isAxisListSet, isDomListSet, isScalarListSet;
 
+/** Distribution calculated in computeClientIndex() based on the knowledge of the entire grid */
         CDistributionClient* clientDistribution_;
+
+/** Distribution calculated upon receiving indexes */
         CDistributionServer* serverDistribution_;
+
         CClientServerMapping* clientServerMap_;
         size_t writtenDataSize_;
         int numberWrittenIndexes_, totalNumberWrittenIndexes_, offsetWrittenIndexes_;
+
+/** Map storing ranks of connected servers. In future: change the key to the server size (?) */
+        std::map<CContextClient*, std::vector<int> > connectedServerRank_;
+
+/** Map storing data size that will be sent to connected servers. In future: change the key to the server size (?) */
         std::map<CContextClient*, std::map<int,size_t> > connectedDataSize_;
-        std::map<CContextClient*, std::vector<int> > connectedServerRank_;
+
         bool isDataDistributed_;        
          //! True if and only if the data defined on the grid can be outputted in a compressed way
@@ -305 +320 @@
         std::map<CGrid*, std::pair<bool,StdString> > gridSrc_;
         bool hasTransform_;
+
+/** Map storing data size that will be sent to connected servers. In future: change the key to the server size (?) */
         std::map<CContextClient*, CClientServerMapping::GlobalIndexMap> globalIndexOnServer_;
-            // List order of axis and domain in a grid, if there is a domain, it will take value 1 (true), axis 0 (false)
+
+/** List order of axis and domain in a grid, if there is a domain, it will take value 1 (true), axis 0 (false) */
         std::vector<int> order_;
+
    }; // class CGrid
 

XIOS/dev/XIOS_DEV_CMIP6/src/server.cpp

@@ -110 +110 @@
           else
           {
+            int firstSndSrvRank = srvRanks.size()*(100.-CXios::ratioServer2)/100. ;
+            int poolLeader = firstSndSrvRank;
+//*********** (1) Comment out the line below to set one process per pool
+//            sndServerGlobalRanks.push_back(srvRanks[poolLeader]);
+            int nbPools = CXios::nbPoolsServer2;
+            if ( nbPools > reqNbProc || nbPools < 1)
+            {
+              error(0)<<"WARNING: void CServer::initialize(void)"<<endl
+                  << "It is impossible to allocate the requested number of pools = "<<nbPools
+                  <<" on the secondary server. It will be set so that there is one process per pool."<<endl;
+              nbPools = reqNbProc;
+            }
+            int remainder = ((int) (srvRanks.size()*CXios::ratioServer2/100.)) % nbPools;
+            int procsPerPool = ((int) (srvRanks.size()*CXios::ratioServer2/100.)) / nbPools;
             for (i=0; i<srvRanks.size(); i++)
             {
-              if (i >= srvRanks.size()*(100.-CXios::ratioServer2)/100.)
+              if (i >= firstSndSrvRank)
               {
+                if (rank_ == srvRanks[i])
+                {
+                  serverLevel=2;
+                }
+                poolLeader += procsPerPool;
+                if (remainder != 0)
+                {
+                  ++poolLeader;
+                  --remainder;
+                }
+//*********** (2) Comment out the two lines below to set one process per pool
+//                if (poolLeader < srvRanks.size())
+//                  sndServerGlobalRanks.push_back(srvRanks[poolLeader]);
+//*********** (3) Uncomment the line below to set one process per pool
                 sndServerGlobalRanks.push_back(srvRanks[i]);
-                if (rank_ == srvRanks[i]) serverLevel=2;
               }
               else
@@ -122 +149 @@
               }
             }
-          }
-        }
+            if (serverLevel==2)
+            {
+              info(50)<<"The number of secondary server pools is "<< sndServerGlobalRanks.size() <<endl ;
+              for (i=0; i<sndServerGlobalRanks.size(); i++)
+              {
+                if (rank_>= sndServerGlobalRanks[i])
+                {
+                  if ( i == sndServerGlobalRanks.size()-1)
+                  {
+                    myColor = colors.size() + sndServerGlobalRanks[i];
+                  }
+                  else if (rank_< sndServerGlobalRanks[i+1])
+                  {
+                    myColor = colors.size() + sndServerGlobalRanks[i];
+                    break;
+                  }
+                }
+              }
+            }
+          }
+        }
+
         // (2) Create intraComm
-        myColor = (serverLevel == 2) ? rank_ : colors[hashServer];
+        if (serverLevel != 2) myColor=colors[hashServer];
         MPI_Comm_split(CXios::globalComm, myColor, rank_, &intraComm) ;
 
@@ -232 +279 @@
           else
           {
+            int firstSndSrvRank = size*(100.-CXios::ratioServer2)/100. ;
+            int poolLeader = firstSndSrvRank;
+//*********** (1) Comment out the line below to set one process per pool
+//            sndServerGlobalRanks.push_back(srvGlobalRanks[poolLeader]);
+            int nbPools = CXios::nbPoolsServer2;
+            if ( nbPools > reqNbProc || nbPools < 1)
+            {
+              error(0)<<"WARNING: void CServer::initialize(void)"<<endl
+                  << "It is impossible to allocate the requested number of pools = "<<nbPools
+                  <<" on the secondary server. It will be set so that there is one process per pool."<<endl;
+              nbPools = reqNbProc;
+            }
+            int remainder = ((int) (size*CXios::ratioServer2/100.)) % nbPools;
+            int procsPerPool = ((int) (size*CXios::ratioServer2/100.)) / nbPools;
             for (int i=0; i<size; i++)
             {
-              if (i >= size*(100.-CXios::ratioServer2)/100.)
+              if (i >= firstSndSrvRank)
               {
+                if (globalRank == srvGlobalRanks[i])
+                {
+                  serverLevel=2;
+                }
+                poolLeader += procsPerPool;
+                if (remainder != 0)
+                {
+                  ++poolLeader;
+                  --remainder;
+                }
+//*********** (2) Comment out the two lines below to set one process per pool
+//                if (poolLeader < size)
+//                  sndServerGlobalRanks.push_back(srvGlobalRanks[poolLeader]);
+//*********** (3) Uncomment the line below to set one process per pool
                 sndServerGlobalRanks.push_back(srvGlobalRanks[i]);
-                if (globalRank == srvGlobalRanks[i]) serverLevel=2;
               }
               else
@@ -244 +318 @@
               }
             }
-            myColor = (serverLevel == 2) ? globalRank : 0;
+            if (serverLevel==2)
+            {
+              info(50)<<"The number of secondary server pools is "<< sndServerGlobalRanks.size() <<endl ;
+              for (int i=0; i<sndServerGlobalRanks.size(); i++)
+              {
+                if (globalRank>= sndServerGlobalRanks[i])
+                {
+                  if (i == sndServerGlobalRanks.size()-1)
+                  {
+                    myColor = sndServerGlobalRanks[i];
+                  }
+                  else if (globalRank< sndServerGlobalRanks[i+1])
+                  {
+                    myColor = sndServerGlobalRanks[i];
+                    break;
+                  }
+                }
+              }
+            }
+            if (serverLevel != 2) myColor=0;
             MPI_Comm_split(localComm, myColor, rank_, &intraComm) ;
           }
@@ -270 +363 @@
 
 //      (3) Create interComms between primary and secondary servers
-        MPI_Comm_size(intraComm,&size) ;
+        int intraCommSize, intraCommRank ;
+        MPI_Comm_size(intraComm,&intraCommSize) ;
+        MPI_Comm_rank(intraComm, &intraCommRank) ;
+
         if (serverLevel == 1)
         {
@@ -276 +372 @@
           {
             int srvSndLeader = sndServerGlobalRanks[i];
-            info(50)<<"intercommCreate::client (server level 1) "<<globalRank<<" intraCommSize : "<<size
-                <<" intraCommRank :"<<rank_<<"  clientLeader "<< srvSndLeader<<endl ;
+            info(50)<<"intercommCreate::client (server level 1) "<<globalRank<<" intraCommSize : "<<intraCommSize
+                <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvSndLeader<<endl ;
             MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvSndLeader, 0, &newComm) ;
             interCommRight.push_back(newComm) ;
@@ -284 +380 @@
         else if (serverLevel == 2)
         {
-          info(50)<<"intercommCreate::server "<<rank_<<" intraCommSize : "<<size
-                   <<" intraCommRank :"<<rank_<<"  clientLeader "<< srvGlobalRanks[0] <<endl ;
+          info(50)<<"intercommCreate::server "<<globalRank<<" intraCommSize : "<<intraCommSize
+                   <<" intraCommRank :"<<intraCommRank<<"  clientLeader "<< srvGlobalRanks[0] <<endl ;
           MPI_Intercomm_create(intraComm, 0, CXios::globalComm, srvGlobalRanks[0], 0, &newComm) ;
           interCommLeft.push_back(newComm) ;

XIOS/dev/XIOS_DEV_CMIP6/src/server.hpp

@@ -46 +46 @@
 
       public:
-        //! Get rank of the current process
+        //! Get rank of the current process in the intraComm
         static int getRank();
 
@@ -67 +67 @@
 
       private:
-        static vector<int> sndServerGlobalRanks;  //!< Global ranks of secondary server processes
-        static int rank_;                   //!< If (!oasis) global rank, else rank in the intraComm returned by oasis
-        static int nbContexts;              //!< Number of contexts registered by server
+        static vector<int> sndServerGlobalRanks;  //!< Global ranks of pool leaders on the secondary server
+        static int rank_;                         //!< If (!oasis) global rank, else rank in the intraComm returned by oasis
+        static int nbContexts;                    //!< Number of contexts registered by server
         static StdOFStream m_infoStream;
         static StdOFStream m_errorStream;