database.h

FastDb是高效的内存数据库系统

     * @param table table descriptor
     * @param ref   [out] pointer to the references where ID of created object will be stored
     * @param record pointer to the transient object to be inserted in the table
     */
    void insertRecord(dbTableDescriptor* table, dbAnyReference* ref, 
                      void const* record);

    /**
     * Check if database is opened
     */
    bool isOpen() const { return opened; }


    /**
     * Check if current transaction is committed. Used mostly for debugging purposes.
     */
    bool isCommitted();

    /**
     * Check if thread was attached to the database. Used mostly for debugging purposes.
     */
    bool isAttached();

    /**
     * Check if current transaction is updating database
     */
    bool isUpdateTransaction();

    /**
     * Get database version
     */
    int  getVersion();

    /**
     * Specify database file size limit. Attempt to exeed this limit cause database error.
     * @param limit maximal file size in bytes
     */
    void setFileSizeLimit(size_t limit) { 
        dbFileSizeLimit = limit;
    }

#ifdef FUZZY_CHECKPOINT
    /**
     * Specify maximal length of qeueue of write requests during fuzzy checkpoint.
     * If this limit is reached, commit is blocked until some of the requests are proceeded
     * @param nPages maximal number of pages in write queue
     */
    void setFuzzyCheckpointBuffer(size_t nPages) { 
        file.setCheckpointBufferSize(nPages);
    }
#endif

#ifndef NO_MEMBER_TEMPLATES
    /**
     * Insert record in the database
     * @param record transient object to be insrted in the database
     * @return reference to the created object
     */
    template<class T>
    dbReference<T> insert(T const& record) {
        dbReference<T> ref;
        insertRecord(lookupTable(&T::dbDescriptor), &ref, &record);
        return ref;
    }
#endif
    /**
     * Find cloned table desciptor assigned to this database 
     * @param des static unassigned table descriptor
     * @return clone of this table descriptor assigned to this databae or NULL
     * if not found.
     */
    dbTableDescriptor* lookupTable(dbTableDescriptor* desc);

    enum dbAccessType { 
        dbReadOnly,
        dbAllAccess,
        dbConcurrentRead,
        dbConcurrentUpdate
    };
    /**
     * Database constructor
     * @param type access type: <code>dbDatabase::dbReadOnly</code> or <code>dbDatabase::dbAllAcces</code>
     * @param dbInitSize initial size of the database. If FastDB is compiled with 
     * DISKLESS_CONFIGURATION option, then in this parameter <B>MAXIMAL</B> size of the 
     * database should be specified (in this mode database can not be reallocated)
     * @param dbExtensionQuantum quentum for extending memory allocation bitmap 
     * @param dbInitIndexSize initial index size (objects)
     * @param nThreads concurrency level for sequential search and sort operations 
     * @see setConcurrency(unsigned nThreads)
     */
    dbDatabase(dbAccessType type = dbAllAccess,
               size_t dbInitSize = dbDefaultInitDatabaseSize,
               size_t dbExtensionQuantum = dbDefaultExtensionQuantum,
               size_t dbInitIndexSize = dbDefaultInitIndexSize,
               int nThreads = 1 
               // Do not specify the last parameter - it is only for checking
               // that application and FastDB library were built with the 
               // same compiler options (-DNO_PTHREADS is critical)
               // Mismached parameters should cause linker error
#ifdef NO_PTHREADS
               , bool usePthreads = false
#endif
               );
    /** 
     * Database detructor
     */
    virtual ~dbDatabase(); 

    const dbAccessType accessType;
    const size_t initSize;
    const size_t extensionQuantum;
    const size_t initIndexSize;

    static unsigned dbParallelScanThreshold; 

  protected:
    static size_t internalObjectSize[];

    dbThreadPool threadPool;

    dbThreadContext<dbDatabaseThreadContext> threadContext;

    byte*     baseAddr;         // base address of database file mapping
    dbHeader* header;           // database header information
    offs_t*   currIndex;        // current database object index
    offs_t*   index[2];
    unsigned  parThreads;
    bool      modified;

    size_t    currRBitmapPage;  //current bitmap page for allocating records
    size_t    currRBitmapOffs;  //offset in current bitmap page for allocating
                                //unaligned records
    size_t    currPBitmapPage;  //current bitmap page for allocating page objects
    size_t    currPBitmapOffs;  //offset in current bitmap page for allocating
                                //page objects
    struct dbLocation { 
        offs_t      pos;
        size_t      size;
        dbLocation* next;
    };
    dbLocation* reservedChain;
    
    char*     databaseName;
    int       databaseNameLen;
    char*     fileName;
    int       version;

    size_t    mmapSize;
    
    size_t    committedIndexSize;
    size_t    currIndexSize;
    oid_t     updatedRecordId;

    unsigned  dbWaitLockTimeout;

    size_t    dbFileSizeLimit;

    bool      uncommittedChanges;

    dbFile                    file;
    dbSharedObject<dbMonitor> shm;
    dbGlobalCriticalSection   cs;
    dbGlobalCriticalSection   mutatorCS;
    dbInitializationMutex     initMutex;
    dbSemaphore               writeSem; 
    dbSemaphore               readSem; 
    dbSemaphore               upgradeSem; 
    dbEvent                   backupCompletedEvent;
    dbMonitor*                monitor;

    dbTableDescriptor*        tables;

    int*                      bitmapPageAvailableSpace;
    bool                      opened;

    long                      allocatedSize;
    
    time_t                    commitDelay;     
    time_t                    commitTimeout;
    time_t                    commitTimerStarted;
    
    dbMutex                   delayedCommitStartTimerMutex;
    dbMutex                   delayedCommitStopTimerMutex;
    dbLocalEvent              delayedCommitStartTimerEvent; 
    dbEvent                   delayedCommitStopTimerEvent; 
    dbLocalEvent              commitThreadSyncEvent;
    bool                      delayedCommitEventsOpened;

    dbMutex                   backupMutex;    
    dbLocalEvent              backupInitEvent;
    char*                     backupFileName;
    time_t                    backupPeriod;
    bool                      stopDelayedCommitThread;

    dbThread                  backupThread;
    dbThread                  commitThread;
 
    int                       accessCount;

    dbL2List                  threadContextList;
    dbMutex                   threadContextListMutex;

    dbErrorHandler            errorHandler;
    void*                     errorHandlerContext;
    int                       schemeVersion;
    dbVisitedObject*          visitedChain;

    bool                      confirmDeleteColumns;

    int                       maxClientId;
    int                       selfId;
