MMapBookmarkStore.hpp

Go to the documentation of this file.

//
// Copyright (c) 2010-2020 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..
//

#include <ampsplusplus.hpp>

#ifndef _MMAPBOOKMARKSTORE_H_
#define _MMAPBOOKMARKSTORE_H_

#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>
#include <set>



namespace AMPS
{
class MMapBookmarkStore : public MemoryBookmarkStore
{
private:
#ifdef _WIN32
    typedef HANDLE FileType;
    HANDLE _mapFile;
#else
    typedef int FileType;
#endif
    static void _clearBookmark(std::pair<const Message::Field, size_t>& pair)
    {
        Message::Field f(pair.first);
        f.clear();
    }

public:
    MMapBookmarkStore(const char* fileName_)
        : MemoryBookmarkStore(true), _fileName(fileName_), _fileSize(0)
        , _logOffset(0) , _log(0)
#ifdef _WIN32
        , _file(INVALID_HANDLE_VALUE), _mapFile(INVALID_HANDLE_VALUE)
#else
        , _file(0)
#endif
    {
        init();
        recover();
    }

    MMapBookmarkStore(const std::string& fileName_)
        : MemoryBookmarkStore(true), _fileName(fileName_), _fileSize(0)
        , _logOffset(0) , _log(0)
#ifdef _WIN32
        , _file(INVALID_HANDLE_VALUE), _mapFile(INVALID_HANDLE_VALUE)
#else
        , _file(0)
#endif
    {
        init();
        recover();
    }

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

    virtual size_t log(Message& message_)
    {
        Message::Field bookmark = message_.getBookmark();
        Subscription* sub = (Subscription*)(message_.getSubscriptionHandle());
        Lock<Mutex> guard(_lock);
        if (!sub)
        {
            Message::Field subId = message_.getSubscriptionId();
            if (subId.empty())
                subId = message_.getSubscriptionIds();
            sub = find(subId);
            message_.setSubscriptionHandle(static_cast<amps_subscription_handle>(sub));
        }
        write(sub->id(), ENTRY_BOOKMARK, bookmark);
        return MemoryBookmarkStore::_log(message_);
    }

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

    virtual void discard(const Message::Field& subId_, size_t bookmarkSeqNo_)
    {
        Lock<Mutex> guard(_lock);
        Subscription::Entry* entry = find(subId_)->getEntryByIndex(bookmarkSeqNo_);
        if (!entry || entry->_val.empty()) return;
        write(subId_, ENTRY_DISCARD, entry->_val);
        MemoryBookmarkStore::_discard(subId_, bookmarkSeqNo_);
    }

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

    virtual bool isDiscarded(Message& message_)
    {
        Lock<Mutex> l(_lock);
        bool retVal = MemoryBookmarkStore::_isDiscarded(message_);
        if (retVal)
        {
            Message::Field subId = message_.getSubscriptionId();
            if (subId.empty())
                subId = message_.getSubscriptionIds();
            write(subId, ENTRY_BOOKMARK, message_.getBookmark());
            write(subId, ENTRY_DISCARD, message_.getBookmark());
        }
        return retVal;
    }

    virtual void purge()
    {
        Lock<Mutex> guard(_lock);
        Lock<Mutex> fileGuard(_fileLock);
        memset(_log, 0, _logOffset);
        _logOffset = 0;
        MemoryBookmarkStore::_purge();
    }

    virtual void purge(const Message::Field& subId_)
    {
        Lock<Mutex> guard(_lock);
        Lock<Mutex> fileGuard(_fileLock);
        MemoryBookmarkStore::_purge(subId_);
        std::string tmpFileName = _fileName + ".tmp";
        __prune(tmpFileName);
    }

    virtual void noPersistedAcks(const Message::Field& subId_)
    {
        MemoryBookmarkStore::noPersistedAcks(subId_);
    }

