MMapStoreBuffer.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2026 60East Technologies Inc., All Rights Reserved.
//
// This computer software is owned by 60East Technologies Inc. and is
// protected by U.S. copyright laws and other laws and by international
// treaties.  This computer software is furnished by 60East Technologies
// Inc. pursuant to a written license agreement and may be used, copied,
// transmitted, and stored only in accordance with the terms of such
// license agreement and with the inclusion of the above copyright notice.
// This computer software or any other copies thereof may not be provided
// or otherwise made available to any other person.
//
// U.S. Government Restricted Rights.  This computer software: (a) was
// developed at private expense and is in all respects the proprietary
// information of 60East Technologies Inc.; (b) was not developed with
// government funds; (c) is a trade secret of 60East Technologies Inc.
// for all purposes of the Freedom of Information Act; and (d) is a
// commercial item and thus, pursuant to Section 12.212 of the Federal
// Acquisition Regulations (FAR) and DFAR Supplement Section 227.7202,
// Government's use, duplication or disclosure of the computer software
// is subject to the restrictions set forth by 60East Technologies Inc..
//

#ifndef _MMAPSTOREBUFFER_H_
#define _MMAPSTOREBUFFER_H_

#include <amps/ampsplusplus.hpp>
#include <amps/MemoryStoreBuffer.hpp>
#include <limits>
#include <string>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#ifdef _WIN32
  #include <windows.h>
#else
  #include <sys/mman.h>
#endif