#ifdef AUTO_DETECT_PROCESS_CRASH
    dbWatchDog                selfWatchDog;
    dbL2List                  watchDogThreadContexts;
    dbMutex*                  watchDogMutex;
#endif
    
   /**
     * Loads all class descriptors. This method should be used SubSQL and any other apllication
     * which is should work with ANY database file.     
     * @return metatable descriptor
     */
    dbTableDescriptor* loadMetaTable();

    void cleanup(dbInitializationMutex::initializationStatus status, int step);

    void delayedCommit();
    void backupScheduler();

    static void thread_proc delayedCommitProc(void* arg) { 
        ((dbDatabase*)arg)->delayedCommit();
    }

    static void thread_proc backupSchedulerProc(void* arg) { 
        ((dbDatabase*)arg)->backupScheduler();
    }

    virtual bool isReplicated();

#ifdef AUTO_DETECT_PROCESS_CRASH
    /**
     * Revoke lock of crashed client
     * @param clientId ID of crashed client
     */
    void revokeLock(int clientId);

    /**
     * Watch dog thread procedure
     * @param ctx watch dog context
     */
    static void watchDogThread(dbWatchDogContext* ctx);

    /**
     * Start watch dog threads for new clients
     */
    void startWatchDogThreads();
    
    /**
     * Add information about lock owner
     */
    void addLockOwner();

    /**
     * Remove information about lock owner
     */
    void removeLockOwner(int clientId);