    void setServerVersion(const VersionInfo& version_)
    {
        Lock<Mutex> guard(_lock);
        MemoryBookmarkStore::setServerVersion(version_);
    }

    void setServerVersion(size_t version_)
    {
        Lock<Mutex> guard(_lock);
        MemoryBookmarkStore::setServerVersion(version_);
    }

    // Yes, the argument is a non-const copy of what is passed in
    void _prune(std::string tmpFileName_)
    {
        if (tmpFileName_.empty()) tmpFileName_ = _fileName + ".tmp";
        Lock<Mutex> guard(_lock);
        Lock<Mutex> fileGuard(_fileLock);
        // If nothing's changed with most recent, don't rewrite the file
        if (!_recentChanged)
        {
            return;
        }
        __prune(tmpFileName_);
        _recentChanged = false;
    }

private:
    void __prune(std::string tmpFileName_)
    {
        size_t sz = AMPS_INITIAL_LOG_SIZE;
        FileType file;
        char* log = NULL;
        size_t bytesWritten = 0;
#ifdef _WIN32
        file = CreateFileA(tmpFileName_.c_str(), 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 " << tmpFileName_ << " for pruning MMapBookmarkStore";
            error(os.str(), err);
        }
        HANDLE mapFile = NULL;
        try
        {
            sz = _setFileSize(sz, &log, file, &mapFile);
        }
        catch (StoreException& ex)
        {
            if(mapFile == NULL || mapFile == INVALID_HANDLE_VALUE)
            {
                CloseHandle(file);
                std::ostringstream os;
                os << "Failed to create map of log file for prune"
                   << ex.what();
                throw StoreException(os.str());
                return;
            }
            if(log == NULL)
            {
                CloseHandle(mapFile);
                CloseHandle(file);
                std::ostringstream os;
                os << "Failed to map log file to memory for prune"
                   << ex.what();
                throw StoreException(os.str());
                return;
            }
        }
        if (sz == 0)
        {
            DWORD err = getErrorNo();
            UnmapViewOfFile(log);
            CloseHandle(mapFile);
            CloseHandle(file);
            error("Failed to prepare tmp file for prune.", err);
        }
#else
        file = open(tmpFileName_.c_str(), O_RDWR | O_CREAT, (mode_t)0644);
        if(file == -1)
        {
            int err = getErrorNo();
            std::ostringstream os;
            os << "Failed to open log file " << tmpFileName_ << " for MMapBookmarkStore";
            error(os.str(), err);
            return;
        }
        if(::write(file, "\0\0\0\0", 4) == -1)
        {
            int err = getErrorNo();
            std::ostringstream os;
            os << "Failed to write header to log file " << tmpFileName_
               << " in prune for MMapBookmarkStore";
            error(os.str(), err);
            return;
        }
        try
        {
            sz = _setFileSize(sz, &log, file, 0);
        }
        catch (StoreException& ex)
        {
            std::ostringstream os;
            os << "Failed to prepare tmp file \"" << tmpFileName_
               << "\" for prune. " << ex.what();
            throw StoreException(os.str());
        }
        if (sz == 0)
        {
            int err = getErrorNo();
            log = NULL;
            ::close(file);
            error("Failed to prepare tmp file for prune.", err);
        }
#endif
        try
        {
            for (SubscriptionMap::iterator i = _subs.begin(); i != _subs.end(); ++i)
            {
                Message::Field subId = i->first;
                assert(!subId.empty());
                size_t subIdLen = subId.len();
                Message::Field recent = i->second->getMostRecent(false);
                amps_uint64_t recentPub, recentSeq;
                Subscription::parseBookmark(recent, recentPub, recentSeq);
                Subscription::PublisherMap publishersDiscarded =
                                                i->second->_publishers;
                MemoryBookmarkStore::EntryPtrList recovered;
                i->second->getRecoveryEntries(recovered);
                i->second->setPublishersToDiscarded(&recovered,
                                                    &publishersDiscarded);
                char tmpBookmarkBuffer[128];
                for (Subscription::PublisherIterator pub =
                                                publishersDiscarded.begin(),
                                                e = publishersDiscarded.end();
                     pub != e; ++pub)
                {
                    // Don't log EPOCH if it got in the map
                    if (pub->first == 0 || pub->second == 0) continue;
                    // Don't log the most recent yet
                    if (pub->first == recentPub) continue;
                    int written = AMPS_snprintf_amps_uint64_t(
                                                    tmpBookmarkBuffer,
                                                    sizeof(tmpBookmarkBuffer),
                                                    pub->first);
                    *(tmpBookmarkBuffer+written++) = '|';
                    written += AMPS_snprintf_amps_uint64_t(
                                            tmpBookmarkBuffer+written,
                                            sizeof(tmpBookmarkBuffer)
                                              - (size_t)written,
                                            pub->second);
                    *(tmpBookmarkBuffer+written++) = '|';
                    Message::Field tmpBookmark(tmpBookmarkBuffer, (size_t)written);
                    // Check we'll be in the current boundaries
                    size_t blockLen = subIdLen + 2*sizeof(size_t) + tmpBookmark.len() + 1;
                    if(bytesWritten + blockLen + blockLen >= sz)
                    {
#ifdef _WIN32
                        sz = _setFileSize(sz*2, &log, file, &mapFile);
#else
                        sz = _setFileSize(sz*2, &log, file, sz);
#endif
                    }
                    write(&log, &bytesWritten, subId, ENTRY_BOOKMARK, tmpBookmark);
                    write(&log, &bytesWritten, subId, ENTRY_DISCARD, tmpBookmark);
                }
                if (isWritableBookmark(recent.len()))
                {
                    // Check we'll be in the current boundaries
                    size_t blockLen = subIdLen + 2*sizeof(size_t) + recent.len() + 1;
                    if(bytesWritten + blockLen + blockLen >= sz)
                    {
#ifdef _WIN32
                        sz = _setFileSize(sz*2, &log, file, &mapFile);
#else
                        sz = _setFileSize(sz*2, &log, file, sz);
#endif
                    }
                    write(&log, &bytesWritten, subId, ENTRY_BOOKMARK, recent);
                    write(&log, &bytesWritten, subId, ENTRY_DISCARD, recent);
                }
                else // set up _recentList
                {
                    i->second->getMostRecentList();
                }
                Message::Field bookmark = i->second->getLastPersisted();
                if (isWritableBookmark(bookmark.len()))
                {
                    // Check we'll be in the current boundaries
                    size_t blockLen = subIdLen + 2*sizeof(size_t) +
                                      bookmark.len() + 1;
                    if(bytesWritten + blockLen >= sz)
                    {
#ifdef _WIN32
                        sz = _setFileSize(sz*2, &log, file, &mapFile);
#else
                        sz = _setFileSize(sz*2, &log, file, sz);
#endif
                    }
                    write(&log, &bytesWritten, subId, ENTRY_PERSISTED,
                          i->second->getLastPersisted());
                }
                i->second->getActiveEntries(recovered);
                for (MemoryBookmarkStore::EntryPtrList::iterator entry =
                                                              recovered.begin();
                     entry != recovered.end(); ++entry)
                {
                    if ((*entry)->_val.empty() ||
                        !isWritableBookmark((*entry)->_val.len()))
                        continue;
                    // Check we'll be in the current boundaries
                    size_t blockLen = subIdLen + 2*sizeof(size_t) +
                                      (*entry)->_val.len() + 1;
                    if(bytesWritten + blockLen >= sz)
                    {
#ifdef _WIN32
                        sz = _setFileSize(sz*2, &log, file, &mapFile);
#else
                        sz = _setFileSize(sz*2, &log, file, sz);
#endif
                    }
                    write(&log, &bytesWritten, subId, ENTRY_BOOKMARK,
                          (*entry)->_val);
                    if (!(*entry)->_active)
                    {
                        // Check we'll be in the current boundaries
                        if(bytesWritten + blockLen >= sz)
                        {
#ifdef _WIN32
                            sz = _setFileSize(sz*2, &log, file, &mapFile);
#else
                            sz = _setFileSize(sz*2, &log, file, sz);
#endif
                        }
                        write(&log, &bytesWritten, subId, ENTRY_DISCARD,
                              (*entry)->_val);
                    }
                }
            }
        }
        catch (StoreException& ex)
        {
#ifdef _WIN32
            UnmapViewOfFile(log);
            CloseHandle(mapFile);
            CloseHandle(file);
#else
            ::close(file);
            ::unlink(tmpFileName_.c_str());
#endif
            std::ostringstream os;
            os << "Exception during prune: " << ex.what();
            throw StoreException(os.str());
        }
#ifdef _WIN32
        BOOL success = FlushViewOfFile(_log,0);
        success |= UnmapViewOfFile(_log);
        _log = NULL;
        success |= CloseHandle(_mapFile);
        success |= CloseHandle(_file);
        if (!success)
        {
            DWORD err = getErrorNo();
            std::ostringstream os;
            os << "Failed to flush, unmap, and close current file "
               << _fileName
               << " in prune in MMapBookmarkStore. ";
            error(os.str(), err);
        }
        _mapFile = INVALID_HANDLE_VALUE;
        _file = INVALID_HANDLE_VALUE;
        success = FlushViewOfFile(log,0);
        success |= UnmapViewOfFile(log);
        log = NULL;
        success |= CloseHandle(mapFile);
        success |= CloseHandle(file);
        if (!success)
        {
            DWORD err = getErrorNo();
            std::ostringstream os;
            os << "Failed to flush, unmap and close completed temp file "
               << tmpFileName_
               << " in prune in MMapBookmarkStore. ";
            error(os.str(), err);
        }
        mapFile = INVALID_HANDLE_VALUE;
        file = INVALID_HANDLE_VALUE;
        // Replace current file with pruned file
        int retryCount = 3;
        while (!MoveFileExA(tmpFileName_.c_str(), _fileName.c_str(),
                MOVEFILE_COPY_ALLOWED|MOVEFILE_REPLACE_EXISTING|MOVEFILE_WRITE_THROUGH))
        {
            DWORD err = getErrorNo();
            if (--retryCount > 0) continue;
            // Try to set _file to the tmp file that won't move then throw
            std::string desiredFileName = _fileName;
            _fileName = tmpFileName_;
            init();
            std::ostringstream os;
            os << "Failed to move completed temp file " << tmpFileName_
               << " to " << desiredFileName
               << " in prune in MMapBookmarkStore. Continuing by using "
               << tmpFileName_ << " as the MMapBookmarkStore file.";
            error(os.str(), err);
        }
        // Call init to set up file again
        init();
#else
        munmap(_log, _fileSize);
        _log = NULL;
        ::close(_file);
        munmap(log, sz);
        ::close(file);
        if (-1 == ::unlink(_fileName.c_str()))
        {
            int err = getErrorNo();
            // Try to set _file to the tmp file that won't move then throw
            std::string desiredFileName = _fileName;
            _fileName = tmpFileName_;
            init();
            std::ostringstream os;
            os << "Failed to delete file " << desiredFileName
               << " after creating temporary file " << tmpFileName_
               << " in prune in MMapBookmarkStore. Continuing by using "
               << tmpFileName_ << " as the MMapBookmarkStore file.";
            error(os.str(), err);
        }
        if (-1 == ::rename(tmpFileName_.c_str(), _fileName.c_str()))
        {
            int err = getErrorNo();
            // Try to set _file to the tmp file that won't move then throw
            std::string desiredFileName = _fileName;
            _fileName = tmpFileName_;
            init();
            std::ostringstream os;
            os << "Failed to move completed temp file " << tmpFileName_
               << " to " << desiredFileName
               << " in prune in MMapBookmarkStore. Continuing by using "
               << tmpFileName_ << " as the MMapBookmarkStore file.";
            error(os.str(), err);
        }
        // Call init to set up file again
        init();
#endif
        _logOffset = bytesWritten;
    }

