database.cpp

俄罗斯牛人KK的作品,著名的ORDBMS,这里上传最新的3.39版本源代码.希望了解对象关系数据库的同好,请不要错过.

char const* const dbDatabase::errorMessage[] =
{
    "No error", 
    "Query syntax error",
    "Arithmetic exception",
    "Index out of range",
    "Database open error",
    "File access error",
    "Out of memory",
    "Deadlock",
    "Null reference",
    "Lock revoked",
    "File limit exeeded",
    "Inconsistent inverse reference",
    "Attempt to modify read-only database"
};

void dbDatabase::handleError(dbErrorClass error, char const* msg, int arg)
{
    if (errorHandler != NULL) { 
        (*errorHandler)(error, msg, arg, errorHandlerContext);
    }
#ifdef THROW_EXCEPTION_ON_ERROR
    if (error != NoError) {
        if (msg == NULL) { 
            msg = errorMessage[error];
        }
        if (error == DatabaseOpenError) {
            fprintf(stderr, "%s\n", msg);
        } else { 
            throw dbException(error, msg, arg);
        }
    }
#else
    char buf[256];
    switch (error) { 
      case QueryError:
        fprintf(stderr, "%s in position %d\n", msg, arg);
        return;
      case ArithmeticError:
        fprintf(stderr, "%s\n", msg);
        break;
      case IndexOutOfRangeError:
        fprintf(stderr, "Index %d is out of range\n", arg);
        break;
      case DatabaseOpenError:
        fprintf(stderr, "%s\n", msg);
        return;
      case FileError:
        fprintf(stderr, "%s: %s\n", msg, 
                dbFile::errorText(arg, buf, sizeof(buf)));
        break;
      case OutOfMemoryError:
        fprintf(stderr,"Not enough memory: failed to allocate %d bytes\n",arg);
        break;
      case NullReferenceError:
        fprintf(stderr, "Null object reference is accessed\n");
        break;
      case Deadlock:
        fprintf(stderr, "Deadlock is caused by upgrading shared locks to exclusive");
        break;
      case LockRevoked:
        fprintf(stderr, "Lock is revoked by some other client");
        break;  
      case InconsistentInverseReference:
        fprintf(stderr, "%s\n", msg);
        return;
      case DatabaseReadOnly:
        fprintf(stderr, "Attempt to modify readonly database");
        break;
      default:
        return;
    }   
    abort();
#endif
}

bool dbDatabase::isReplicated()
{
    return false;
}

