Merge pull request #628 from dr-robertk/PR/async-output

Pr/async output

CMakeLists_files.cmake

@@ -49,6 +49,7 @@ list (APPEND MAIN_SOURCE_FILES
   opm/autodiff/WellMultiSegment.cpp
   opm/autodiff/StandardWells.cpp
   opm/autodiff/BlackoilSolventState.cpp
+  opm/autodiff/ThreadHandle.hpp
   opm/polymer/PolymerState.cpp
   opm/polymer/PolymerBlackoilState.cpp
   opm/polymer/CompressibleTpfaPolymer.cpp

opm/autodiff/SimulatorFullyImplicitBlackoilOutput.cpp

@@ -1,6 +1,6 @@
 /*
   Copyright (c) 2014 SINTEF ICT, Applied Mathematics.
-  Copyright (c) 2015 IRIS AS
+  Copyright (c) 2015-2016 IRIS AS
 
   This file is part of the Open Porous Media project (OPM).
 
@@ -253,6 +253,40 @@ namespace Opm
         }
     }
 
+
+    namespace detail {
+
+        struct WriterCall : public ThreadHandle :: ObjectInterface
+        {
+            BlackoilOutputWriter& writer_;
+            std::unique_ptr< SimulatorTimerInterface > timer_;
+            const SimulationDataContainer state_;
+            const WellState wellState_;
+            const bool substep_;
+
+            explicit WriterCall( BlackoilOutputWriter& writer,
+                                 const SimulatorTimerInterface& timer,
+                                 const SimulationDataContainer& state,
+                                 const WellState& wellState,
+                                 bool substep )
+                : writer_( writer ),
+                  timer_( timer.clone() ),
+                  state_( state ),
+                  wellState_( wellState ),
+                  substep_( substep )
+            {
+            }
+
+            // callback to writer's serial writeTimeStep method
+            void run ()
+            {
+                // write data
+                writer_.writeTimeStepSerial( *timer_, state_, wellState_, substep_ );
+            }
+        };
+    }
+
+
     void
     BlackoilOutputWriter::
     writeTimeStep(const SimulatorTimerInterface& timer,
@@ -265,7 +299,7 @@ namespace Opm
             vtkWriter_->writeTimeStep( timer, localState, localWellState, false );
         }
 
-        bool isIORank = true ;
+        bool isIORank = output_ ;
         if( parallelOutput_ && parallelOutput_->isParallel() )
         {
             // collect all solutions to I/O rank
@@ -275,15 +309,35 @@ namespace Opm
         const SimulationDataContainer& state = (parallelOutput_ && parallelOutput_->isParallel() ) ? parallelOutput_->globalReservoirState() : localState;
         const WellState& wellState  = (parallelOutput_ && parallelOutput_->isParallel() ) ? parallelOutput_->globalWellState() : localWellState;
 
-        // output is only done on I/O rank
+        // serial output is only done on I/O rank
         if( isIORank )
+        {
+            if( asyncOutput_ ) {
+                // dispatch the write call to the extra thread
+                asyncOutput_->dispatch( detail::WriterCall( *this, timer, state, wellState, substep ) );
+            }
+            else {
+                // just write the data to disk
+                writeTimeStepSerial( timer, state, wellState, substep );
+            }
+        }
+    }
+
+    void
+    BlackoilOutputWriter::
+    writeTimeStepSerial(const SimulatorTimerInterface& timer,
+                        const SimulationDataContainer& state,
+                        const WellState& wellState,
+                        bool substep)
     {
         // Matlab output
         if( matlabWriter_ ) {
             matlabWriter_->writeTimeStep( timer, state, wellState, substep );
         }
+
         // ECL output
-            if ( eclWriter_ ) {
+        if ( eclWriter_ )
+        {
             const auto initConfig = eclipseState_->getInitConfig();
             if (initConfig->getRestartInitiated() && ((initConfig->getRestartStep()) == (timer.currentStepNum()))) {
                 std::cout << "Skipping restart write in start of step " << timer.currentStepNum() << std::endl;
@@ -293,7 +347,7 @@ namespace Opm
         }
 
         // write backup file
-            if( backupfile_ )
+        if( backupfile_.is_open() )
         {
             int reportStep      = timer.reportStepNum();
             int currentTimeStep = timer.currentStepNum();
@@ -307,33 +361,19 @@ namespace Opm
                 // write resport step number
                 backupfile_.write( (const char *) &reportStep, sizeof(int) );
 
-                    /*
                 try {
-                        const BlackoilState& boState = dynamic_cast< const BlackoilState& > (state);
-                        backupfile_ << boState;
+                    backupfile_ << state;
 
                     const WellStateFullyImplicitBlackoil& boWellState = static_cast< const WellStateFullyImplicitBlackoil& > (wellState);
                     backupfile_ << boWellState;
                 }
                 catch ( const std::bad_cast& e )
                 {
-
                 }
-                    */
 
-                    /*
-                    const WellStateFullyImplicitBlackoil* boWellState =
-                        dynamic_cast< const WellStateFullyImplicitBlackoil* > (&wellState);
-                    if( boWellState ) {
-                        backupfile_ << (*boWellState);
-                    }
-                    else
-                        OPM_THROW(std::logic_error,"cast to WellStateFullyImplicitBlackoil failed");
-                    */
                 backupfile_ << std::flush;
             }
         } // end backup