    virtual void _persisted(Subscription* subP_,
                            const Message::Field& bookmarkField_)
    {
        Lock<Mutex> l(_lock);
        write(subP_->id(), ENTRY_PERSISTED, bookmarkField_);
        MemoryBookmarkStore::_persisted(subP_, bookmarkField_);
    }

    virtual Message::Field _persisted(Subscription* subP_, size_t bookmark_)
    {
        Lock<Mutex> l(_lock);
        Subscription::Entry* entryPtr = subP_->getEntryByIndex(bookmark_);
        if (!entryPtr || entryPtr->_val.empty())
            return Message::Field();
        Message::Field bookmarkField = entryPtr->_val;
        write(subP_->id(), ENTRY_PERSISTED, bookmarkField);
        MemoryBookmarkStore::_persisted(subP_, bookmarkField);
        return bookmarkField;
    }

    void init(void)
    {
#ifdef _WIN32
        _file = CreateFileA(_fileName.c_str(), 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 MMapBookmarkStore";
            error(os.str(), err);
        }
        LARGE_INTEGER liFileSize;
        if(GetFileSizeEx(_file, &liFileSize) == 0)
        {
            DWORD err = getErrorNo();
            CloseHandle(_file);
            std::ostringstream os;
            os << "Failure getting file size for MMapBookmarkStore " << _fileName;
            error(os.str(), err);
        }
#ifdef _WIN64
        size_t fileSize = liFileSize.QuadPart;
#else
        size_t fileSize = liFileSize.LowPart;
#endif
        setFileSize( AMPS_INITIAL_LOG_SIZE > fileSize ?
                 AMPS_INITIAL_LOG_SIZE : fileSize);
#else
        _file = open(_fileName.c_str(), O_RDWR | O_CREAT, (mode_t)0644);
        if(_file == -1)
        {
            int err = getErrorNo();
            std::ostringstream os;
            os << "Failed to open log file " << _fileName << " for MMapBookmarkStore";
            error(os.str(), err);
        }
        struct stat statBuf;
        if(fstat(_file, &statBuf) == -1)
        {
            int err = getErrorNo();
            ::close(_file);
            std::ostringstream os;
            os << "Failed to stat log file " << _fileName << " for MMapBookmarkStore";
            error(os.str(), err);
        }
        size_t fSize = (size_t)statBuf.st_size;
        if(fSize == 0)
        {
            if(::write(_file, "\0\0\0\0", 4) == -1)
            {
                int err = getErrorNo();
                ::close(_file);
                std::ostringstream os;
                os << "Failed to write header to log file " << _fileName
                   << " for MMapBookmarkStore";
                error(os.str(), err);
            }
        }

        setFileSize((fSize > AMPS_INITIAL_LOG_SIZE) ? fSize-1 : AMPS_INITIAL_LOG_SIZE);
#endif
    }

#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 << "File: " << _fileName << ". " << message_ << " with error " << 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
#if defined(sparc)
#define AMPS_WRITE8(p,v) { p[0] = (v>>56)&0xFF; p[1] = (v>>48)&0xFF; p[2] = (v>>40)&0xFF; p[3] = (v>>32)&0xFF; p[4] = (v>>24)&0xFF; p[5] = (v>>16)&0xFF; p[6] = (v>>8)&0xFF; p[7]=v&0xFF; }
#define AMPS_READ8(p, v) { memcpy(&v,p,8); }
#else
#define AMPS_WRITE8(p,v) { *(size_t*)p = (size_t)v; }
#define AMPS_READ8(p,v)  { v = *(const size_t*)p; }
#endif

