utpersist.c

DataDraw is an ultra-fast persistent database for high performance programs written in C. It s so fa

    while(xCommand < length) {
        size = findCommandSize(commands + xCommand);
        if(!(commands[xCommand] & 0x80)) {
            /* Skip undo commands */
            applyCommand(commands + xCommand);
        }
        xCommand += size;
    }
}

/*--------------------------------------------------------------------------------------------------
  Truncate the recentChanges file
--------------------------------------------------------------------------------------------------*/
static void truncateChangesFile(
    uint32 size)
{
    utWarning("recentChanges file damaged... truncating back to last complete transaction.");
    fclose(utRecentChangesFile);
    utTruncateFile(utSprintf("%s%crecentChanges", utDatabaseDirectory, UTDIRSEP), size);
    openRecentChanges("rb");
}

/*--------------------------------------------------------------------------------------------------
  Apply recent changes from the commands in the recent changes buffer.  utBufferStartPosition and
  lastTransactionPosition are used to help truncate the recentChanges file if it is corrupt at the
  end.  Return true if everything is ok, and false if we had to truncate the file.
--------------------------------------------------------------------------------------------------*/
static bool applyCommandsInBuffer(
    bool endOfInput,
    uint32 *checksum,
    uint32 *unparsedCommands)
{
    uint32 xCommand = 0;
    uint32 size;
    bool done = false;

    while(!done && (xCommand + UT_COMMAND_MAX_HEADER_SIZE < utBufferPosition ||
            (endOfInput && xCommand < utBufferPosition))) {
        if(utCommandBuffer[xCommand] == UT_TRANSACTION_COMPLETE) {
            xCommand++;
            if(readUint32(utCommandBuffer + xCommand) != *checksum) {
                truncateChangesFile(utLastTransactionPosition);
                return false;
            }
            xCommand += 4;
            utTransactionCreate(utLastTransactionPosition, utUsedCommands);
            applyCommands(utCommands, utUsedCommands);
            utUsedCommands = 0;
            utLastTransactionPosition = utBufferStartPosition + xCommand;
            *checksum = 0;
        } else {
            size = findCommandSize(utCommandBuffer + xCommand);
            if(size == 0) {
                truncateChangesFile(utLastTransactionPosition);
                return false;
            }
            if(utUsedCommands + size >= utAllocatedCommands) {
                utAllocatedCommands += (utAllocatedCommands >> 1) + size;
                utResizeArray(utCommands, utAllocatedCommands);
            }
            if(xCommand + size <= utBufferPosition) {
                while(size-- != 0) {
                    *checksum = (*checksum ^ utCommandBuffer[xCommand])*1103515245 + 12345;
                    utCommands[utUsedCommands++] = utCommandBuffer[xCommand++];
                }
            } else {
                done = true;
            }
        }
    }
    *unparsedCommands = utBufferPosition - xCommand;
    if(*unparsedCommands > 0) {
        memmove(utCommandBuffer, utCommandBuffer + xCommand, *unparsedCommands);
    }
    return true;
}

/*--------------------------------------------------------------------------------------------------
  Apply recent changes to the database.  The changes will have been recorded in recentChanges.
--------------------------------------------------------------------------------------------------*/
static void applyRecentChanges(void)
{
    uint32 checksum = 0;
    uint32 unparsedCommands = 0;

    utAssert(utChecksum == 0);
    utLastTransactionPosition = 0;
    utBufferStartPosition = 0;
    utUsedCommands = 0;
    fclose(utRecentChangesFile);
    openRecentChanges("rb");
    utDo {
        if(unparsedCommands + UT_CHANGE_BUFFER_LENGTH > utBufferSize) {
            utBufferSize = unparsedCommands + UT_CHANGE_BUFFER_LENGTH;
            utResizeArray(utCommandBuffer, utBufferSize);
        }
        utBufferPosition = fread(utCommandBuffer + unparsedCommands, sizeof(uint8),
            UT_CHANGE_BUFFER_LENGTH, utRecentChangesFile) + unparsedCommands;
    } utWhile(utBufferPosition > 0 &&
            applyCommandsInBuffer(utBufferPosition < UT_CHANGE_BUFFER_LENGTH + unparsedCommands,
            &checksum, &unparsedCommands)) {
        utBufferStartPosition += utBufferPosition - unparsedCommands;
    } utRepeat;
    fclose(utRecentChangesFile);
    openRecentChanges("ab");
    utBufferPosition = 0;
    utAssert(utChecksum == 0);
}

