Merge pull request #212 from C0nsultant/topic/improve_ioqueue_utiliza…

…tion Minimize number of flushes in backend
openPMD · Jun 7, 2018 · aca17f8 · aca17f8
2 parents 9ec149f + 728d51f
commit aca17f8
Show file tree

Hide file tree

Showing 16 changed files with 236 additions and 191 deletions.
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -20,6 +20,11 @@ Bug Fixes
 - SerialIOTest: ``loadChunk`` template missing for ADIOS1 #227
 - prepare running serial applications linked against parallel ADIOS1 library #228
 
+Other
+"""""
+
+- minimize number of flushes in backend #212
+
 
 0.1.0-alpha
 -----------

diff --git a/include/openPMD/RecordComponent.hpp b/include/openPMD/RecordComponent.hpp
@@ -160,8 +160,7 @@ RecordComponent::loadChunk(Offset const& o, Extent const& e, std::shared_ptr< T
         dRead.extent = e;
         dRead.dtype = getDatatype();
         dRead.data = std::static_pointer_cast< void >(data);
-        IOHandler->enqueue(IOTask(this, dRead));
-        IOHandler->flush();
+        m_chunks->push(IOTask(this, dRead));
     }
 }
 

diff --git a/include/openPMD/backend/Container.hpp b/include/openPMD/backend/Container.hpp
@@ -257,7 +257,6 @@ class Container : public Attributable
             Parameter< Operation::CREATE_PATH > pCreate;
             pCreate.path = path;
             IOHandler->enqueue(IOTask(this, pCreate));
-            IOHandler->flush();
         }
 
         flushAttributes();

diff --git a/include/openPMD/backend/PatchRecordComponent.hpp b/include/openPMD/backend/PatchRecordComponent.hpp
@@ -95,8 +95,7 @@ PatchRecordComponent::load(uint64_t idx, std::shared_ptr< T > data)
     dRead.extent = {1};
     dRead.dtype = getDatatype();
     dRead.data = std::static_pointer_cast< void >(data);
-    IOHandler->enqueue(IOTask(this, dRead));
-    IOHandler->flush();
+    m_chunks->push(IOTask(this, dRead));
 }
 
 template< typename T >

diff --git a/src/Iteration.cpp b/src/Iteration.cpp
@@ -92,37 +92,31 @@ Iteration::flushFileBased(uint64_t i)
         Parameter< Operation::CREATE_FILE > fCreate;
         fCreate.name = auxiliary::replace_first(s->iterationFormat(), "%T", std::to_string(i));
         IOHandler->enqueue(IOTask(s, fCreate));
-        IOHandler->flush();
 
         /* create basePath */
         Parameter< Operation::CREATE_PATH > pCreate;
         pCreate.path = auxiliary::replace_first(s->basePath(), "%T/", "");
         IOHandler->enqueue(IOTask(&s->iterations, pCreate));
-        IOHandler->flush();
 
         /* create iteration path */
         pCreate.path = std::to_string(i);
         IOHandler->enqueue(IOTask(this, pCreate));
-        IOHandler->flush();
     } else
     {
         /* open file */
         Series* s = dynamic_cast<Series *>(parent->attributable->parent->attributable);
         Parameter< Operation::OPEN_FILE > fOpen;
         fOpen.name = auxiliary::replace_last(s->iterationFormat(), "%T", std::to_string(i));
         IOHandler->enqueue(IOTask(s, fOpen));
-        IOHandler->flush();
 
         /* open basePath */
         Parameter< Operation::OPEN_PATH > pOpen;
         pOpen.path = auxiliary::replace_first(s->basePath(), "%T/", "");
         IOHandler->enqueue(IOTask(&s->iterations, pOpen));
-        IOHandler->flush();
 
         /* open iteration path */
         pOpen.path = std::to_string(i);
         IOHandler->enqueue(IOTask(this, pOpen));
-        IOHandler->flush();
     }
 
     flush();
@@ -137,7 +131,6 @@ Iteration::flushGroupBased(uint64_t i)
         Parameter< Operation::CREATE_PATH > pCreate;
         pCreate.path = std::to_string(i);
         IOHandler->enqueue(IOTask(this, pCreate));
-        IOHandler->flush();
     }
 
     flush();
@@ -146,34 +139,43 @@ Iteration::flushGroupBased(uint64_t i)
 void
 Iteration::flush()
 {
-    /* Find the root point [Series] of this file,
-     * meshesPath and particlesPath are stored there */
-    Writable *w = this->parent;
-    while( w->parent )
-        w = w->parent;
-    Series* s = dynamic_cast<Series *>(w->attributable);
-
-    if( !meshes.empty() )
+    if( IOHandler->accessType == AccessType::READ_ONLY )
     {
-        if( !s->containsAttribute("meshesPath") )
-            s->setMeshesPath("meshes/");
-        s->flushMeshesPath();
-        meshes.flush(s->meshesPath());
         for( auto& m : meshes )
             m.second.flush(m.first);
-    }
-
-    if( !particles.empty() )
-    {
-        if( !s->containsAttribute("particlesPath") )
-            s->setParticlesPath("particles/");
-        s->flushParticlesPath();
-        particles.flush(s->particlesPath());
         for( auto& species : particles )
             species.second.flush(species.first);
-    }
+    } else
+    {
+        /* Find the root point [Series] of this file,
+         * meshesPath and particlesPath are stored there */
+        Writable *w = this->parent;
+        while( w->parent )
+            w = w->parent;
+        Series* s = dynamic_cast<Series *>(w->attributable);
+
+        if( !meshes.empty() )
+        {
+            if( !s->containsAttribute("meshesPath") )
+                s->setMeshesPath("meshes/");
+            s->flushMeshesPath();
+            meshes.flush(s->meshesPath());
+            for( auto& m : meshes )
+                m.second.flush(m.first);
+        }
 
-    flushAttributes();
+        if( !particles.empty() )
+        {
+            if( !s->containsAttribute("particlesPath") )
+                s->setParticlesPath("particles/");
+            s->flushParticlesPath();
+            particles.flush(s->particlesPath());
+            for( auto& species : particles )
+                species.second.flush(species.first);
+        }
+
+        flushAttributes();
+    }
 }
 
 void
@@ -252,7 +254,6 @@ Iteration::read()
         Parameter< Operation::OPEN_PATH > pOpen;
         pOpen.path = s->meshesPath();
         IOHandler->enqueue(IOTask(&meshes, pOpen));
-        IOHandler->flush();
 
         meshes.readAttributes();
 
@@ -313,7 +314,6 @@ Iteration::read()
         Parameter< Operation::OPEN_PATH > pOpen;
         pOpen.path = s->particlesPath();
         IOHandler->enqueue(IOTask(&particles, pOpen));
-        IOHandler->flush();
 
         particles.readAttributes();
 

diff --git a/src/Mesh.cpp b/src/Mesh.cpp
@@ -182,32 +182,38 @@ Mesh::setTimeOffset(T to)
 void
 Mesh::flush(std::string const& name)
 {
-    if( !written )
+    if( IOHandler->accessType == AccessType::READ_ONLY )
     {
-        if( *m_containsScalar )
-        {
-            MeshRecordComponent& r = at(RecordComponent::SCALAR);
-            r.m_writable->parent = parent;
-            r.parent = parent;
-            r.flush(name);
-            m_writable->abstractFilePosition = r.m_writable->abstractFilePosition;
-            abstractFilePosition = r.abstractFilePosition;
-            written = true;
-        } else
+        for( auto& comp : *this )
+            comp.second.flush(comp.first);
+    } else
+    {
+        if( !written )
         {
-            Parameter< Operation::CREATE_PATH > pCreate;
-            pCreate.path = name;
-            IOHandler->enqueue(IOTask(this, pCreate));
-            IOHandler->flush();
-            for( auto& comp : *this )
-                comp.second.parent = this->m_writable.get();
+            if( *m_containsScalar )
+            {
+                MeshRecordComponent& r = at(RecordComponent::SCALAR);
+                r.m_writable->parent = parent;
+                r.parent = parent;
+                r.flush(name);
+                m_writable->abstractFilePosition = r.m_writable->abstractFilePosition;
+                abstractFilePosition = r.abstractFilePosition;
+                written = true;
+            } else
+            {
+                Parameter< Operation::CREATE_PATH > pCreate;
+                pCreate.path = name;
+                IOHandler->enqueue(IOTask(this, pCreate));
+                for( auto& comp : *this )
+                    comp.second.parent = this->m_writable.get();
+            }
         }
-    }
 
-    for( auto& comp : *this )
-        comp.second.flush(comp.first);
+        for( auto& comp : *this )
+            comp.second.flush(comp.first);
 
-    flushAttributes();
+        flushAttributes();
+    }
 }
 
 void
@@ -315,7 +321,6 @@ Mesh::read()
             MeshRecordComponent& rc = (*this)[component];
             pOpen.path = component;
             IOHandler->enqueue(IOTask(&rc, pOpen));
-            IOHandler->flush();
             *rc.m_isConstant = true;
             rc.read();
         }

diff --git a/src/ParticlePatches.cpp b/src/ParticlePatches.cpp
@@ -47,7 +47,6 @@ ParticlePatches::read()
         PatchRecord& pr = (*this)[record_name];
         pOpen.path = record_name;
         IOHandler->enqueue(IOTask(&pr, pOpen));
-        IOHandler->flush();
         pr.read();
     }
 

diff --git a/src/ParticleSpecies.cpp b/src/ParticleSpecies.cpp
@@ -48,7 +48,6 @@ ParticleSpecies::read()
         {
             pOpen.path = "particlePatches";
             IOHandler->enqueue(IOTask(&particlePatches, pOpen));
-            IOHandler->flush();
             particlePatches.read();
         } else
         {
@@ -107,18 +106,27 @@ ParticleSpecies::read()
 void
 ParticleSpecies::flush(std::string const& path)
 {
-    Container< Record >::flush(path);
-
-    for( auto& record : *this )
-        record.second.flush(record.first);
-
-    if( particlePatches.find("numParticles") != particlePatches.end()
-        && particlePatches.find("numParticlesOffset") != particlePatches.end()
-        && particlePatches.size() >= 3 )
+    if( IOHandler->accessType == AccessType::READ_ONLY )
     {
-        particlePatches.flush("particlePatches");
+        for( auto& record : *this )
+            record.second.flush(record.first);
         for( auto& patch : particlePatches )
             patch.second.flush(patch.first);
+    } else
+    {
+        Container< Record >::flush(path);
+
+        for( auto& record : *this )
+            record.second.flush(record.first);
+
+        if( particlePatches.find("numParticles") != particlePatches.end()
+            && particlePatches.find("numParticlesOffset") != particlePatches.end()
+            && particlePatches.size() >= 3 )
+        {
+            particlePatches.flush("particlePatches");
+            for( auto& patch : particlePatches )
+                patch.second.flush(patch.first);
+        }
     }
 }
 } // openPMD
diff --git a/src/Record.cpp b/src/Record.cpp
@@ -49,32 +49,38 @@ Record::setUnitDimension(std::map< UnitDimension, double > const& udim)
 void
 Record::flush(std::string const& name)
 {
-    if( !written )
+    if( IOHandler->accessType == AccessType::READ_ONLY )
     {
-        if( *m_containsScalar )
-        {
-            RecordComponent& r = at(RecordComponent::SCALAR);
-            r.m_writable->parent = parent;
-            r.parent = parent;
-            r.flush(name);
-            m_writable->abstractFilePosition = r.m_writable->abstractFilePosition;
-            abstractFilePosition = r.abstractFilePosition;
-            written = true;
-        } else
+        for( auto& comp : *this )
+            comp.second.flush(comp.first);
+    } else
+    {
+        if( !written )
         {
-            Parameter< Operation::CREATE_PATH > pCreate;
-            pCreate.path = name;
-            IOHandler->enqueue(IOTask(this, pCreate));
-            IOHandler->flush();
-            for( auto& comp : *this )
-                comp.second.parent = getWritable(this);
+            if( *m_containsScalar )
+            {
+                RecordComponent& r = at(RecordComponent::SCALAR);
+                r.m_writable->parent = parent;
+                r.parent = parent;
+                r.flush(name);
+                m_writable->abstractFilePosition = r.m_writable->abstractFilePosition;
+                abstractFilePosition = r.abstractFilePosition;
+                written = true;
+            } else
+            {
+                Parameter< Operation::CREATE_PATH > pCreate;
+                pCreate.path = name;
+                IOHandler->enqueue(IOTask(this, pCreate));
+                for( auto& comp : *this )
+                    comp.second.parent = getWritable(this);
+            }
         }
-    }
 
-    for( auto& comp : *this )
-        comp.second.flush(comp.first);
+        for( auto& comp : *this )
+            comp.second.flush(comp.first);
 
-    flushAttributes();
+        flushAttributes();
+    }
 }
 
 void
@@ -100,7 +106,6 @@ Record::read()
             RecordComponent& rc = (*this)[component];
             pOpen.path = component;
             IOHandler->enqueue(IOTask(&rc, pOpen));
-            IOHandler->flush();
             *rc.m_isConstant = true;
             rc.read();
         }