    // This implementation will use this when logging a bookmark or a persisted
    void write(const Message::Field& subId_,
               char type_, const Message::Field& bookmark_)
    {
        Lock<Mutex> guard(_fileLock);
        write(&_log, &_logOffset, subId_, type_, bookmark_);
    }

    void write(char** logPtr, size_t* logOffsetPtr, const Message::Field& subId_,
               char type_, const Message::Field& bookmark_)
    {
        if(!_recovering && isWritableBookmark(bookmark_.len()))
        {
            size_t len = subId_.len();
            // Check we'll be in the current boundaries
            size_t blockLen = len + 2*sizeof(size_t) + bookmark_.len() + 1;
            if(*logOffsetPtr + blockLen >= _fileSize)
            {
                setFileSize(_fileSize*2);
            }
            char* offset = *logPtr+*logOffsetPtr;
            AMPS_WRITE8(offset,len);
            offset += sizeof(size_t);
            memcpy(offset, static_cast<const void*>(subId_.data()), len);
            offset += len;
            *offset++ = type_;
            len = bookmark_.len();
            AMPS_WRITE8(offset,len);
            offset += sizeof(size_t);
            memcpy(offset, static_cast<const void*>(bookmark_.data()), len);
            *logOffsetPtr += blockLen;
        }
    }

