UserSet.h

Go to the documentation of this file.

/*****************************************************************************
 *                                                                           *
 *  Copyright 2018 Rice University                                           *
 *                                                                           *
 *  Licensed under the Apache License, Version 2.0 (the "License");          *
 *  you may not use this file except in compliance with the License.         *
 *  You may obtain a copy of the License at                                  *
 *                                                                           *
 *      http://www.apache.org/licenses/LICENSE-2.0                           *
 *                                                                           *
 *  Unless required by applicable law or agreed to in writing, software      *
 *  distributed under the License is distributed on an "AS IS" BASIS,        *
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. *
 *  See the License for the specific language governing permissions and      *
 *  limitations under the License.                                           *
 *                                                                           *
 *****************************************************************************/

#ifndef SRC_CPP_MAIN_DATABASE_HEADERS_USERSET_H_
#define SRC_CPP_MAIN_DATABASE_HEADERS_USERSET_H_

#include "PDBDebug.h"
#include "PartitionedFile.h"
#include "PageCache.h"
#include "PDBLogger.h"
#include "DataTypes.h"
#include "PageIterator.h"
#include "PageCircularBuffer.h"
#include "SequenceID.h"
#include <set>
#include <vector>
#include <memory>
#include "LocalitySet.h"
using namespace std;

class UserSet;
typedef shared_ptr<UserSet> SetPtr;

class UserSet : public LocalitySet {
public:
    UserSet(pdb::PDBLoggerPtr logger,
            SharedMemPtr shm,
            NodeID nodeId,
            DatabaseID dbId,
            UserTypeID typeId,
            SetID setId,
            string setName,
            PageCachePtr pageCache,
            LocalityType localityType = JobData,
            LocalitySetReplacementPolicy policy = MRU,
            OperationType operation = Read,
            DurabilityType durability = TryCache,
            PersistenceType persistence = Persistent,
            size_t pageSize = DEFAULT_PAGE_SIZE);

    UserSet(size_t pageSize,
            pdb::PDBLoggerPtr logger,
            SharedMemPtr shm,
            NodeID nodeId,
            DatabaseID dbId,
            UserTypeID typeId,
            SetID setId,
            string setName,
            PartitionedFilePtr file,
            PageCachePtr pageCache,
            LocalityType localityType = JobData,
            LocalitySetReplacementPolicy policy = MRU,
            OperationType operation = Read,
            DurabilityType durability = TryCache,
            PersistenceType persistence = Persistent);

    ~UserSet();

    bool addObject(PDBObjectPtr object, PDBPagePtr page);


    bool addObject(PDBObjectPtr object);

    void pinBufferPage();


    void unpinBufferPage() {
        if (this->inputBufferPage != nullptr) {
            this->inputBufferPage->decRefCount();
        }
    }


    PDBPagePtr getPageFromFile(FilePartitionID partitionId, unsigned int pageSeqInPartition);


    PDBPagePtr addPage();

    PDBPagePtr addPageByRawBytes(size_t sharedMemOffset);


    inline void* getNewBytes(size_t size, bool evictWhenUnpin = false) {
        if (size == 0) {
            return nullptr;
        }
        pthread_mutex_lock(&this->addBytesMutex);
        if (this->inputBufferPage == nullptr) {

            this->inputBufferPage = this->addPage();
        }
        void* buffer = this->inputBufferPage->addVariableBytes(size);
        if (buffer == nullptr) {
            // current inputBufferPage is full

            // we unpin the inputBufferPage
            this->inputBufferPage->decRefCount();
            /*if(this->getDurabilityType() == CacheThrough) {
               CacheKey key;
               key.dbId = this->getDbID();
               key.typeId = this->getTypeID();
               key.setId = this->getSetID();
               key.pageId = this->inputBufferPage->getPageID();
               this->pageCache->flushPageWithoutEviction(key);
            }*/
            if (evictWhenUnpin == true) {
                this->pageCache->evictPage(this->inputBufferPage);
            }


            // we add a new page as inputBufferPage
            this->inputBufferPage = this->addPage();
            buffer = this->inputBufferPage->addVariableBytes(size);
            pthread_mutex_unlock(&this->addBytesMutex);
            // get new bytes in the new page
            return buffer;
        }
        pthread_mutex_unlock(&this->addBytesMutex);
        return buffer;
    }


    virtual vector<PageIteratorPtr>* getIterators();

