database.cpp

一个功能强大的内存数据库源代码,c++编写,有详细的注释

      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;	
      default:
	return;
    }	
    abort();
#endif
}

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[].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 < items(metaTableFields); i++) { 
	varyingSize += strlen(metaTableFields[i].name) + 3;
	
    }
    offs_t metaTableOffs = allocate(sizeof(dbTable)
				    + sizeof(dbField)*items(metaTableFields)
				    + varyingSize);
    index[0][dbMetaTableId] = metaTableOffs;
    dbTable* table = (dbTable*)(baseAddr + metaTableOffs);
    table->size = sizeof(dbTable) + sizeof(dbField)*items(metaTableFields)
	        + varyingSize;
    table->next = table->prev = 0;
    int offs = sizeof(dbTable) + sizeof(dbField)*items(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 = items(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 < items(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->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);
    }
}

bool dbDatabase::open(char const* dbName, char const* fiName, 
		      time_t waitLockTimeoutMsec, time_t commitDelaySec)
{
    dbWaitLockTimeout = waitLockTimeoutMsec;
    delete[] databaseName;
    delete[] fileName;
    commitDelay = 0;
    commitTimeout = 0;
    commitTimerStarted = 0;
    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;
    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");
	return false;
    }
    monitor = shm.get();
    sprintf(name, "%s.ws", dbName);
    if (!writeSem.open(name)) { 
	handleError(DatabaseOpenError, 
		    "Failed to initialize database writers semaphore");
	return false;
    }
    sprintf(name, "%s.rs", dbName);
    if (!readSem.open(name)) { 
	handleError(DatabaseOpenError, 
		    "Failed to initialize database readers semaphore");
	return false;
    }
    sprintf(name, "%s.us", dbName);
    if (!upgradeSem.open(name)) { 
	handleError(DatabaseOpenError, 
		    "Failed to initialize database upgrade semaphore");
	return false;
    }
    sprintf(name, "%s.bce", dbName);
    if (!backupCompletedEvent.open(name)) { 
	handleError(DatabaseOpenError, 
		    "Failed to initialize database backup completed event");
	return false;
    }    
    if (commitDelaySec != 0) { 
	sprintf(name, "%s.dce", dbName);
	if (!delayedCommitStopTimerEvent.open(name)) { 
	    handleError(DatabaseOpenError, 
			"Failed to initialize delayed commit event");
	    return false;
	}    
	delayedCommitStartTimerEvent.open();
        commitThreadSyncEvent.open();
    }
    backupInitEvent.open();
    backupFileName = NULL;

    allocatedSize = 0;
	size_t indexSize = initIndexSize < dbFirstUserId 
	? size_t(dbFirstUserId) : initIndexSize;
    indexSize = DOALIGN(indexSize, dbHandlesPerPage);
	    
    size_t fileSize = initSize ? initSize : dbDefaultInitDatabaseSize;
    fileSize = DOALIGN(fileSize, dbBitmapSegmentSize);

    if (fileSize < indexSize*sizeof(offs_t)*4) {
	fileSize = indexSize*sizeof(offs_t)*4;
    }
#ifdef DISKLESS_CONFIGURATION
    mmapSize = fileSize;
#else
    mmapSize = 0;
