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 {
 
  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.
    // 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 {
      public:
        void wait(boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> &externLock,
                  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, 
      };
      File(const boost::filesystem::path &filename_, FileAccess type_,
           const std::function<void()> &closeRequestCallback_=std::function<void()>(),
           const std::function<void()> &refreshCallback_=std::function<void()>());
      ~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);
 
    private:
      static int defaultCompression;
      static int defaultChunkSize;
      static int defaultCacheSize;
 
      void close() override;
 
      const boost::filesystem::path filename;
      const FileAccess type;
      const std::function<void()> closeRequestCallback;
      const std::function<void()> refreshCallback;
 
      std::string shmName;
      const boost::uuids::uuid processUUID;
 
      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.
      };
 
      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 openReader();
      void closeReader();
      void openWriter();
      void closeWriter();
 
      void wait(Internal::ScopedLock &lock,
                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();
 
      void openOrCreateShm();
  };
}
 
#endif