void dbDatabase::initializeMetaTable()
{
    static struct { 
        char const* name;
        int         type;
        int         size;
        int         offs;
    } metaTableFields[] = { 
        { "name", dbField::tpString, sizeof(dbVarying), 
          offsetof(dbTable, name)},
        { "fields", dbField::tpArray, sizeof(dbVarying), 
          offsetof(dbTable, fields)},
        { "fields[]", dbField::tpStructure, sizeof(dbField), 0},
        { "fields[].name", dbField::tpString, sizeof(dbVarying), 
          offsetof(dbField, name)},
        { "fields[].tableName",dbField::tpString,sizeof(dbVarying), 
          offsetof(dbField, tableName)},
        { "fields[].inverse", dbField::tpString, sizeof(dbVarying), 
          offsetof(dbField, inverse)},
//        { "fields[].type", dbField::tpInt4, 4, offsetof(dbField, type)},
        { "fields[].type", dbField::tpInt4, 4, offsetof(dbField, offset)-4},
        { "fields[].offset", dbField::tpInt4, 4, offsetof(dbField, offset)},
        { "fields[].size", dbField::tpInt4, 4, offsetof(dbField, size)},
        { "fields[].hashTable", dbField::tpReference, sizeof(oid_t), 
          offsetof(dbField, hashTable)},
        { "fields[].tTree", dbField::tpReference, sizeof(oid_t), 
          offsetof(dbField, tTree)},
        { "fixedSize", dbField::tpInt4, 4, offsetof(dbTable, fixedSize)},
        { "nRows", dbField::tpInt4, 4, offsetof(dbTable, nRows)},
        { "nColumns", dbField::tpInt4, 4, offsetof(dbTable, nColumns)},
        { "firstRow", dbField::tpReference, sizeof(oid_t), offsetof(dbTable, firstRow)},
        { "lastRow", dbField::tpReference, sizeof(oid_t), offsetof(dbTable, lastRow)}
 #ifdef AUTOINCREMENT_SUPPORT
       ,{ "count", dbField::tpInt4, 4, offsetof(dbTable, count)}
 #endif
    };

    unsigned i;
    size_t varyingSize = strlen(dbMetaTableName)+1;
    for (i = 0; i < itemsof(metaTableFields); i++) { 
        varyingSize += strlen(metaTableFields[i].name) + 3;
        
    }
    offs_t metaTableOffs = allocate(sizeof(dbTable)
                                    + sizeof(dbField)*itemsof(metaTableFields)
                                    + varyingSize);
    index[0][dbMetaTableId] = metaTableOffs;
    dbTable* table = (dbTable*)(baseAddr + metaTableOffs);
    table->size = sizeof(dbTable) + sizeof(dbField)*itemsof(metaTableFields)
                + varyingSize;
    table->next = table->prev = 0;
    int offs = sizeof(dbTable) + sizeof(dbField)*itemsof(metaTableFields);
    table->name.offs = offs;
    table->name.size = strlen(dbMetaTableName)+1;
    strcpy((char*)table + offs, dbMetaTableName);
    offs += table->name.size;
    table->fields.offs = sizeof(dbTable);
    table->fields.size = itemsof(metaTableFields);
    table->fixedSize = sizeof(dbTable);
    table->nRows = 0;
    table->nColumns = 5;
    table->firstRow = 0;
    table->lastRow = 0;
#ifdef AUTOINCREMENT_SUPPORT
    table->count = 0;
#endif

    dbField* field = (dbField*)((char*)table + table->fields.offs);
    offs -= sizeof(dbTable);
    for (i = 0; i < itemsof(metaTableFields); i++) { 
        field->name.offs = offs;
        field->name.size = strlen(metaTableFields[i].name) + 1;
        strcpy((char*)field + offs, metaTableFields[i].name);
        offs += field->name.size;

        field->tableName.offs = offs;
        field->tableName.size = 1;
        *((char*)field + offs++) = '\0';
        
        field->inverse.offs = offs;
        field->inverse.size = 1;
        *((char*)field + offs++) = '\0';
        
        field->flags = 0;
        field->type = metaTableFields[i].type;
        field->size = metaTableFields[i].size;
        field->offset = metaTableFields[i].offs;
        field->hashTable = 0;
        field->tTree = 0;
        field += 1;
        offs -= sizeof(dbField);
    }
}

void dbDatabase::cleanup(dbInitializationMutex::initializationStatus status, int step) 
{
    switch (step) { 
      case 9:
        if (status == dbInitializationMutex::NotYetInitialized) {         
            file.close();
        }
        // no break
      case 8:
        if (accessType == dbConcurrentUpdate || accessType == dbConcurrentRead) { 
            mutatorCS.close();
        }
        // no break
      case 7:
        if (delayedCommitEventsOpened) { 
            delayedCommitStopTimerEvent.close();
            delayedCommitStartTimerEvent.close();
            commitThreadSyncEvent.close();
            delayedCommitEventsOpened = false;
        }
        // no break
      case 6:
        cs.close();
        // no break
      case 5:
        backupCompletedEvent.close();
        // no break
      case 4:
        upgradeSem.close();
        // no break
      case 3:
        readSem.close();
        // no break
      case 2:
        writeSem.close();
        // no break
      case 1:
        shm.close();
        // no break
      default:
        if (status == dbInitializationMutex::NotYetInitialized) {         
            initMutex.done();
        }
        initMutex.close();
    }
}

bool dbDatabase::open(OpenParameters& params)
{
    accessType = params.accessType;
    extensionQuantum = params.extensionQuantum;
    initIndexSize = params.initIndexSize; 
    initSize = params.initSize;
    freeSpaceReuseThreshold = params.freeSpaceReuseThreshold;
    parallelScanThreshold = params.parallelScanThreshold;
    setConcurrency(params.nThreads);
    return open(params.databaseName, params.databaseFilePath, params.waitLockTimeoutMsec, params.transactionCommitDelay);
}