    // This implementation will only ever use this when discarding a bookmark
    // Could be used to add a feature where generated bookmarks are logged in
    // addition to the bookmark field.
    void write(const Message::Field& subId_, char type_, size_t bookmark_)
    {
        Lock<Mutex> guard(_fileLock);
        write(&_log, &_logOffset, subId_, type_, bookmark_);
    }

    void write(char** logPtr, size_t* logOffsetPtr, const Message::Field& subId_,
               char type_, size_t bookmark_)
    {
        if(!_recovering)
        {
            size_t len = subId_.len();
            size_t blockLen = len + 2*sizeof(size_t) + 1;
            // Check we'll be in the current boundaries
            if(*logOffsetPtr + blockLen >= _fileSize)
            {
                setFileSize(_fileSize*2);
            }
            char* offset = *logPtr+*logOffsetPtr;
            *(reinterpret_cast<size_t*>(offset)) = len;
            offset += sizeof(size_t);
            memcpy(offset, static_cast<const void*>(subId_.data()), len);
            offset += len;
            *offset++ = type_;
            *(reinterpret_cast<size_t*>(offset)) = bookmark_;
            *logOffsetPtr += blockLen;
        }
    }

    void setFileSize(size_t newSize_)
    {
        if(_log && newSize_ <= _fileSize) // Improper resize attempt
            return;
#ifdef _WIN32
        _fileSize = _setFileSize(newSize_, &_log, _file, &_mapFile);
#else
        _fileSize = _setFileSize(newSize_, &_log, _file, _fileSize);
#endif
    }