    PDBPagePtr getPage(FilePartitionID partitionId, unsigned int pageSeqInPartition, PageID pageId);

    int getNumPages();


    DatabaseID getDbID() {
        return this->dbId;
    }


    UserTypeID getTypeID() {
        return this->typeId;
    }


    SetID getSetID() {
        return this->setId;
    }

    string getSetName() {
        return this->setName;
    }

    PartitionedFilePtr getFile() {
        return this->file;
    }

    void setFile(PartitionedFilePtr file) {
        this->file = file;
    }

    PageID getLastFlushedPageId() {
        return this->lastFlushedPageId;
    }


    void setLastFlushedPageId(PageID pageId) {
        this->lastFlushedPageId = pageId;
    }

    void addPageToDirtyPageSet(PageID pageId) {
        lockDirtyPageSet();
        auto search = dirtyPagesInPageCache->find(pageId);
        if (search != dirtyPagesInPageCache->end()) {
            cout << "UserSet: Error in addPageToDirtyPageSet, key exists.\n";
        } else {
            FileSearchKey key;
            key.inCache = true;
            key.partitionId = (unsigned int)(-1);
            key.pageSeqInPartition = (unsigned int)(-1);
            dirtyPagesInPageCache->insert({pageId, key});
        }
        unlockDirtyPageSet();
    }

    void removePageFromDirtyPageSet(PageID pageId,
                                    FilePartitionID partitionId,
                                    unsigned int pageSeqInPartition) {
        if (isPinned == false) {
            PDB_COUT << "the set is not pinned" << std::endl;
            dirtyPagesInPageCache->erase(pageId);
            return;
        }
        PDB_COUT << "the set is pinned" << std::endl;
        auto search = dirtyPagesInPageCache->find(pageId);
        if (search != dirtyPagesInPageCache->end()) {
            search->second.inCache = false;
            search->second.partitionId = partitionId;
            search->second.pageSeqInPartition = pageSeqInPartition;
        }
        return;
    }

    unordered_map<PageID, FileSearchKey>* getDirtyPageSet() {
        return dirtyPagesInPageCache;
    }

    void lockDirtyPageSet() {
        pthread_mutex_lock(&this->dirtyPageSetMutex);
    }

    void unlockDirtyPageSet() {
        pthread_mutex_unlock(&this->dirtyPageSetMutex);
    }


    void dump(char* buffer);

    void evictPages();

    bool getPinned() {
        return this->isPinned;
    }

    /*
     * TODO: change the name because current name is easy to be confused with pin()/unpin() defined
     * in LocalitySet class (which is the base class).
     */
    void setPinned(bool isPinned) {
        lockDirtyPageSet();
        this->isPinned = isPinned;
        unlockDirtyPageSet();
    }

    pdb::PDBLoggerPtr getLogger() {
        return this->logger;
    }

    void cleanDirtyPageSet();

    // MUST be used after set->setPinned(true) is invoked
    void flushDirtyPages();

    size_t getPageSize() {
        return this->pageSize;
    }

    void setPageSize(size_t pageSize) {
        this->pageSize = pageSize;
    }

    bool getSorted () {
        return this->isSorted;
    }

    void setSorted (bool isSorted) {
        this->isSorted = isSorted;
    }

    bool getPartitioned () {
        return this->isPartitioned;
    }

    void setPartitioned (bool isPartitioned) {
        this->isPartitioned = isPartitioned;
    }


protected:
    PartitionedFilePtr file = nullptr;
    PageCachePtr pageCache = nullptr;
    SharedMemPtr shm;
    pdb::PDBLoggerPtr logger = nullptr;
    NodeID nodeId;
    DatabaseID dbId;
    UserTypeID typeId;
    SetID setId;
    string setName;
    PageID lastFlushedPageId;
    PageID latestPageId;
    PDBPagePtr inputBufferPage = nullptr;
    SequenceID seqId;
    unordered_map<PageID, FileSearchKey>* dirtyPagesInPageCache = nullptr;
    pthread_mutex_t dirtyPageSetMutex;
    bool isPinned = false;
    int numPages;
    pthread_mutex_t addBytesMutex;
    size_t pageSize;
    bool isSorted = false;
    bool isPartitioned = false;


};


#endif /* SRC_CPP_MAIN_DATABASE_HEADERS_PDBSET_H_ */