-        } // end isIORank
     }
 
     void

opm/autodiff/SimulatorFullyImplicitBlackoilOutput.hpp

@@ -34,6 +34,7 @@
 #include <opm/autodiff/ParallelDebugOutput.hpp>
 
 #include <opm/autodiff/WellStateFullyImplicitBlackoil.hpp>
+#include <opm/autodiff/ThreadHandle.hpp>
 
 #include <opm/parser/eclipse/EclipseState/EclipseState.hpp>
 #include <opm/parser/eclipse/EclipseState/InitConfig/InitConfig.hpp>
@@ -43,6 +44,7 @@
 #include <sstream>
 #include <iomanip>
 #include <fstream>
+#include <thread>
 
 #include <boost/filesystem.hpp>
 
@@ -220,6 +222,12 @@ namespace Opm
                            const Opm::WellState& wellState,
                            bool substep = false);
 
+        /** \copydoc Opm::OutputWriter::writeTimeStep */
+        void writeTimeStepSerial(const SimulatorTimerInterface& timer,
+                                 const SimulationDataContainer& reservoirState,
+                                 const Opm::WellState& wellState,
+                                 bool substep);
+
         /** \brief return output directory */
         const std::string& outputDirectory() const { return outputDir_; }
 
@@ -257,6 +265,8 @@ namespace Opm
         std::unique_ptr< OutputWriter  > matlabWriter_;
         std::unique_ptr< EclipseWriter > eclWriter_;
         EclipseStateConstPtr eclipseState_;
+
+        std::unique_ptr< ThreadHandle > asyncOutput_;
     };
 
 
@@ -289,8 +299,15 @@ namespace Opm
                                       parallelOutput_->numCells(),
                                       parallelOutput_->globalCell() )
                    : 0 ),