    // Returns new file size, 0 if there is a failure
    size_t _setFileSize(size_t newSize_, char** log_, FileType file_,
#ifdef WIN32
                       HANDLE* mapFile_
#else
                       size_t fileSize_
#endif
                       )
    {
        // Make sure we're using a multiple of page size
        size_t sz = newSize_ & (size_t)(~(getPageSize()-1));
        if(sz < newSize_ || sz == 0)
        {
            sz += getPageSize();
        }
#ifdef _WIN32
        if(*mapFile_ && *mapFile_ != INVALID_HANDLE_VALUE)
        {
            if(*log_)
            {
                FlushViewOfFile(*log_, 0);
                UnmapViewOfFile(*log_);
            }
            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_ == NULL || *mapFile_ == INVALID_HANDLE_VALUE)
        {
            DWORD errNo = getErrorNo();
            CloseHandle(file_);
            error("Failed to create map of log file", errNo);
            *log_ = 0;
            return 0;
        }
        else
        {
            *log_ = (char*)MapViewOfFile(*mapFile_, FILE_MAP_ALL_ACCESS, 0, 0, sz);
            if(*log_ == NULL)
            {
                DWORD errNo = getErrorNo();
                CloseHandle(*mapFile_);
                CloseHandle(file_);
                error("Failed to map log file to memory", errNo);
                *log_ = 0;
                return 0;
            }
        }
#else
        // Extend the underlying file
        if(lseek(file_, (off_t)sz, SEEK_SET) == -1)
        {
            int err = getErrorNo();
            ::close(file_);
            error("Seek failed for bookmark log", err);
        }
        if(::write(file_, "", 1) == -1)
        {
            int err = getErrorNo();
            ::close(file_);
            error("Failed to grow bookmark log", err);
        }
        if(*log_)
        {
#if defined(linux)
            *log_ = static_cast<char*>(mremap(*log_, fileSize_, sz,
                                              MREMAP_MAYMOVE));
#else
            munmap(*log_,fileSize_);
            *log_ = static_cast<char*>(mmap(0, sz, PROT_READ | PROT_WRITE,
                                            MAP_SHARED, file_, 0));
#endif
        }
        else // New mapping
        {
            // New mapping, map the full file size for recovery or else it std size
            *log_ = static_cast<char*>(mmap(0, sz, PROT_READ | PROT_WRITE,
                                            MAP_SHARED, file_, 0));
        }

        if((void*)(*log_) == MAP_FAILED)
        {
            int err = getErrorNo();
            ::close(file_);
            error("Failed to map log file to memory", err);
            *log_ = 0;
            return 0;
        }
#endif
        return sz;
    }