#endif

    for (int i = dbBitmapId + dbBitmapPages; --i >= 0;) { 
	bitmapPageAvailableSpace[i] = INT_MAX;
    }
    currRBitmapPage = currPBitmapPage = dbBitmapId;
    currRBitmapOffs = currPBitmapOffs = 0;
    reservedChain = NULL;
    tables = NULL;
    modified = false;
    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");
	    return false;
	}
        if (accessType == dbConcurrentUpdate) { 
            sprintf(name, "%s.mcs", dbName);
            if (!mutatorCS.create(name, &monitor->mutatorSem)) { 
                handleError(DatabaseOpenError,
                            "Failed to initialize database monitor");
                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->delayedCommitContext = NULL;
	monitor->concurrentTransId = 1;
	monitor->commitInProgress = false;
        monitor->uncommittedChanges = false;
	memset(monitor->dirtyPagesMap, 0, dbDirtyPageBitmapSize);
    
	sprintf(databaseName, "%s.%d", dbName, version);
	if (file.open(fileName, databaseName, 
		      accessType == dbReadOnly || accessType == dbConcurrentRead, fileSize, false) != dbFile::ok)
	{
	    handleError(DatabaseOpenError, "Failed to create database file");
	    return false;
	}
	baseAddr = (byte*)file.getAddr();
	fileSize = file.getSize();
	header = (dbHeader*)baseAddr;
	updatedRecordId = 0;
	
	if ((unsigned)header->curr > 1) { 
	    handleError(DatabaseOpenError, "Database file was corrupted: "
			"invalid root index");
	    return false;
	}
	if (header->initialized != 1) {
	    if (accessType == dbReadOnly || accessType == dbConcurrentRead) { 
		handleError(DatabaseOpenError, "Can not open uninitialized "
			    "file in read only mode");
		return false;
	    }
	    monitor->curr = header->curr = 0;
	    header->size = fileSize;
	    size_t used = dbPageSize;
	    header->root[0].index = used;
	    header->root[0].indexSize = indexSize;
	    header->root[0].indexUsed = dbFirstUserId;
	    header->root[0].freeList = 0;
	    used += indexSize*sizeof(offs_t);
	    header->root[1].index = used;
	    header->root[1].indexSize = indexSize;
	    header->root[1].indexUsed = dbFirstUserId;
	    header->root[1].freeList = 0;
	    used += indexSize*sizeof(offs_t);

	    header->root[0].shadowIndex = header->root[1].index;
	    header->root[1].shadowIndex = header->root[0].index;
	    header->root[0].shadowIndexSize = indexSize;
	    header->root[1].shadowIndexSize = indexSize;
	    
	    header->majorVersion= FASTDB_MAJOR_VERSION;
	    header->minorVersion = FASTDB_MINOR_VERSION;

	    index[0] = (offs_t*)(baseAddr + header->root[0].index);
	    index[1] = (offs_t*)(baseAddr + header->root[1].index);
	    index[0][dbInvalidId] = dbFreeHandleMarker;

	    size_t bitmapPages = 
		(used + dbPageSize*(dbAllocationQuantum*8-1) - 1)
		/ (dbPageSize*(dbAllocationQuantum*8-1));
	    memset(baseAddr+used, 0xFF, (used + bitmapPages*dbPageSize)
		                        / (dbAllocationQuantum*8));
	    size_t i;
	    for (i = 0; i < bitmapPages; i++) { 
		index[0][dbBitmapId + i] = used + dbPageObjectMarker;
		used += dbPageSize;
	    }
	    while (i < dbBitmapPages) { 
		index[0][dbBitmapId+i] = dbFreeHandleMarker;
		i += 1;
	    }
	    currIndex = index[0];
	    currIndexSize = dbFirstUserId;
	    committedIndexSize = 0;
	    initializeMetaTable();
	    header->dirty = true;
	    memcpy(index[1], index[0], currIndexSize*sizeof(offs_t));
	    file.markAsDirty(0, used);
	    file.flush(true);
	    header->initialized = true;
	    file.markAsDirty(0, sizeof(dbHeader));
	    file.flush(true);
	} else {
	    monitor->curr = header->curr;
	    if (header->dirty) { 
		TRACE_MSG(("Database was not normally closed: "
			   "start recovery\n"));
		if (accessType == dbReadOnly || accessType == dbConcurrentRead) { 
		    handleError(DatabaseOpenError,
				"Can not open dirty file in read only moode");
		    return false;
		}
		recovery();
		TRACE_MSG(("Recovery completed\n"));
	    } else { 
		if (file.getSize() != header->size) { 
		    handleError(DatabaseOpenError, "Database file was "
				"corrupted: file size in header differs "
				"from actual file size");
		    return false;
		}
	    }	    
	}
	if (!loadScheme(true)) { 
	    return false;
	}
	initMutex.done();
    } else { 
	sprintf(name, "%s.cs", dbName);
	if (!cs.open(name, &monitor->sem)) { 
	    handleError(DatabaseOpenError, "Failed to open shared semaphore");
	    return false;
	}
        if (accessType == dbConcurrentUpdate) { 
            sprintf(name, "%s.mcs", dbName);
            if (!mutatorCS.open(name, &monitor->mutatorSem)) { 
                handleError(DatabaseOpenError, "Failed to open shared semaphore");
                return false;
            }
        }
	version = 0;
	if (!loadScheme(false)) { 
	    return false;
	}
    }
    cs.enter();            
    monitor->users += 1;
    cs.leave();
    opened = true;

    if (commitDelaySec != 0) { 
	dbCriticalSection cs(delayedCommitStartTimerMutex);
        commitTimeout = commitDelay = commitDelaySec;
	commitThread.create((dbThread::thread_proc_t)delayedCommitProc, this);
        commitThreadSyncEvent.wait(delayedCommitStartTimerMutex);
    }
    return true;
}

void dbDatabase::scheduleBackup(char const* fileName, time_t period)
{
    if (backupFileName == NULL) { 
	backupFileName = new char[strlen(fileName) + 1];
	strcpy(backupFileName, fileName);
	backupPeriod = period;
	backupThread.create((dbThread::thread_proc_t)backupSchedulerProc, this);
    }
}
 
void dbDatabase::backupScheduler() 
{ 
    backupThread.setPriority(dbThread::THR_PRI_LOW);
    dbCriticalSection cs(backupMutex); 
    while (true) { 
	time_t timeout = backupPeriod;
	if (backupFileName[strlen(backupFileName)-1] != '?') {
	    struct stat st;
	    if (::stat(backupFileName, &st) == 0) { 
		time_t howOld = time(NULL) - st.st_atime;
		if (timeout < howOld) { 
		    timeout = 0;
		} else { 
		    timeout -= howOld;
		}
	    }
	}
	
	backupInitEvent.wait(backupMutex, timeout*1000);
	
	if (backupFileName != NULL) { 
	    if (backupFileName[strlen(backupFileName)-1] == '?') {
		time_t currTime = time(NULL);
		char* fileName = new char[strlen(backupFileName) + 32];
		struct tm* t = localtime(&currTime);
		sprintf(fileName, "%.*s-%04d.%02d.%02d_%02d.%02d.%02d", 
			(int)strlen(backupFileName)-1, backupFileName,
			t->tm_year + 1900, t->tm_mon+1, t->tm_mday, 
			t->tm_hour, t->tm_min, t->tm_sec);
		backup(fileName, false);
		delete[] fileName;
	    } else { 
		char* newFileName = new char[strlen(backupFileName) + 5];
		sprintf(newFileName,"%s.new", backupFileName);
		backup(newFileName, false);