/*--------------------------------------------------------------------------------------------------
  Load the module from the file.
--------------------------------------------------------------------------------------------------*/
static bool loadModule(
    utModule module,
    FILE *file)
{
    utClass theClass;
    utField field;
    uint16 xField;
    uint8 *values;
    uint64 numUsed, numAllocated;

    module->stop();
    if(fread(module->globalData, module->globalSize, 1, file) != 1) {
        utWarning("File too short, and does not contain header info");
        return false;
    }
    if(module->hashValue != *(uint32 *)(module->globalData)) {
        utWarning("Incompatible database format");
        return false;
    }
    for(xField = 0; xField < module->numFields; xField++) {
        field = utFields + module->firstFieldIndex + xField;
        theClass = utClasses + field->classIndex;
        if(field->array) {
            numUsed = *(field->numUsedPtr);
            numAllocated =  *(field->numAllocatedPtr);
        } else {
            numUsed = utFindIntValue(theClass->numUsedPtr, theClass->referenceSize);
            numAllocated = utFindIntValue(theClass->numAllocatedPtr, theClass->referenceSize);
        }
        values = utCalloc(numAllocated, field->size);
        *(uint8 **)(field->arrayPtr) = values;
        if(fread(values, field->size, numUsed, file) != numUsed) {
            utWarning("Unable to read from file");
            return false;
        }
    }
    return true;
}

/*--------------------------------------------------------------------------------------------------
  Load the database from memory.
--------------------------------------------------------------------------------------------------*/
void utLoadBinaryDatabase(
    FILE *databaseFile)
{
    char *fileName;
    utModule module;
    uint32 xModule;
    bool useDefault = databaseFile == NULL;

    if(useDefault) {
        fileName = utSprintf("%s%cdatabase", utDatabaseDirectory, UTDIRSEP);
        databaseFile = fopen(fileName, "rb");
    }
    if(databaseFile == NULL) {
        utExit("Could not read from %s", fileName);
    }
    for(xModule = 0; xModule < utUsedModules; xModule++) {
        module = utModules + xModule;
        if(module->initialized && !loadModule(module, databaseFile)) {
            utExit("Could not read from %s", fileName);
        }
    }
    if(useDefault) {
        fclose(databaseFile);
    }
}

/*--------------------------------------------------------------------------------------------------
  Save the module to the file.
--------------------------------------------------------------------------------------------------*/
static bool saveModule(
    utModule module,
    FILE *file)
{
    utClass theClass;
    utField field;
    uint16 xField;
    uint8 *values;
    uint64 numUsed;

    if(fwrite(module->globalData, module->globalSize, 1, file) != 1) {
        utWarning("Unable to write to file");
        return false;
    }
    for(xField = 0; xField < module->numFields; xField++) {
        field = utFields + module->firstFieldIndex + xField;
        theClass = utClasses + field->classIndex;
        values = *(uint8 **)(field->arrayPtr);
        if(field->array) {
            numUsed = *(field->numUsedPtr);
        } else {
            numUsed = utFindIntValue(theClass->numUsedPtr, theClass->referenceSize);
        }
        if(fwrite(values, field->size, numUsed, file) != numUsed) {
            utWarning("Unable to write to file");
            return false;
        }
    }
    return true;
}

/*--------------------------------------------------------------------------------------------------
  Save the database to disk.
--------------------------------------------------------------------------------------------------*/
void utSaveBinaryDatabase(
    FILE *databaseFile)
{
    char *fileName;
    utModule module;
    uint32 xModule;
    bool useDefault = databaseFile == NULL;

    if(useDefault) {
        fileName = utSprintf("%s%cdatabase", utDatabaseDirectory, UTDIRSEP);
        databaseFile = fopen(fileName, "wb");
    }
    if(databaseFile == NULL) {
        utExit("Could not write to %s", fileName);
    }
    for(xModule = 0; xModule < utUsedModules; xModule++) {
        module = utModules + xModule;
        if(module->initialized && !saveModule(module, databaseFile)) {
            utExit("Could not save database", fileName);
        }
    }
    if(useDefault) {
        fclose(databaseFile);
    }
}

/*--------------------------------------------------------------------------------------------------
  Find the oldest transaction indicated by numChanges, and load it into the database.
--------------------------------------------------------------------------------------------------*/
static void loadChangesIntoCommandBuffer(
    uint32 xTransaction)
{
    utTransaction transaction = utTransactions + xTransaction;
    uint32 filePosition, bufferSize;
    
    if(transaction->position >= utBufferStartPosition) {
        return; /* Already loaded */
    }
    flushRecentChanges();
    if(utBufferStartPosition >= UT_CHANGE_BUFFER_LENGTH) {
        filePosition = utBufferStartPosition - UT_CHANGE_BUFFER_LENGTH;
    } else {
        filePosition = 0;
    }
    filePosition = utMin(filePosition, transaction->position);
    bufferSize = utBufferStartPosition + utBufferPosition - filePosition;
    if(bufferSize > utBufferSize) {
        utBufferSize = bufferSize + (bufferSize >> 1);
        utResizeArray(utCommandBuffer, utBufferSize);
    }
    fclose(utRecentChangesFile);
    openRecentChanges("rb");
    fseek(utRecentChangesFile, filePosition, SEEK_SET);
    utBufferPosition = fread(utCommandBuffer, sizeof(uint8), bufferSize, utRecentChangesFile);
    utAssert(utBufferPosition == bufferSize);
    utBufferStartPosition = filePosition;
    fclose(utRecentChangesFile);
    utTruncateFile(utSprintf("%s%crecentChanges", utDatabaseDirectory, UTDIRSEP), filePosition);
    openRecentChanges("ab");
}