bool dbDatabase::open(char const* dbName, char const* fiName, 
                      time_t waitLockTimeoutMsec, time_t commitDelaySec)
{
    waitLockTimeout = (unsigned)waitLockTimeoutMsec;
    delete[] databaseName;
    delete[] fileName;
    commitDelay = 0;
    commitTimeout = 0;
    commitTimerStarted = 0;
    delayedCommitEventsOpened = false;
    backupFileName = NULL;
    backupPeriod = 0;
    opened = false;
    stopDelayedCommitThread = false;
    databaseNameLen = strlen(dbName);
    char* name = new char[databaseNameLen+16];
    sprintf(name, "%s.in", dbName);
    databaseName = name;
    if (fiName == NULL) { 
        fileName = new char[databaseNameLen + 5];
        sprintf(fileName, "%s.fdb", dbName);
    } else { 
        fileName = new char[strlen(fiName)+1];
        strcpy(fileName, fiName);
    }

    dbInitializationMutex::initializationStatus status = initMutex.initialize(name);
    if (status == dbInitializationMutex::InitializationError) { 
        handleError(DatabaseOpenError, 
                    "Failed to start database initialization");
        return false;
    }
    sprintf(name, "%s.dm", dbName);
    if (!shm.open(name)) { 
        handleError(DatabaseOpenError, "Failed to open database monitor");
        cleanup(status, 0);
        return false;
    }
    monitor = shm.get();
    sprintf(name, "%s.ws", dbName);
    if (!writeSem.open(name)) { 
        handleError(DatabaseOpenError, 
                    "Failed to initialize database writers semaphore");
        cleanup(status, 1);
        return false;
    }
    sprintf(name, "%s.rs", dbName);
    if (!readSem.open(name)) { 
        handleError(DatabaseOpenError, 
                    "Failed to initialize database readers semaphore");
        cleanup(status, 2);
        return false;
    }
    sprintf(name, "%s.us", dbName);
    if (!upgradeSem.open(name)) { 
        handleError(DatabaseOpenError, 
                    "Failed to initialize database upgrade semaphore");
        cleanup(status, 3);
        return false;
    }
    sprintf(name, "%s.bce", dbName);
    if (!backupCompletedEvent.open(name)) { 
        handleError(DatabaseOpenError, 
                    "Failed to initialize database backup completed event");
        cleanup(status, 4);
        return false;
    }    
    if (commitDelaySec != 0) { 
        sprintf(name, "%s.dce", dbName);
        delayedCommitEventsOpened = true;
        if (!delayedCommitStopTimerEvent.open(name)) { 
            handleError(DatabaseOpenError, 
                        "Failed to initialize delayed commit event");
            cleanup(status, 5);
            return false;
        }    
        delayedCommitStartTimerEvent.open();
        commitThreadSyncEvent.open();
    }
    backupInitEvent.open();
    backupFileName = NULL;

    fixedSizeAllocator.reset();
    allocatedSize = 0;
    deallocatedSize = 0;
    size_t indexSize = initIndexSize < dbFirstUserId 
        ? size_t(dbFirstUserId) : initIndexSize;
    indexSize = DOALIGN(indexSize, dbHandlesPerPage);
            
    size_t fileSize = initSize ? initSize : dbDefaultInitDatabaseSize;

    if (fileSize < indexSize*sizeof(offs_t)*4) {
        fileSize = indexSize*sizeof(offs_t)*4;
    }
    fileSize = DOALIGN(fileSize, dbBitmapSegmentSize);

    for (int i = dbBitmapId + dbBitmapPages; --i >= 0;) { 
        bitmapPageAvailableSpace[i] = INT_MAX;
    }
    currRBitmapPage = currPBitmapPage = dbBitmapId;
    currRBitmapOffs = currPBitmapOffs = 0;
    reservedChain = NULL;
    tables = NULL;
    modified = false;
    selfId = 0;
    maxClientId = 0;
    threadContextList.reset();
    attach();

    if (status == dbInitializationMutex::NotYetInitialized) { 
        sprintf(name, "%s.cs", dbName);
        if (!cs.create(name, &monitor->sem)) { 
            handleError(DatabaseOpenError, "Failed to initialize database monitor");
            cleanup(status, 6);
            return false;
        }
        if (accessType == dbConcurrentUpdate || accessType == dbConcurrentRead) { 
            sprintf(name, "%s.mcs", dbName);
            if (!mutatorCS.create(name, &monitor->mutatorSem)) { 
                handleError(DatabaseOpenError,
                            "Failed to initialize database monitor");
                cleanup(status, 7);
                return false;
            }
        }
        readSem.reset();
        writeSem.reset();
        upgradeSem.reset();
        monitor->nReaders = 0;
        monitor->nWriters = 0;
        monitor->nWaitReaders = 0;
        monitor->nWaitWriters = 0;
        monitor->waitForUpgrade = false;
        monitor->version = version = 1;
        monitor->users = 0;
        monitor->backupInProgress = 0;
        monitor->forceCommitCount = 0;
        monitor->lastDeadlockRecoveryTime = 0;
        monitor->delayedC