    void recover()
    {
        Message::Field sub;
        Message::Field bookmarkField;
        size_t bookmarkLen = 0;
        size_t lastGoodOffset = 0;
        bool inError = false;
        Lock<Mutex> guard(_lock);
        Lock<Mutex> fileGuard(_fileLock);
        _recovering = true;
        // Map of bookmark to sequence number
        typedef std::map<Message::Field, size_t, Message::Field::FieldHash> BookmarkMap;
        typedef std::map<Message::Field, size_t,
                         Message::Field::FieldHash>::iterator BookmarkMapIter;
        // Map of subId to set of recovered bookmarks
        typedef std::map<Message::Field, BookmarkMap*,
                         Message::Field::FieldHash> ReadMap;
        typedef std::map<Message::Field, BookmarkMap*,
                         Message::Field::FieldHash>::iterator ReadMapIter;
        ReadMap recovered;
        size_t subLen = *(reinterpret_cast<size_t*>(_log));
        while(!inError && subLen > 0)
        {
            _logOffset += sizeof(size_t);
            sub.assign(_log+_logOffset, subLen);
            _logOffset += subLen;
            switch(_log[_logOffset++])
            {
            case -1:
                return;
            case ENTRY_BOOKMARK:
            {
                AMPS_READ8((_log+_logOffset),bookmarkLen);
                _logOffset += sizeof(size_t);
                bookmarkField.assign(_log+_logOffset, bookmarkLen);
                _logOffset += bookmarkLen;
                Subscription* subP = find(sub);
                BookmarkMap* bookmarks = NULL;
                ReadMapIter iter = recovered.find(sub);
                if (iter == recovered.end())
                {
                    Message::Field subKey;
                    subKey.deepCopy(sub);
                    bookmarks = new BookmarkMap();
                    recovered[subKey] = bookmarks;
                }
                else
                {
                    bookmarks = iter->second;
                }
                if (bookmarks->find(bookmarkField) != bookmarks->end())
                {
                    std::for_each(bookmarks->begin(), bookmarks->end(),
                                  _clearBookmark);
                    bookmarks->clear();
                    subP->getMostRecent();
                }
                if (!subP->isDiscarded(bookmarkField))
                {
                    size_t sequence = subP->log(bookmarkField);
                    Message::Field copy;
                    copy.deepCopy(bookmarkField);
                    bookmarks->insert(std::make_pair(copy, sequence));
                }
                else
                {
                    // We know it's discarded, but there may still be a
                    // discard entry in the log, so avoid a search.
                    Message::Field copy;
                    copy.deepCopy(bookmarkField);
                    bookmarks->insert(std::make_pair(copy,0));
                }
            }
                break;
            case ENTRY_DISCARD:
            {
                AMPS_READ8((_log+_logOffset),bookmarkLen);
                _logOffset += sizeof(size_t);
                bookmarkField.assign(_log+_logOffset, bookmarkLen);
                _logOffset += bookmarkLen;
                size_t sequence = AMPS_UNSET_INDEX;
                ReadMapIter iter = recovered.find(sub);
                if (iter != recovered.end())
                {
                    BookmarkMap* bookmarks = iter->second;
                    BookmarkMapIter bookmarkIter = bookmarks->find(bookmarkField);
                    if (bookmarkIter != bookmarks->end())
                    {
                        sequence = bookmarkIter->second;
                        Message::Field bookmarkToClear(bookmarkIter->first);
                        bookmarkToClear.clear();
                        bookmarks->erase(bookmarkIter);
                    }
                }
                Subscription* subP = find(sub);
                if (sequence != AMPS_UNSET_INDEX)
                {
                    // A sequence of 0 means it was already discarded
                    if (sequence) subP->discard(sequence);
                }
                else // Shouldn't end up here, but just in case we'll search
                {
                    subP->discard(bookmarkField);
                }
            }
                break;
            case ENTRY_PERSISTED:
            {
                AMPS_READ8((_log+_logOffset),bookmarkLen);
                _logOffset += sizeof(size_t);
                bookmarkField.assign(_log+_logOffset, bookmarkLen);
                _logOffset += bookmarkLen;
                MemoryBookmarkStore::_persisted(find(sub), bookmarkField);
            }
                break;
            default:
                if (lastGoodOffset == 0)
                {
                    error("Error while recovering bookmark store file.", getErrorNo());
                }
                else
                {
                    _logOffset = lastGoodOffset;
                    inError = true;
                }
            }
            lastGoodOffset = _logOffset;
            if (!inError) subLen = *(reinterpret_cast<size_t*>(_log + _logOffset));
        }
        for (SubscriptionMap::iterator i=_subs.begin(); i != _subs.end(); ++i)
        {
            if (recovered.count(i->first) && !recovered[i->first]->empty())
            {
                if (i->second->getMostRecent(false).len() > 1)
                {
                    i->second->justRecovered();
                }
                else
                {
                    // Unlikely, but we may have recovered only undiscarded bookmarks
                    // so we should really just restart as a new subscription.
                    delete i->second;
                    _subs[i->first] = new Subscription(this, i->first);
                }
            }
        }
        for (ReadMapIter i = recovered.begin(), e = recovered.end(); i!=e; ++i)
        {
            std::for_each(i->second->begin(), i->second->end(), _clearBookmark);
            delete i->second;
            Message::Field f = i->first;
            f.clear();
        }
        _recovering = false;
    }

    Mutex _fileLock;
    std::string _fileName;
    size_t _fileSize;
    size_t _logOffset;
    char* _log;
    FileType _file;
    // Each entry begins with a single byte indicating the type of entry:
    // a new bookmark, or a discard of a previous one.
    static size_t getPageSize()
    {
        static size_t pageSize;
        if(pageSize == 0)
        {
#ifdef _WIN32
            SYSTEM_INFO SYS_INFO;
            GetSystemInfo(&SYS_INFO);
            pageSize = SYS_INFO.dwPageSize;
#else
            pageSize = (size_t)sysconf(_SC_PAGESIZE);
#endif
        }
        return pageSize;
    }

};

} // end namespace AMPS


#endif // _MMAPBOOKMARKSTORE_H_