RingBookmarkStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2021 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 _RINGBOOKMARKSTORE_H_
#define _RINGBOOKMARKSTORE_H_

#define AMPS_RING_POSITIONS 3
// Setting bookmark max at 6 bookmarks in a range (5 commas, open, :, and close)
#define AMPS_RING_BYTES_BOOKMARK (AMPS_MAX_BOOKMARK_LEN * 6 + 8)
#define AMPS_RING_ENTRY_SIZE 1024
#define AMPS_RING_BYTES_SUBID ( AMPS_RING_ENTRY_SIZE - ( AMPS_RING_POSITIONS * AMPS_RING_BYTES_BOOKMARK ) )
#define AMPS_RING_ENTRIES 32

#include <MemoryBookmarkStore.hpp>
#ifdef _WIN32
#include <windows.h>
#else
#include <sys/mman.h>
#include <unistd.h>
#endif
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#if !defined(MREMAP_MAYMOVE)
#define MREMAP_MAYMOVE 0
#endif


namespace AMPS
{
class RingBookmarkStore : public MemoryBookmarkStore
{
 struct SubscriptionPosition
 {
 size_t _index;
 size_t _current;
 };

public:
 RingBookmarkStore(const char* fileName_)
 : MemoryBookmarkStore(), _fileSize(0), _currentIndex(0), _log(0)
#ifdef _WIN32
 , _file(INVALID_HANDLE_VALUE), _mapFile(INVALID_HANDLE_VALUE)
#else
 , _fd(0)
#endif
 , _ringRecovering(true)
 {
 init(fileName_);
 }

 RingBookmarkStore(const std::string& fileName_)
 : MemoryBookmarkStore(), _fileSize(0), _currentIndex(0), _log(0)
#ifdef _WIN32
 , _file(INVALID_HANDLE_VALUE), _mapFile(INVALID_HANDLE_VALUE)
#else
 , _fd(0)
#endif
 , _ringRecovering(true)
 {
 init(fileName_.c_str());
 }

 virtual ~RingBookmarkStore()
 {
#ifdef _WIN32
 UnmapViewOfFile(_log);
 _log = 0;
 CloseHandle(_mapFile);
 _mapFile = INVALID_HANDLE_VALUE;
 CloseHandle(_file);
 _file = INVALID_HANDLE_VALUE;
#else
 munmap(_log, _fileSize);
 _log = 0;
 close(_fd);
 _fd = 0;
#endif
 // In case _lock gets acquired by reader thread between end of this
 // destructor and start of base class destructor, prevent write()
 _ringRecovering = true;
 }

 virtual size_t log(Message& message_)
 {
 Lock<Mutex> guard(_lock);
 size_t ret = MemoryBookmarkStore::_log(message_);
 if (BookmarkRange::isRange(message_.getBookmark()))
 {
 Message::Field subId = message_.getSubscriptionId();
 if (subId.empty())
 subId = message_.getSubscriptionIds();
 write(subId, MemoryBookmarkStore::_getMostRecent(subId, false));
 }
 return ret;
 }

 virtual void discard(const Message& message_)
 {
 Lock<Mutex> guard(_lock);
 if (MemoryBookmarkStore::_discard(message_) && _recentChanged)
 {
 Message::Field subId = message_.getSubscriptionId();
 if (subId.empty())
 subId = message_.getSubscriptionIds();
 write(subId, MemoryBookmarkStore::_getMostRecent(subId, false));
 _recentChanged = false;
 }
 }

 virtual void discard(const Message::Field& subId_, size_t bookmarkSeqNo_)
 {
 Lock<Mutex> guard(_lock);
 if (MemoryBookmarkStore::_discard(subId_, bookmarkSeqNo_)
 && _recentChanged)
 {
 write(subId_, MemoryBookmarkStore::_getMostRecent(subId_, false));
 _recentChanged = false;
 }
 }