#endif

    /**
     * Commit transaction
     * @param ctx thread context
     */
    void commit(dbDatabaseThreadContext* ctx);

    /**
     * Restore consistency of table list of rows (last record should contain null reference     
     * in next field). This method is used during recovery after crash and during rollback.
     */
    void restoreTablesConsistency();

    /**
     * Get table row
     * @param oid object indentifier
     * @return object with this oid
     */
    dbRecord* getRow(oid_t oid) { 
        assert(!(currIndex[oid]&(dbFreeHandleMarker|dbInternalObjectMarker)));
        return (dbRecord*)(baseAddr + currIndex[oid]); 
    }

    /**
     * Prepare for row insertion or update. If record with such OID
     * not exists or it is first time when it was changed during this transaction or
     * size of recrod is changed, than new record is alocated in the database.
     * Otherwisepointer to existed recordis returned.
     * @param oid object indetifier
     * @param newSize size of new object
     * @return pointer inside database where object should should be stored
     */
    dbRecord* putRow(oid_t oid, size_t newSize); 

    /**
     * Prepare for object update ithout changing its size
     * @param oid object indetifier
     * @return pointer inside database where object should should be stored
     */ 
    dbRecord* putRow(oid_t oid) { 
        if (oid < committedIndexSize && index[0][oid] == index[1][oid]) { 
            size_t size = getRow(oid)->size;
            size_t pageNo = oid/dbHandlesPerPage;
            monitor->dirtyPagesMap[pageNo >> 5] |= 1 << (pageNo & 31);
            cloneBitmap(currIndex[oid], size);
            allocate(size, oid);
        } 
        return (dbRecord*)(baseAddr + currIndex[oid]); 
    }

    /**
     * Get record by OID
     * @param oid object identifier
     * @return pointer to the record inside database
     */
    byte* get(oid_t oid) { 
        return baseAddr + (currIndex[oid] & ~dbInternalObjectMarker); 
    }

    /**
     * Prepare for update of internal object.
     * @param oid internal object identifier
     * @return pointer to the record inside database
     */
    byte* put(oid_t oid) { 
        if (oid < committedIndexSize && index[0][oid] == index[1][oid]) { 
            offs_t offs = currIndex[oid];
            size_t size = internalObjectSize[offs & dbInternalObjectMarker];
            size_t pageNo = oid/dbHandlesPerPage;
            monitor->dirtyPagesMap[pageNo >> 5] |= 1 << (pageNo & 31);
            allocate(size, oid);
            cloneBitmap(offs & ~dbInternalObjectMarker, size);
        } 
        return baseAddr + (currIndex[oid] & ~dbInternalObjectMarker); 
    }
    
    /**
     * Check whether query is prefix search: "'key' like field+'%'"
     * @param cursor result set
     * @param expr evaluated expression
     * @param andExpr if not null, then it is used as filter to all records selected by
     * index search
     * @param indexedFile [out] used to return information about which field was used 
     * to perfrom index search and so order in which selected records are sorted. 
     * If this order is the same as requested by "order by" clause, then no extra sorting
     * is needed.
     * @return true if search was performed using indeices,  false if index is not applicable 
     * and sequential search is required
     */
    bool isPrefixSearch(dbAnyCursor* cursor, 
                        dbExprNode* expr, dbExprNode* andExpr, 
                        dbFieldDescriptor* &indexedField);

    /**
     * Check whether search can be performed using indices
     * @param cursor result set
     * @param expr evaluated expression
     * @param andExpr if not null, then it is used as filter to all records selected by
     * index search
     * @param indexedFile [out] used to return information about which field was used 
     * to perfrom index search and so order in which selected records are sorted. 
     * If this order is the same as requested by "order by" clause, then no extra sorting
     * is needed.
     * @return true if search was performed using indeices,  false if index is not applicable 
     * and sequential search is required
     */
    bool isIndexApplicable(dbAnyCursor* cursor, 
                           dbExprNode* expr, dbExprNode* andExpr, 
                           dbFieldDescriptor* &indexedField);

    /**
     * Check whether expression can be evaluated unsing index.
     * If index is applicable, than index search is performed and result
     * is stored in the cursor.
     * @param cursor result set
     * @param expr evaluated expression
     * @param andExpr if not null, then it is used as filter to all records selected by
     * index search
     * @return true if expression was evaluated using index, false if index is not applicable 
     * and sequential search is required
     */
    bool isIndexApplicable(dbAnyCursor* cursor, 
                           dbExprNode* expr, dbExprNode* andExpr);

    /**
     * If query predicate contains operands from other tables (accessed by references)
     * and inverse references exists, then FastDB performs
     * indexed search in referenced table and then go back using inverse referenced to
     * query table. followInverseReference method performs this backward traversal of inverse 
     * references.
     * @param expr evaluated expression
     * @param andExpr if not null, then it is used as filter to all records selected by
     * index search
     * @param cursor cursor to collect selected records
     * @param iref OID of the selected records in referenced table
     */
    bool followInverseReference(dbExprNode* expr, dbExprNode* andExpr, 
                                dbAnyCursor* cursor, oid_t iref);

    /**
     * Check if there is inverse reference in the table rerefrenced from search predicate.
     * @param expr evaluated expression
     * @param nExistsClause number of exists clauses in search wrapping this expression
     * @return true if inverse reference(s) exists and it is possible to perform backward
     * traversal
     */
    bool existsInverseReference(dbExprNode* expr, int nExistsClauses);

    /**
     * Execute expression. This method is most frequently recursivly called during 
     * evaluation of search predicate.
     * @param expr expression to be executed
     * @param iattr inherited attributes - attributes passed top-down 
     * (information like cursor, current record, ...)
     * @param sattr synthesized attribute - sttributes passed down-top
     * (value of expression)
     */
    static void _fastcall execute(dbExprNode* expr, 
                                  dbInheritedAttribute& iattr,
                                  dbSynthesizedAttribute& sattr);

    /**
     * Evaluate epression. This method initialie initiainherited attributes and invoke
     * execute method
     * @param expr expression to be evaluated
     * @param oid OID of the inspected record
     * @param seaqched table
     * @param cursor result set
     * @return true if this record match search condition, false otherwise
     */
    bool evaluate(dbExprNode* expr, oid_t oid, dbTable* table, dbAnyCursor* cursor);

    /**
     * Select all records from the table
     * @param cursor result set
     */
    void select(dbAnyCursor* cursor);

    /**
     * Execute select query
     * @param cursor result set
     * @param query query expression
     */
    void select(dbAnyCursor* cursor, dbQuery& query);

    /**
     * Perform table traverse: execute queries with "start from (follow by)" clause
     * @param cursor result set
     * @param query  query expression
     */
    void traverse(dbAnyCursor* cursor, dbQuery& query);

    /**
     * Update record
     * @param oid record identifier
     * @param table descriptor of the table to which record belongs
     * @param record updated image of the record
     */
    void update(oid_t oid, dbTableDescriptor* table, void const* record);
    
    /**
     * Remove record from the database
     * @param table descriptor of the table to which record belongs
     * @param oid record identifier
     */
    void remove(dbTableDescriptor* table, oid_t oid);