file.h

/* Copyright (C) 2009 Markus Friedrich
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
 *
 * Contact:
 *   friedrich.at.gc@googlemail.com
 *
 */
 
#ifndef _HDF5SERIE_FILE_H_
#define _HDF5SERIE_FILE_H_
 
#include <hdf5serie/group.h>
#include <boost/filesystem.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
#ifdef _WIN32
  #include <boost/interprocess/windows_shared_memory.hpp>
#else
  #include <boost/interprocess/shared_memory_object.hpp>
#endif
#include <boost/interprocess/mapped_region.hpp>
#include <boost/container/static_vector.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <thread>
#include <boost/thread/thread.hpp>
 
namespace H5 {
 
  class File;
 
  namespace Internal {
#ifdef _WIN32
    using SharedMemory = boost::interprocess::windows_shared_memory;
    inline void SharedMemoryRemove(const char* shmName) {
    }
    inline SharedMemory SharedMemoryCreate(const char *shmName, boost::interprocess::mode_t mode,
                                 size_t size) {
      auto shm=SharedMemory(boost::interprocess::create_only, shmName, mode, size);
      return shm;
    }
#else
    using SharedMemory = boost::interprocess::shared_memory_object;
    inline void SharedMemoryRemove(const char* shmName) {
      SharedMemory::remove(shmName);
    }
    inline SharedMemory SharedMemoryCreate(const char *shmName, boost::interprocess::mode_t mode,
                                 size_t size) {
      auto shm=SharedMemory(boost::interprocess::create_only, shmName, mode);
      shm.truncate(size);
      return shm;
    }
#endif
 
    class ScopedLock;
 
    // A simple robust condition variable,
    // since boost::interprocess::interprocess:condition does not provide a robust condition on Linux.
    // A none robust condition variable is blocking in notify_all or dtor if a process waiting on the condition crashed!
    // The robustness is simply achived by polling in constant time intervalls instead of putting thread to sleep and wakeup.
    // This polling delay is no problem in this scope since its only used for user visible things.
    // So a polling delay of 1/25 second (25 frames per second) is used which equal the human visible reaction time.
    // The interface equals boost::interprocess::interprocess_condition but not with all member functions.
    // However, only N threads (which may be from different processes) can be waiting.
    // This class must be implemented in a address-free way since it is placed usually in shared memory.
    // The implementation is similar to
    // https://cseweb.ucsd.edu/classes/sp17/cse120-a/applications/ln/lecture7.html
    template<int N>
    class ConditionVariable {
      friend class H5::File; // to allow File::dumpSharedMemory to access the private members
      public:
        void wait(boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> &externLock,
                  const std::function<bool()> &pred);
        bool wait_for(boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> &externLock,
                      const std::chrono::milliseconds& relTime,
                      const std::function<bool()> &pred);
        void notify_all();
      private:
        boost::container::static_vector<boost::uuids::uuid, N> waiter;
        boost::interprocess::interprocess_mutex waiterMutex;
    };
  }
 
  class Dataset;
 
  /* A wrapper around a HDF5 file.
   * It handles automatically HDF5 SWMR access using a shared memory based inter-process communication:
   * This ensure that:
   * - only one writer is active (more writers are blocked until the writer closes the file)
   * - readers are blocked until no writer is active or the writer is in SWMR mode
   * - if readers are active and a writer wants to start writing the readers are notified by a reopen request.
   */
  class File : public GroupBase {
    friend class Internal::ScopedLock;
    public:
      enum FileAccess {
        read,            
        write,           
        writeWithRename, 
      };
      File(const boost::filesystem::path &filename_, FileAccess type_,
           const std::function<void()> &closeRequestCallback_={},
           const std::function<void()> &refreshCallback_={},
           const std::function<void()> &renameAtomicFunc_={});
      ~File() override;
      void enableSWMR() override;
 
      static int getDefaultCompression() { return defaultCompression; }
      static void setDefaultCompression(int comp) { defaultCompression=comp; }
      static int getDefaultChunkSize() { return defaultChunkSize; }
      static void setDefaultChunkSize(int chunk) { defaultChunkSize=chunk; }
      static int getDefaultCacheSize() { return defaultCacheSize; }
      static void setDefaultCacheSize(int cache) { defaultCacheSize=cache; }
 