 virtual void persisted(const Message::Field& subId_,
 const Message::Field& bookmark_)
 {
 Lock<Mutex> guard(_lock);
 MemoryBookmarkStore::_persisted(find(subId_), bookmark_);
 if (_recentChanged) {
 write(subId_, MemoryBookmarkStore::_getMostRecent(subId_, false));
 _recentChanged = false;
 }
 }

 virtual Message::Field getMostRecent(const Message::Field& subId_)
 {
 Lock<Mutex> guard(_lock);
 return MemoryBookmarkStore::_getMostRecent(subId_);
 }

 virtual void purge()
 {
 Lock<Mutex> guard(_lock);
 _positionMap.clear();
 memset(_log, 0, _fileSize);
 MemoryBookmarkStore::_purge();
 _currentIndex = 0;
 }

 virtual void purge(const Message::Field& subId_)
 {
 Lock<Mutex> guard(_lock);
 Lock<Mutex> fileGuard(_fileLock);
 Lock<Mutex> posGuard(_posLock);
 if(_positionMap.count(subId_) == 0)
 {
 return;
 }
 // Remove from memory
 MemoryBookmarkStore::_purge(subId_);
 // Remove from the file
 SubscriptionPosition pos = _positionMap[subId_];
 memset(_log + (pos._index * AMPS_RING_ENTRY_SIZE), 0,
 AMPS_RING_ENTRY_SIZE);
 // Move any following subs back an index
 Message::Field sub;
 for (size_t index = pos._index; index < _currentIndex - 1; ++index)
 {
 char* start = _log + (index * AMPS_RING_ENTRY_SIZE);
 memcpy(start, start + AMPS_RING_ENTRY_SIZE, AMPS_RING_ENTRY_SIZE);
 char* end = (char*)memchr(start, '\0', AMPS_RING_BYTES_SUBID);
 if (!end) break;
 sub.assign(start, (size_t)(end - start));
 _positionMap[sub]._index = index;
 }
 _positionMap.erase(subId_);
 // We have one less sub
 --_currentIndex;
 // Clear the end
 memset(_log + (_currentIndex * AMPS_RING_ENTRY_SIZE), 0,
 AMPS_RING_ENTRY_SIZE);
 }

private:
 void init(const char* fileName_)
 {
#ifdef _WIN32
 _file = CreateFileA(fileName_, GENERIC_READ | GENERIC_WRITE, 0,
 NULL, OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
 if( _file == INVALID_HANDLE_VALUE )
 {
 DWORD err = getErrorNo();
 std::ostringstream os;
 os << "Failed to create file " << fileName_ << " for RingBookmarkStore\n";
 error(os.str(), err);
 }
 LARGE_INTEGER liFileSize;
 if(GetFileSizeEx(_file, &liFileSize) == 0)
 {
 error("Failure getting file size for RingBookmarkStore.", getErrorNo());
 }
 DWORD fsLow = liFileSize.LowPart;
 DWORD fsHigh = liFileSize.HighPart;
#ifdef _WIN64
 size_t fileSize = liFileSize.QuadPart;
#else
 size_t fileSize = liFileSize.LowPart;
#endif
 size_t existingSize = AMPS_RING_ENTRIES*AMPS_RING_ENTRY_SIZE;
 if(existingSize > fileSize)
 {
 fsLow = (DWORD)existingSize;
#ifdef _WIN64
 fsHigh = (DWORD)(existingSize >> 32);
#endif
 }
 setFileSize(fsHigh, fsLow);
#else
 _fd = open(fileName_, O_RDWR | O_CREAT, (mode_t)0644);
 if(_fd == -1)
 {
 int err = getErrorNo();
 std::ostringstream os;
 os << "Failed to open log file " << fileName_ << " for RingBookmarkStore";
 error(os.str(), err);
 }
 struct stat statBuf;
 if(fstat(_fd, &statBuf) == -1)
 {
 int err = getErrorNo();
 std::ostringstream os;
 os << "Failed to stat log file " << fileName_ << " for RingBookmarkStore";
 error(os.str(), err);
 }
 size_t fSize = (size_t)statBuf.st_size;
 if (fSize == 0)
 if (::write(_fd, "\0\0\0\0", 4) != 4)
 error("Failed to initialize empty file.", getErrorNo());
 setFileSize((fSize > AMPS_RING_ENTRIES*AMPS_RING_ENTRY_SIZE ?
 fSize - 1 : AMPS_RING_ENTRIES * AMPS_RING_ENTRY_SIZE));
#endif
 recover();
 }

#ifdef _WIN32
 DWORD getErrorNo() const
 {
 return GetLastError();
 }

 void error(const std::string& message_, DWORD err)
 {
 std::ostringstream os;
 static const DWORD msgSize = 2048;
 char pMsg[msgSize];
 DWORD sz = FormatMessageA(FORMAT_MESSAGE_FROM_SYSTEM|
 FORMAT_MESSAGE_ARGUMENT_ARRAY,
 NULL, err, LANG_NEUTRAL,
 pMsg, msgSize, NULL);
 os << message_ << ". Error is " << pMsg;
 throw StoreException(os.str());
 }
#else
 int getErrorNo() const
 {
 return errno;
 }

 void error(const std::string& message_, int err)
 {
 std::ostringstream os;
 os << message_ << ". Error is " << strerror(err);
 throw StoreException(os.str());
 }
#endif

 // Used to log the new _mostRecent bookmark
 void write(const Message::Field& subId_, 
 const Message::Field& bookmark_)
 {
 Lock<Mutex> guard(_fileLock);
 if( !_ringRecovering)
 {
 if (bookmark_.len() > AMPS_RING_BYTES_BOOKMARK)
 {
 throw StoreException("Bookmark is too large for fixed size storage. Consider rebuilding after changing AMPS_RING_BYTES_BOOKMARK in include/RingBookmarkStore.hpp");
 }
 SubscriptionPosition& pos = findPos(subId_);
 size_t nextPos = (pos._current + 1) % AMPS_RING_POSITIONS;
 // Get pointer to start of next position for cursor
 char* offset = _log + (pos._index * AMPS_RING_ENTRY_SIZE) + AMPS_RING_BYTES_SUBID + (nextPos * AMPS_RING_BYTES_BOOKMARK);
 // Write the 'cursor' to start of following entry and advance offset
 *offset = '*';
 // Change offset to beginning of current bookmark
 offset = _log + ((pos._index * AMPS_RING_ENTRY_SIZE) + AMPS_RING_BYTES_SUBID + (pos._current * AMPS_RING_BYTES_BOOKMARK) + 1);
 size_t len = bookmark_.len();
 // Write the bookmark and advance offset
 memcpy(offset, static_cast<const void*>(bookmark_.data()), len);
 offset += len;
 // Set extra bytes to NULL
 memset(offset, 0, AMPS_RING_BYTES_BOOKMARK - (len + 2));
 // Return to beginning and change the cursor
 offset = offset - len - 1;
 *offset = '+';
 // Update current for the next write
 pos._current = nextPos;

 // Sync the changes to disk
#ifdef _WIN32
#ifdef _WIN64
 size_t syncStart = (pos._index * AMPS_RING_ENTRY_SIZE) & ~((getPageSize()-1)&0xFFFFFFFFFFFFFFFF);
#else
 size_t syncStart = (pos._index * AMPS_RING_ENTRY_SIZE) & (size_t)~(getPageSize()-1);
#endif
 if(!FlushViewOfFile(_log + syncStart, (pos._index * AMPS_RING_ENTRY_SIZE) - syncStart + AMPS_RING_ENTRY_SIZE))
#else
 size_t syncStart = (pos._index * AMPS_RING_ENTRY_SIZE) & ~(getPageSize()-1);
 if(msync(_log+syncStart, (pos._index * AMPS_RING_ENTRY_SIZE) - syncStart + AMPS_RING_ENTRY_SIZE, MS_ASYNC) != 0)
#endif
 {
 error("Failed to sync mapped memory", getErrorNo());
 }
 }
 }

#ifdef _WIN32
 void setFileSize(DWORD newSizeHigh_, DWORD newSizeLow_)
 {
 bool remap = (_mapFile && _mapFile != INVALID_HANDLE_VALUE);
 if(remap)
 {
 UnmapViewOfFile(_log);
 CloseHandle(_mapFile);
 _positionMap.clear();
 }
 _mapFile = CreateFileMappingA( _file, NULL, PAGE_READWRITE, newSizeHigh_, newSizeLow_, NULL);
 if(_mapFile == NULL || _mapFile == INVALID_HANDLE_VALUE)
 {
 error("Failed to create map of log file", getErrorNo());
 _log = 0;
 _fileSize = 0;
 }
#ifdef _WIN64
 size_t sz = ((size_t)newSizeHigh_ << 32) | (size_t)newSizeLow_;
#else
 size_t sz = (size_t)newSizeLow_;
#endif
 _log = (char*)MapViewOfFile(_mapFile, FILE_MAP_ALL_ACCESS, 0, 0, sz);
 if(_log == NULL)
 {
 error("Failed to map log file to memory", getErrorNo());
 _log = 0;
 _fileSize = 0;
 return;
 }
 _fileSize = sz;
 // Call recover to reset the _positionMap
 if(remap)
 recover();
 }
#else
 void setFileSize(size_t newSize_)
 {
 bool remap = (_log != 0);
 // Make sure we're using a multiple of page size
 size_t sz = newSize_ & (size_t)(~(getPageSize()-1));
 if(sz < newSize_)
 {
 sz += getPageSize();
 }
 // Improper resize attempt
 if(newSize_ <= _fileSize)
 return;
 // Extend the underlying file
 if(lseek(_fd, (off_t)sz, SEEK_SET) == -1)
 {
 error("Seek failed for RingBookmarkStore", getErrorNo());
 }
 if(::write(_fd, "", 1) == -1)
 {
 error("Failed to grow RingBookmarkStore", getErrorNo());
 }
 void* newLog = MAP_FAILED;
 if(_log)
 {
 _positionMap.clear();

#ifdef linux
 newLog = (mremap(_log, _fileSize, sz, MREMAP_MAYMOVE));
#else
 // try making a new mmap right after the current one.
 newLog = mmap(_log + _fileSize, sz, PROT_READ|PROT_WRITE,
 MAP_SHARED|MAP_FIXED, _fd, (off_t)sz);
 if(newLog!=_log)
 {
 // this mmap is relatively small; better to just close the old mmap and reset.
 munmap(_log,_fileSize);
 newLog = mmap(0,sz,PROT_READ|PROT_WRITE,MAP_SHARED,_fd,0);
 }
#endif
 }
 else // New mapping
 {
 // New mapping, map the full file size for recovery or else it std size
 newLog = (mmap(0, sz, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, 0));
 }
 _fileSize = sz;

 if(newLog == MAP_FAILED)
 {
 _log = 0;
 _fileSize = 0;
 error("Failed to map log file to memory", getErrorNo());
 }
 _log = static_cast<char*>(newLog);
 if(remap)
 recover();
 }
#endif

 void recover(void)
 {
 //Lock<Mutex> guard(_lock);
 _ringRecovering = true;
 Message::Field sub;
 Message::Field bookmarkField;

 _currentIndex = 0;
 size_t maxEntries = _fileSize/AMPS_RING_ENTRY_SIZE > AMPS_RING_ENTRIES ? _fileSize/AMPS_RING_ENTRY_SIZE : AMPS_RING_ENTRIES;
 for(; _currentIndex < maxEntries; ++_currentIndex)
 {
 char* offset = _log + (_currentIndex * AMPS_RING_ENTRY_SIZE);
 if(*offset == '\0')
 {
 break;
 }
 //It's possible we wrote the subId and not NULLs, so be careful
 char* end = (char*)memchr(offset, '\0', AMPS_RING_BYTES_SUBID);
 if(!end)
 {
 // Failed subscription id read, we're done
 break;
 }
 // Safe to continue
 sub.assign(offset, (size_t)(end - offset));
 // Put this sub into the MemoryBookmarkStore
 Subscription* subPtr = MemoryBookmarkStore::find(sub);
 // Put this sub into the _positionMap
 // This is recovery, so do it directly and not with findPos
 SubscriptionPosition& pos = _positionMap[sub];
 pos._index = _currentIndex;
 offset += AMPS_RING_BYTES_SUBID;
 size_t foundCursor = AMPS_RING_POSITIONS;
 for(pos._current = 0; pos._current < AMPS_RING_POSITIONS; pos._current++)
 {
 if(offset[pos._current*AMPS_RING_BYTES_BOOKMARK] == '*')
 {
 // Subtract one position
 pos._current = (pos._current+(AMPS_RING_POSITIONS-1)) % AMPS_RING_POSITIONS;
 // Subtract one more if a second bookmark is found
 if(offset[foundCursor*AMPS_RING_BYTES_BOOKMARK] == '*')
 {
 pos._current = (pos._current+(AMPS_RING_POSITIONS-1)) % AMPS_RING_POSITIONS;
 }
 break;
 }
 }
 if(pos._current >= AMPS_RING_POSITIONS)
 {
 // No valid bookmark found, just use 0
 pos._current = 0;
 }
 else
 {
 // We found a cursor
 offset += pos._current * AMPS_RING_BYTES_BOOKMARK;
 //It's possible we wrote bookmark and not NULLs, so be careful
 end = (char*)memchr(offset, '\0', AMPS_RING_BYTES_BOOKMARK);
 if(end && end != offset)
 {
 // add 1 to account for leading '+'
 bookmarkField.assign(offset + 1, (size_t)(end - offset - 1));
 // log, discard to make it the most recent
 if (!BookmarkRange::isRange(bookmarkField))
 {
 // This adds bookmark to _publishers
 subPtr->isDiscarded(bookmarkField);
 }
 subPtr->discard(subPtr->log(bookmarkField));
 }
 }
 }
 _ringRecovering = false;
 }

 SubscriptionPosition& findPos(const Message::Field& subId_)
 {
 Lock<Mutex> guard(_posLock);
 if(_positionMap.count(subId_) == 0)
 {
 // New subid
 // Move to its start position and write the sub id
 char* offset = _log + (_currentIndex * AMPS_RING_ENTRY_SIZE);
 size_t len = subId_.len();
 memcpy(offset, static_cast<const void*>(subId_.data()), len);
 // Add it to the map with the current index
 // Use the data written to the mmap for the subid
 Message::Field subId;
 subId.assign(offset, len);
 _positionMap[subId]._index = _currentIndex;
 _positionMap[subId]._current = 0;
 // Fill extra spaces with NULL
 offset += len;
 memset(offset, 0, AMPS_RING_BYTES_SUBID - len);
 // Advance current index
 ++_currentIndex;
 }
 return _positionMap[subId_];
 }

 Subscription* find(const Message::Field& subId_)
 {
 if(subId_.empty())
 {
 throw StoreException("A valid subscription ID must be provided to the RingBookmarkStore");
 }
 findPos(subId_);
 return MemoryBookmarkStore::find(subId_);
 }


 Mutex _fileLock;
 size_t _fileSize;
 size_t _currentIndex;
 char* _log;
#ifdef _WIN32
 HANDLE _file;
 HANDLE _mapFile;
#else
 int _fd;
#endif
 Mutex _posLock;
 typedef std::map<Message::Field, SubscriptionPosition, Message::Field::FieldHash> PositionMap;
 PositionMap _positionMap;
 bool _ringRecovering;
#ifdef _WIN32
 static DWORD getPageSize()
 {
 static DWORD pageSize = 0;
 if(pageSize == 0)
 {
 SYSTEM_INFO SYS_INFO;
 GetSystemInfo(&SYS_INFO);
 pageSize = SYS_INFO.dwPageSize;
#else
 static size_t getPageSize()
 {
 static size_t pageSize = 0UL;
 if(pageSize == 0)
 {
 pageSize = (size_t)sysconf(_SC_PAGESIZE);
#endif
 }
 return pageSize;
 }


};

} // end namespace AMPS


#endif // _RINGBOOKMARKSTORE_H_