Changeset 2547 for XIOS3/trunk/src/server.cpp

Timestamp:

08/29/23 17:24:04 (10 months ago)

Author:

ymipsl

Message:

Major update :

• New method to lock and unlock one-sided windows (window_dynamic) to avoid network overhead
• Introducing multithreading on server sided to manage more efficiently dead-lock occuring (similar to co-routine which will be available and implemented in futur c++ standard), based on c++ threads
• Suprression of old "attached mode" which is replaced by online writer and reder filters

YM

File:

• XIOS3/trunk/src/server.cpp (modified) (7 diffs)

XIOS3/trunk/src/server.cpp

@@ -25 +25 @@
 #include "workflow_graph.hpp"
 #include "release_static_allocation.hpp"
+#include "thread_manager.hpp"
 #include <sys/stat.h>
 #include <unistd.h>
@@ -68 +69 @@
       if (!CXios::usingOasis)
       {
-        if (!is_MPI_Initialized) MPI_Init(NULL, NULL);
+        if (!is_MPI_Initialized) 
+        {
+          int required = MPI_THREAD_SERIALIZED ;
+          int provided ;
+          MPI_Init_thread(NULL,NULL, required, &provided) ;
+        }
        
         // split the global communicator
@@ -99 +105 @@
       else // using OASIS
       {
-        if (!is_MPI_Initialized) driver_ = new CThirdPartyDriver();
+        
+        if (!is_MPI_Initialized) 
+        {
+          int required = MPI_THREAD_SERIALIZED ;
+          int provided ;
+          MPI_Init_thread(NULL,NULL, required, &provided) ;
+        }
+
+        driver_ = new CThirdPartyDriver();
 
         driver_->getComponentCommunicator( serverComm );
@@ -200 +214 @@
           {
             ressourcesManager->createPool(CXios::defaultPoolId, nbRessources) ;
-            ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!ressourcesManager->hasPool(CXios::defaultPoolId)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+          
             servicesManager->createServices(CXios::defaultPoolId, CXios::defaultWriterId, CServicesManager::WRITER,nbRessources,1) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultWriterId,0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            
             servicesManager->createServicesOnto(CXios::defaultPoolId, CXios::defaultReaderId, CServicesManager::READER, CXios::defaultWriterId) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            if (CThreadManager::isUsingThreads()) 
+            {
+              daemonsManager->eventLoop() ;
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultReaderId, 0)) CThreadManager::yield() ;
+            }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
           }
           else
@@ -214 +247 @@
             if (nbPoolsServer2 == 0) nbPoolsServer2 = nprocsServer;
             ressourcesManager->createPool(CXios::defaultPoolId, nbRessources) ;
-            ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!ressourcesManager->hasPool(CXios::defaultPoolId)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else ressourcesManager->waitPoolRegistration(CXios::defaultPoolId) ;
+
             servicesManager->createServices(CXios::defaultPoolId,  CXios::defaultGathererId, CServicesManager::GATHERER, nprocsGatherer, 1) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultGathererId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultGathererId,0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultGathererId) ;
+
             servicesManager->createServicesOnto(CXios::defaultPoolId, CXios::defaultReaderId, CServicesManager::READER, CXios::defaultGathererId) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            if (CThreadManager::isUsingThreads()) 
+              while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultReaderId, 0)) 
+              {
+                daemonsManager->eventLoop() ;
+                CThreadManager::yield() ;
+              }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultReaderId) ;
+            
             servicesManager->createServices(CXios::defaultPoolId,  CXios::defaultWriterId, CServicesManager::WRITER, nprocsServer, nbPoolsServer2) ;
-            servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
+            if (CThreadManager::isUsingThreads())
+              for(int i=0; i<nbPoolsServer2; i++)
+                while(!servicesManager->hasService(CXios::defaultPoolId, CXios::defaultWriterId,i)) 
+                {
+                  daemonsManager->eventLoop() ;
+                  CThreadManager::yield() ;
+                }
+            else servicesManager->waitServiceRegistration(CXios::defaultPoolId, CXios::defaultWriterId) ;
           }
         }
@@ -233 +294 @@
       {
         daemonsManager->eventLoop() ;
+        if (CThreadManager::isUsingThreads()) CThreadManager::yield();
         MPI_Test(&req,&ok,&status) ;
       }
 
 
-      testingEventScheduler() ;
+      //testingEventScheduler() ;
 /*
       MPI_Request req ;
@@ -262 +324 @@
       {
         finished=daemonsManager->eventLoop() ;
+        if (CThreadManager::isUsingThreads()) CThreadManager::yield() ;
       }
       CTimer::get("XIOS event loop").suspend() ;