      void refresh() override;
      void requestFlush();
      void flushIfRequested();
 
      static void dumpSharedMemory(const boost::filesystem::path &filename);
      static void removeSharedMemory(const boost::filesystem::path &filename);
 
      FileAccess getType(bool originalType=false); // returns write for write and writeWithRename and read for read
 
    private:
      static int defaultCompression;
      static int defaultChunkSize;
      static int defaultCacheSize;
 
      void close() override;
 
      boost::filesystem::path filename;
      boost::filesystem::path getFilename(bool originalFilename=false); // gets the filename dependent on the current preSWMR
 
      FileAccess type;
      bool preSWMR { false };
 
      enum class WriterState {
        none,         //<! no writer is currently active
        writeRequest, //<! a writer wants to write the file but readers still exists
        active,       //<! a writer exists and is currently in dateset/attribute creation mode
        swmr,         //<! a writer exists and is currently in SWMR mode (readers can use the file using SWMR)
      };
      constexpr static size_t MAXREADERS { 100 };
      constexpr static size_t MAXWRITERS { 10 };
      struct ProcessInfo {
        boost::uuids::uuid processUUID;         
        boost::posix_time::ptime lastAliveTime; 
        FileAccess type;                        
      };
      struct SharedMemObject {
        // the following member is only used for life-time handling of the shared memroy object itself
        size_t shmUseCount { 0 }; //<! the number users of this shared memory object
        // the following members are used to synchronize the writer and all readers.
        boost::interprocess::interprocess_mutex mutex;    //<! mutex for synchronization handling.
        Internal::ConditionVariable<MAXREADERS+MAXWRITERS> cond; //<! a condition variable for signaling state changes.
        // the following members represent the state of the writer and readers
        // after setting any of these variables sharedData->cond.notify_all() must be called to notify all waiting process about the change
        WriterState writerState { WriterState::none }; //<! the current state of the write of this file.
        size_t activeReaders { 0 };                    //<! the number of active readers on this file.
        // the follwing members are only used for still-alive/crash detection handling
        boost::container::static_vector<ProcessInfo, MAXREADERS+1> processes; //<! a list of all processes accessing the shared memory
        // the follwing members are only used for flush/refresh handling
        bool flushRequest { false }; //<! Is set to true by reader if a flush of the writer should be done. The writer resets to false after a flush.
      };
 
      const std::function<void()> closeRequestCallback;
      const std::function<void()> refreshCallback;
      const std::function<void()> renameAtomicFunc;
 
      std::string shmName;
      const boost::uuids::uuid processUUID;
 
      Internal::SharedMemory shm;
      boost::interprocess::mapped_region region;
 
      SharedMemObject *sharedData {nullptr};
 
      bool flushRequested { false };
      WriterState lastWriterState { WriterState::none };
 
      static std::string createShmName(const boost::filesystem::path &filename);
 
      void preOpenReader();
      void openReader();
      void closeReader();
      void postCloseReader();
      void preOpenWriter();
      void openWriter();
      void closeWriter();
      void postCloseWriter();
 
      void wait(Internal::ScopedLock &lock, const std::chrono::milliseconds& relTime,
                std::string_view blockingMsg, const std::function<bool()> &pred);
 
      // still alive pings use boost::thread since interruption point of boost, which are not availabe for std::thread, are used
      boost::thread stillAlivePingThread; // the thread for still alive pings
      void stillAlivePing(); // the worker function
 
      std::thread listenForRequestThread;
      void listenForRequest();
      bool exitThread { false }; // access is object is guarded by sharedData->mutex (interprocess wide)
 
      void initProcessInfo();
      void deinitProcessInfo();
 
      static void openOrCreateShm(const boost::filesystem::path &filename, File *self,
                                  std::string &shmName, Internal::SharedMemory &shm, boost::interprocess::mapped_region &region,
                                  SharedMemObject *&sharedData);
      static void deinitShm(SharedMemObject *sharedData, const boost::filesystem::path &filename, File *self, const std::string &shmName);
  };
}
 
#endif