namespace AMPS
{
  class MMapStoreBuffer : public MemoryStoreBuffer
  {
  public:
#ifdef _WIN32
    MMapStoreBuffer(const std::string& fileName_,
                    size_t initialSize_ = AMPS_MEMORYBUFFER_DEFAULT_LENGTH,
                    size_t sizeAlignment_ = 1,
                    bool useStoredSizeAlignment_ = false)
      : MemoryStoreBuffer(MemoryStoreBuffer::EMPTY)
      , _sizeAlignment(sizeAlignment_ ? sizeAlignment_ : 1)
      , _mapFile(INVALID_HANDLE_VALUE), _file(INVALID_HANDLE_VALUE)
    {
      try
      {
        if (initialSize_ == 0)
        {
          initialSize_ = AMPS_MEMORYBUFFER_DEFAULT_LENGTH;
        }
        _file = CreateFileA(fileName_.c_str(), GENERIC_READ | GENERIC_WRITE, 0,
                            NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
        if ( _file == INVALID_HANDLE_VALUE )
        {
          std::ostringstream os;
          os << "Failed to create file " << fileName_ << " for MMapStoreBuffer";
          error(os.str());
        }
        LARGE_INTEGER liFileSize;
        if (GetFileSizeEx(_file, &liFileSize) == 0)
        {
          std::ostringstream os;
          os << "Failure getting file size of " << fileName_ << " for MMapStoreBuffer";
          error(os.str());
        }
#ifdef _WIN64
        size_t fileSize = liFileSize.QuadPart;
#else
        size_t fileSize = liFileSize.LowPart;
#endif
        setStoredSizeAlignment(fileName_, fileSize,
                               useStoredSizeAlignment_);
        _setSize( initialSize_ > fileSize ?
                  initialSize_ : fileSize);
      }
      catch (...)
      {
        closeOnConstructionError();
        throw;
      }
    }
#else
    MMapStoreBuffer(const std::string& fileName_,
                    size_t initialSize_ = AMPS_MEMORYBUFFER_DEFAULT_LENGTH,
                    size_t sizeAlignment_ = 1,
                    bool useStoredSizeAlignment_ = false)
      : MemoryStoreBuffer(MemoryStoreBuffer::EMPTY)
      , _sizeAlignment(sizeAlignment_ ? sizeAlignment_ : 1)
      , _fd(-1)
    {
      try
      {
        if (initialSize_ == 0)
        {
          initialSize_ = AMPS_MEMORYBUFFER_DEFAULT_LENGTH;
        }
        _fd = ::open(fileName_.c_str(), O_CREAT | O_RDWR, (mode_t)0644);
        struct stat statBuf;
        memset(&statBuf, 0, sizeof(statBuf));
        if (fstat(_fd, &statBuf) == -1)
        {
          std::ostringstream os;
          os << "Failed to stat file " << fileName_ << " for MMapStoreBuffer";
          error(os.str());
        }
        bool recovery = (size_t)statBuf.st_size >= initialSize_;
        setStoredSizeAlignment(fileName_, (size_t)statBuf.st_size,
                               useStoredSizeAlignment_);
        _setSize(recovery ? (size_t)statBuf.st_size
                          : initialSize_,
                 recovery);
      }
      catch (...)
      {
        closeOnConstructionError();
        throw;
      }
    }
#endif

    ~MMapStoreBuffer()
    {
      if (_buffer)
      {
        close();
      }
    }

    void close()
    {
      sync();
#ifdef _WIN32
      FlushFileBuffers(_file);
      UnmapViewOfFile(_buffer);
      CloseHandle(_mapFile);
      CloseHandle(_file);
      _mapFile = INVALID_HANDLE_VALUE;
      _file = INVALID_HANDLE_VALUE;
#else
      munmap(_buffer, _bufferLen);
      ::close(_fd);
      _fd = 0;
#endif
      _buffer = NULL;
      _bufferLen = 0;
    }

    void sync()
    {
      if (_buffer != NULL && _bufferLen)
      {
#ifdef _WIN32
        if (!FlushViewOfFile(_buffer, _bufferPos))
#else
        if (msync(_buffer, _bufferPos, MS_ASYNC) != 0)
#endif
        {
          std::ostringstream os;
          os << "Failed to sync mapped memory; buffer: " << (size_t)_buffer
             << " pos: " << _bufferPos;
          error(os.str());
        }
      }
    }

    virtual void setSize(size_t newSize_)
    {
      _setSize(newSize_);
    }

    void setSizeAlignment(size_t sizeAlignment_)
    {
      const size_t alignment = sizeAlignment_ ? sizeAlignment_ : 1;
      if (_sizeAlignment == alignment)
      {
        return;
      }
      const size_t alignedSize = getMappedSize(_bufferLen, alignment);
      _sizeAlignment = alignment;
      if (_bufferLen && alignedSize != _bufferLen)
      {
        _setSize(alignedSize);
      }
    }

    void _setSize(size_t newSize_, bool recovery_ = false)
    {
      if (_bufferLen > 0)
      {
        sync();
      }
      // Keep the logical buffer aligned to the caller's allocation unit.
      // Mapping APIs handle partial final pages internally.
      size_t sz = getMappedSize(newSize_);
#ifdef _WIN32
      if (_mapFile != INVALID_HANDLE_VALUE && _mapFile != NULL)
      {
        FlushFileBuffers(_file);
        UnmapViewOfFile(_buffer);
        CloseHandle(_mapFile);
      }
#ifdef _WIN64
      _mapFile = CreateFileMapping( _file, NULL, PAGE_READWRITE, (DWORD)((sz >> 32) & 0xffffffff), (DWORD)sz, NULL);
#else
      _mapFile = CreateFileMapping( _file, NULL, PAGE_READWRITE, 0, (DWORD)sz, NULL);
#endif
      if (_mapFile == INVALID_HANDLE_VALUE || _mapFile == NULL)
      {
        _buffer = 0;
        _bufferLen = 0;
        error("Failed to create map of log file");
      }
      else
      {
        _buffer = (char*)MapViewOfFile(_mapFile, FILE_MAP_ALL_ACCESS, 0, 0, sz);
        if (_buffer == NULL)
        {
          std::ostringstream os;
          os << "Failed to map log file to memory; buffer: " << (size_t)_buffer << " length: " << sz << " previous size: " << _bufferLen;
          _buffer = 0;
          _bufferLen = 0;
          error(os.str());
        }
      }
#else
      // If not at current size, extend the underlying file
      if (sz > _bufferLen)
      {
        if (lseek(_fd, (off_t)sz - 1, SEEK_SET) == -1)
        {
          std::ostringstream os;
          os << "Seek failed for buffer extension; buffer: " << (size_t)_buffer
             << " length: " << _bufferLen << " pos: " << _bufferPos
             << " requested new size: " << newSize_;
          error(os.str());
        }
        if (!recovery_ || sz > newSize_)
        {
          if (::write(_fd, "", 1) == -1)
          {
            std::ostringstream os;
            os << "Failed to grow buffer; buffer: " << (size_t)_buffer << " length: "
               << _bufferLen << " pos: " << _bufferPos << " requested new size: "
               << newSize_;
            error(os.str());
          }
        }
      }

      void* result = NULL;
      if (_buffer == NULL)
      {
        result = mmap(0, sz, PROT_WRITE | PROT_READ, MAP_SHARED, _fd, 0);
      }
      else if (_bufferLen < sz)
      {
#if defined(linux)
        result = mremap(_buffer, _bufferLen, sz, MREMAP_MAYMOVE);
#else
        munmap(_buffer, _bufferLen);
        result = mmap(0, sz, PROT_WRITE | PROT_READ, MAP_SHARED, _fd, 0);
#endif
      }
      if (result == MAP_FAILED || result == NULL)
      {
        std::ostringstream os;
        os << "Failed to map log file to memory; buffer: "
           << (size_t)_buffer << " length: " << sz
           << " previous size: " << _bufferLen;
        _buffer = 0;
        _bufferLen = 0;
        error(os.str());
      }
      else
      {
        _buffer = (char*)result;
      }
#endif
      if (_buffer)
      {
        _bufferLen = sz;
      }
    }

  private:
    void setStoredSizeAlignment(const std::string& fileName_, size_t fileSize_,
                                bool useStoredSizeAlignment_)
    {
      if (!useStoredSizeAlignment_
          || fileSize_ < sizeof(amps_uint32_t))
      {
        return;
      }
      amps_uint32_t storedSizeAlignment = 0;
#ifdef _WIN32
      LARGE_INTEGER beginning;
      beginning.QuadPart = 0;
      DWORD bytesRead = 0;
      const DWORD alignmentSize = (DWORD)sizeof(storedSizeAlignment);
      bool readFailed = !SetFilePointerEx(_file, beginning, NULL, FILE_BEGIN)
                        || !ReadFile(_file, &storedSizeAlignment,
                                     alignmentSize, &bytesRead, NULL)
                        || bytesRead != alignmentSize;
#else
      bool readFailed = pread(_fd, &storedSizeAlignment,
                              sizeof(storedSizeAlignment), 0)
                        != (ssize_t)sizeof(storedSizeAlignment);
#endif
      if (readFailed)
      {
        std::ostringstream os;
        os << "Failure reading size alignment from " << fileName_
           << " for MMapStoreBuffer";
        error(os.str());
      }
      if (storedSizeAlignment > 1)
      {
        if ((size_t)storedSizeAlignment > fileSize_)
        {
          throw StoreException(
              "MMapStoreBuffer contains an invalid size alignment.");
        }
        _sizeAlignment = storedSizeAlignment;
      }
    }

    void closeOnConstructionError()
    {
#ifdef _WIN32
      if (_buffer)
      {
        UnmapViewOfFile(_buffer);
      }
      if (_mapFile != INVALID_HANDLE_VALUE && _mapFile != NULL)
      {
        CloseHandle(_mapFile);
      }
      if (_file != INVALID_HANDLE_VALUE)
      {
        CloseHandle(_file);
      }
      _mapFile = INVALID_HANDLE_VALUE;
      _file = INVALID_HANDLE_VALUE;
#else
      if (_bufferLen > 0)
      {
        munmap(_buffer, _bufferLen);
      }
      if (_fd >= 0)
      {
        ::close(_fd);
      }
      _fd = -1;
#endif
      _buffer = NULL;
      _bufferLen = 0;
    }

    size_t getMappedSize(size_t requestedSize_) const
    {
      return getMappedSize(requestedSize_, _sizeAlignment);
    }

    size_t getMappedSize(size_t requestedSize_, size_t sizeAlignment_) const
    {
      const size_t alignment = sizeAlignment_ > 1
                               ? sizeAlignment_ : getPageSize();
      if (requestedSize_ > (std::numeric_limits<size_t>::max)()
                           - (alignment - 1))
      {
        throw StoreException("MMapStoreBuffer size overflow.");
      }
      return (requestedSize_ + alignment - 1) / alignment * alignment;
    }

    void error(const std::string & message)
    {
      std::ostringstream os;
#ifdef _WIN32
      const size_t errorBufferSize = 1024;
      char errorBuffer[errorBufferSize];
      memset(errorBuffer, 0, errorBufferSize);
      strerror_s(errorBuffer, errorBufferSize, errno);
      os << message << ". Error is " << errorBuffer;
#else
      os << message << ". Error is " << strerror(errno);
#endif
      throw StoreException(os.str());
    }

    size_t _sizeAlignment;
#ifdef _WIN32
    HANDLE _mapFile;
    HANDLE _file;
#else
    int    _fd;
#endif
    static size_t getPageSize()
    {
      static const size_t pageSize = queryPageSize(); // -V1096
      return pageSize;
    }

    static size_t queryPageSize()
    {
      size_t pageSize;
#ifdef _WIN32
      SYSTEM_INFO SYS_INFO;
      GetSystemInfo(&SYS_INFO);
      pageSize = SYS_INFO.dwPageSize;
#else
      pageSize = (size_t)sysconf(_SC_PAGESIZE);
#endif
      if (pageSize == (size_t)-1)
      {
        // Guess
        pageSize = 4096;
      }
      return pageSize;
    }

  };

} // end namespace AMPS

#endif //_MMAPSTOREBUFFER_H_