/*--------------------------------------------------------------------------------------------------
  Compute the transaction command positions.
--------------------------------------------------------------------------------------------------*/
static void computeTransactionCommandPositions(
    utTransaction transaction)
{
    uint8 *commands = utCommandBuffer + transaction->position - utBufferStartPosition;
    uint32 xCommand = 0;
    uint32 size;

    utTransactionUsedCommands = 0;
    while(xCommand < transaction->length) {
        size = findCommandSize(commands + xCommand);
        if(utTransactionUsedCommands == utTransactionAllocatedCommands) {
            utTransactionAllocatedCommands += utTransactionAllocatedCommands >> 1;
            utResizeArray(utTransactionCommands, utTransactionAllocatedCommands);
        }
        utTransactionCommands[utTransactionUsedCommands++] = xCommand + transaction->position -
            utBufferStartPosition;
        xCommand += size;
    }
}

/*--------------------------------------------------------------------------------------------------
  Undo the transaction.
--------------------------------------------------------------------------------------------------*/
static void undoTransaction(
    utTransaction transaction)
{
    uint8 *command;
    uint32 xCommand;

    computeTransactionCommandPositions(transaction);
    xCommand = utTransactionUsedCommands;
    while(xCommand-- != 0) {
        command = utCommandBuffer + utTransactionCommands[xCommand];
        if(*command & 0x80) {
            applyCommand(command);
        }
    }
}

/*--------------------------------------------------------------------------------------------------
  Undo the last transaction, by scanning backwards through the recent changes buffer, executing
  undo commands.
--------------------------------------------------------------------------------------------------*/
uint32 utUndo(
    uint32 numChanges)
{
    utTransaction transaction = NULL;
    uint32 xTransaction;
    uint32 firstTransaction, lastTransaction;

    if(utTransactionInProgress) {
        utExit("The current transaction has not yet been completed... can't undo!");
    }
    utAssert(utChecksum == 0);
    if(utRedoTransaction == NULL) {
        firstTransaction = utUsedTransactions;
    } else {
        firstTransaction = utRedoTransaction - utTransactions;
    }
    lastTransaction = firstTransaction >= numChanges? firstTransaction - numChanges : 0;
    if(firstTransaction == lastTransaction) {
        /* Nothing to do */
        return 0;
    }
    if(utPersistenceInitialized) {
        loadChangesIntoCommandBuffer(lastTransaction);
    }
    for(xTransaction = firstTransaction; xTransaction > lastTransaction; xTransaction--) {
        transaction = utTransactions + xTransaction - 1;
        undoTransaction(transaction);
    }
    utRedoTransaction = transaction;
    utAssert(utChecksum == 0);
    return firstTransaction - lastTransaction;
}

/*--------------------------------------------------------------------------------------------------
  Redo the transaction.
--------------------------------------------------------------------------------------------------*/
static void redoTransaction(
    utTransaction transaction)
{
    uint8 *command = utCommandBuffer + transaction->position - utBufferStartPosition;
    uint32 length = transaction->length - 5; /* Subract out the transaction-complete command */

    applyCommands(command, length);
}

/*--------------------------------------------------------------------------------------------------
  Redo the last transaction, by scanning forwards through the recent changes buffer, executing
  non-undo commands.
--------------------------------------------------------------------------------------------------*/
uint32 utRedo(
    uint32 numChanges)
{
    utTransaction transaction;
    uint32 xTransaction;
    uint32 firstTransaction, lastTransaction;

    if(utRedoTransaction == NULL) {
        return 0;
    }
    utAssert(utChecksum == 0);
    firstTransaction = utRedoTransaction - utTransactions;
    lastTransaction = utMin(firstTransaction + numChanges, utUsedTransactions);
    for(xTransaction = firstTransaction; xTransaction < lastTransaction; xTransaction++) {
        transaction = utTransactions + xTransaction;
        redoTransaction(transaction);
    }
    if(lastTransaction == utUsedTransactions) {
        utRedoTransaction = NULL;
    } else {
        utRedoTransaction = utTransactions + lastTransaction;
    }
    utAssert(utChecksum == 0);
    return firstTransaction - lastTransaction;
}