-        eclipseState_(eclipseState)
+        eclipseState_(eclipseState),
+        asyncOutput_()
     {
+        // create output thread if needed
+        if( output_ && param.getDefault("async_output", bool( false ) ) )
+        {
+            asyncOutput_.reset( new ThreadHandle() );
+        }
+
         // For output.
         if (output_ && parallelOutput_->isIORank() ) {
             // Ensure that output dir exists

opm/autodiff/ThreadHandle.hpp

@@ -0,0 +1,158 @@
+#ifndef OPM_THREADHANDLE_HPP
+#define OPM_THREADHANDLE_HPP
+
+#include <cassert>
+#include <dune/common/exceptions.hh>
+
+#include <thread>
+#include <mutex>
+#include <queue>
+
+namespace Opm
+{
+
+  class ThreadHandle
+  {
+  public:
+    class ObjectInterface
+    {
+    protected:
+      ObjectInterface() {}
+    public:
+      virtual ~ObjectInterface() {}
+      virtual void run() = 0;
+      virtual bool isEndMarker () const { return false; }
+    };
+
+    template <class Object>
+    class ObjectWrapper : public ObjectInterface
+    {
+      Object obj_;
+    public:
+      ObjectWrapper( Object&& obj ) : obj_( std::move( obj ) ) {}
+      void run() { obj_.run(); }
+    };
+
+  protected:
+    class EndObject : public ObjectInterface
+    {
+    public:
+      void run () { }
+      bool isEndMarker () const { return true; }
+    };
+
+    ////////////////////////////////////////////
+    // class ThreadHandleQueue
+    ////////////////////////////////////////////
+    class ThreadHandleQueue
+    {
+    protected:
+      std::queue< std::unique_ptr< ObjectInterface > > objQueue_;
+      std::mutex  mutex_;
+
+      // no copying
+      ThreadHandleQueue( const ThreadHandleQueue& ) = delete;
+
+    public:
+      //! constructor creating object that is executed by thread
+      ThreadHandleQueue()
+        : objQueue_(), mutex_()
+      {
+      }
+
+      //! insert object into threads queue
+      void push_back( std::unique_ptr< ObjectInterface >&& obj )
+      {
+        // lock mutex to make sure objPtr is not used
+        mutex_.lock();
+        objQueue_.emplace( std::move(obj) );
+        mutex_.unlock();
+      }
+
+      //! do the work until the queue received an end object
+      void run()
+      {
+        // wait until objects have been pushed to the queue
+        while( objQueue_.empty() )
+        {
+          // sleep one second
+          std::this_thread::sleep_for( std::chrono::seconds(1) );
+        }
+
+        {
+            // lock mutex for access to objQueue_
+            mutex_.lock();
+
+            // get next object from queue
+            std::unique_ptr< ObjectInterface > obj( objQueue_.front().release() );
+            // remove object from queue
+            objQueue_.pop();
+
+            // unlock mutex for access to objQueue_
+            mutex_.unlock();
+
+            // if object is end marker terminate thread
+            if( obj->isEndMarker() ){
+                if( ! objQueue_.empty() ) {
+                    OPM_THROW(std::logic_error,"ThreadHandleQueue: not all queued objects were executed");
+                }
+                return;
+            }
+
+            // execute object action
+            obj->run();
+        }
+
+        // keep thread running
+        run();
+      }
+    }; // end ThreadHandleQueue
+
+    ////////////////////////////////////////////////////
+    //  end ThreadHandleQueue
+    ////////////////////////////////////////////////////
+
+    // start the thread by calling method run
+    static void startThread( ThreadHandleQueue* obj )
+    {
+       obj->run();
+    }
+
+    ThreadHandleQueue threadObjectQueue_;
+    std::thread thread_;
+
+  private:
+    // prohibit copying
+    ThreadHandle( const ThreadHandle& ) = delete;
+
+  public:
+    //! default constructor
+    ThreadHandle()
+      : threadObjectQueue_(),
+        thread_( startThread, &threadObjectQueue_ )
+    {
+      // detach thread into nirvana
+      thread_.detach();
+    } // end constructor
+
+    //! dispatch object to queue of separate thread
+    template <class Object>
+    void dispatch( Object&& obj )
+    {
+      typedef ObjectWrapper< Object >  ObjectPointer;
+      ObjectInterface* objPtr = new ObjectPointer( std::move(obj) );
+
+      // add object to queue of objects
+      threadObjectQueue_.push_back( std::unique_ptr< ObjectInterface > (objPtr) );
+    }
+
+    //! destructor terminating the thread
+    ~ThreadHandle()
+    {
+      // dispatch end object which will terminate the thread
+      threadObjectQueue_.push_back( std::unique_ptr< ObjectInterface > (new EndObject()) ) ;
+    }
+  };
+
+} // end namespace Opm
+#endif