pblkf.c

B树算法实现

        if( pblItemAppend( block, savekey, savekeylen, item ))
        {
            return( -1 );
        }
        return( index );
    }

    /*
     * read the previous item on the block
     */
    if( index > 0 )
    {
        if( pblItemGet( block, index - 1, &previtem ))
        {
            return( -1 );
        }
    }

    /*
     * calculate the number of bytes the key of the item has in
     * common with the key of its predecessor
     */
    if(( index > 0 ) && ( previtem.keylen > 0 ))
    {
        item->keycommon = pbl_memcmplen( previtem.key, previtem.keylen,
                                         item->key, item->keylen );
    }
    else
    {
        item->keycommon = 0;
    }

    /*
     * calculate the size the item needs on the block
     */
    itemsize = pblItemSize( block, item );

    /*
     * check if the item fits here, the "+ 2" is for the slot!
     */
    if( PBLINDEXBLOCKNFREE( block ) < itemsize + 2 )
    {
        pbl_errno = PBL_ERROR_NOFIT;
        return( -1 );
    }

    /*
     * read the next item on the block
     */
    if( pblItemGet( block, index, &peeritem ))
    {
        return( -1 );
    }

    /*
     * calculate the number of bytes the key of the next item has in
     * common with the key of the new item
     */
    if( item->keylen > 0 )
    {
        keycommon = pbl_memcmplen( item->key, item->keylen,
                                   peeritem.key, peeritem.keylen );
    }
    else
    {
        keycommon = 0;
    }

    /*
     * if the next item has to change
     */
    if( keycommon != peeritem.keycommon )
    {
        /*
         * set the new keycommon value for the next item
         */
        peeritem.keycommon = keycommon;

        /*
         * save the data of that item
         */
        if( peeritem.datalen <= PBLDATALENGTH )
        {
            memcpy( data, peeritem.data, peeritem.datalen );
            peeritem.data = data;
        }

        /*
         * save the key of the item
         */
        memcpy( savekey, peeritem.key, peeritem.keylen );
        peeritem.key = savekey;

        /*
         * delete the next item
         */
        rc = pblItemDelete( block, index );
        if( rc )
        {
            return( rc );
        }

        /*
         * insert the saved item again
         */
        rc = pblItemInsert( block, &peeritem, index );
        if( rc )
        {
            return( rc );
        }
    }

    /*
     * insert the new item
     */
    rc = pblItemInsert( block, item, index );
    if( rc )
    {
        return( rc );
    }

    return( index );
}

/*
 * BLOCK functions
 *
 * The size of a block in the file is PBLDATASIZE
 *
 * The layout is as follows:
 *
 * OFFSET    NAME    SEMANTICS
 *
 * 0        LEVEL    Level of the block in the tree
 *                   if 0:      the block is a leaf node
 *                   if 254:    the block is a free block, can be reused
 *                   if 255:    the block is a data block, not belonging to
 *                              the tree.
 *                   otherwise: block is an inner node of the tree
 *
 * 1  -  4 NOFFSET   block number of next block of same level, as the root block
 *                   always is the only block of it's level, the block number
 *                   of the last data block is stored here, we use this block
 *                   for appending data if necessary.
 *
 * 5  -  8 POFFEST   file offset of previous block of the same level,
 *                   as the root block always is the only block of it's level,
 *                   the block number of the first free block with level 254
 *                   is stored on the root block
 *
 * 9  - 10 NENTRIES  number of items stored in the data of the block.
 *                   always 0 for data blocks.
 *
 * 11 - 12 FREE      relative offset of first free byte in the data of the block
 *
 * 13 - ( PBLDATASIZE - 1 ) data area of the block used for storing items on
 *                          tree blocks ( level 0 - 253 ) or used for storing
 *                          the data of the items on data blocks ( level 255 ).
 *
 * The root block of the tree is always stored at file offset 0. The first data
 * block at file offset PBLDATASIZE. There are at least two blocks in a file
 * even if there are no items stored at all.
 *
 * For records with a datalen of less or equal to PBLDATALENGTH characters
 * the data is stored on the level 0 index blocks. For records with
 * data longer than PBLDATALENGTH characters the data is stored on data blocks.
 *
 * This is done in order to keep the height of the tree small and to allow
 * data lengths of more than PBLDATASIZE bytes. What we pay for this is that
 * the space in the file used for the data of records stored on data blocks
 * and that are deleted or updated is lost.
 *
 * Blocks read from disk are cached in an LRU list, references to blocks
 * in that list are kept in a hash table in order to speed up access.
 *
 * Blocks changed during a transaction are kept in the writeable block
 * list. If a cached block that is dirty has to be made writeable a copy
 * of the block is created in the writeable list, if the block is not
 * dirty the block is moved from the cached list to the writeable list
 * without creating a copy.
 *
 * During a transaction blocks from the writeable list take precedence
 * over the copy of the same block from the cached list.
 *
 * During a rollback all blocks from the writeable list are discarded,
 * thus reverting the file to the state before the beginning of the 
 * transaction.
 *
 * During a commit all blocks are moved from the writeable list to the
 * cached list.
 */

/*
 * The following two procedures are the only ones that 'know' the layout
 * of a the data of a block in the file
 */
static void pblDataToBlock( PBLBLOCK_t * block )
{
    block->data[ 0 ] = block->level;

    pbl_LongToBuf ( &( block->data[ 1 ]),  block->nblock );
    pbl_LongToBuf ( &( block->data[ 5 ]),  block->pblock );
    pbl_ShortToBuf( &( block->data[ 9 ]),  block->nentries );
    pbl_ShortToBuf( &( block->data[ 11 ]), block->free );
}

static void pblBlockToData( PBLBLOCK_t * block )
{
    block->level = block->data[ 0 ];

    block->nblock   = pbl_BufToLong ( &( block->data[ 1 ]));
    block->pblock   = pbl_BufToLong ( &( block->data[ 5 ]));
    block->nentries = pbl_BufToShort( &( block->data[ 9 ]));
    block->free     = pbl_BufToShort( &( block->data[ 11 ]));
}

/*
 * release all memory blocks of a file
 */
static void pblBlocksRelease( int bf )
{
    PBLBLOCK_t * block;

    /*
     * all blocks that were belonging to the file are marked unused
     */
    for( block = blockListHead; block; block = block->next )
    {
        if( block->bf == bf )
        {
            pblBlockHashRemove( block->blockno, block->bf );
            pblBlockKeysRelease( block );
            block->bf         = -1;
            block->filesettag = NULL;
        }
    }
    return;
}

/*
 * release the expanded keys buffer of a block
 */
static void pblBlockKeysRelease( PBLBLOCK_t * block )
{
    PBLBLOCKLINK_t * blink;

    if( block->expandedkeys )
    {
        blink = ( PBLBLOCKLINK_t * )block;

        PBL_LIST_UNLINK( linkListHead, linkListTail, blink, next, prev );
        pblnlinks--;

        PBL_FREE( block->expandedkeys );

        block->expandedkeys = 0;
    }
}

/*
 * expand all keys of a block
 */
static int pblBlockKeysExpand( PBLBLOCK_t * block )
{
    unsigned int      i;
    PBLITEM_t         item;
    unsigned char   * expandedkeys = 0;
    PBLBLOCKLINK_t  * blink;

    if( block->expandedkeys )
    {
        blink = ( PBLBLOCKLINK_t * )block;

        /*
         * make the block link the first in the LRU chain
         */
        if( blink != linkListHead )
        {
            PBL_LIST_UNLINK( linkListHead, linkListTail, blink, next, prev );
            PBL_LIST_PUSH( linkListHead, linkListTail, blink, next, prev );
        }
        return( 0 );
    }

    /*
     * get space for the expanded keys of the block
     */
    if( block->nentries > 0 )
    {
        i = block->nentries * PBLKEYLENGTH;
    }
    else
    {
        i = 1;
    }

    expandedkeys = pbl_malloc( "pblBlockKeysExpand expandedkeys", i );
    if( !expandedkeys )
    {
        return( -1 );
    }

    /*
     * run through all items of the block
     */
    for( i = 0; i < block->nentries; i++ )
    {
        pblItemGet( block, i, &item );

        if( item.keylen < 1 )
        {
            continue;
        }

        if( item.keycommon < 1 || i < 1 )
        {
            /*
             * this item does not have bytes in common with the predecessor
             */
            pbl_memlcpy( expandedkeys + i * PBLKEYLENGTH,
                         PBLKEYLENGTH, item.key, item.keylen );
        }
        else
        {
            /*
             * copy the bytes the key has in common with
             * the key of the predecessor
             */
            pbl_memlcpy( expandedkeys + i * PBLKEYLENGTH, PBLKEYLENGTH,
                         expandedkeys + ( i - 1 ) * PBLKEYLENGTH,
                         item.keycommon );

            if( item.keylen - item.keycommon > 0 )
            {
                /*
                 * copy the bytes that are unique for the key
                 */
                pbl_memlcpy( expandedkeys + i * PBLKEYLENGTH + item.keycommon,
                             PBLKEYLENGTH, item.key,
                             item.keylen - item.keycommon );
            }
        }
    }

    block->expandedkeys = expandedkeys;

    /*
     * release some expanded key structures if there are too many
     */
    while( pblnlinks > 8 + ( pblexpandedperfile * pblnfiles ))
    {
        pblBlockKeysRelease( ( PBLBLOCK_t * )linkListTail );
    }

    /*
     * link the block to the list of blocks that have expanded keys
     */
    blink = ( PBLBLOCKLINK_t * )block;
    PBL_LIST_PUSH( linkListHead, linkListTail, blink, next, prev );
    pblnlinks++;

    return( 0 );
}

static int pblBlockWrite( PBLBLOCK_t * block )
{
    long rc;

    /*
     * prepare the block data for writing
     */
    pblDataToBlock( block );

    /*
     * write the block to the big file
     */
    rc = pbf_blockwrite( block->bf, block->blockno, block->data );
    if( rc < 0 )
    {
        block->bf         = -1;
        block->filesettag = NULL;
        pbl_errno = PBL_ERROR_WRITE;
        return( -1 );
    }

    return( 0 );
}

static int pblBlockFlush( int bf, int release )
{
    PBLBLOCK_t * block;
    PBLBLOCK_t * tmp;

    for( tmp = blockListHead; tmp; )
    {
        /*
         * move through the list of blocks before handling this one
         */
        block = tmp;
        tmp = tmp->next;

        if( block->bf != bf )
        {
            continue;
        }

        /*
         * if a file set tag is set for the block we flush
         * all blocks having the same tag set
         */
        if( block->dirty && block->filesettag )
        {
            PBLBLOCK_t * b;

            /*
             * run through all blocks on all files in the set and write them
             */
            for( b = blockListHead; b; b = b->next )
            {
                if( b->dirty && b->bf >= 0
                 && ( block->filesettag == b->filesettag ))
                {
                    if( pblBlockWrite( b ))
                    {
                        pblBlocksRelease( b->bf );
                        break;
                    }
                    else
                    {
                        b->dirty = 0;
                    }
                }
            }
        }
                
        if( block->dirty )
        {         
            if( pblBlockWrite( block ))
            {
                /*
                 * this write always is a rewrite of an existing block
                 * therefore a write error is a strange condition,
                 * we unlink all blocks from the file
                 * most likely the file is inconsistent after that !!!!
                 */
                pblBlocksRelease( bf );
                return( -1 );
            }
            else
            {
                block->dirty = 0;
            }
        }

        if( release )
        {
            pblBlockHashRemove( block->blockno, block->bf );
            pblBlockKeysRelease( block );
            block->bf         = -1;
            block->filesettag = NULL;

            /*
             * put the block to the end of the LRU list
             */
            if( block != blockListTail )
            {
                PBL_LIST_UNLINK( blockListHead, blockListTail,
                                 block, next, prev );
                PBL_LIST_APPEND( blockListHead, blockListTail,
                                 block, next, prev );
            }
        }
    }
    return( 0 );
}

static PBLBLOCK_t * pblBlockGetVictim( PBLKFILE_t * file )
{
    int          rc;
    PBLBLOCK_t * block;

    /*
     * if we have not exceeded the number of blocks we can have at most
     * or of the last block in the LRU chain is dirty and we are updating
     */
    if(( pblnblocks < 8 + ( pblblocksperfile * pblnfiles ) )
     ||( blockListTail && blockListTail->dirty
      && blockListTail->bf != -1 && file->writeableListHead ))
    {
        block = pbl_malloc0( "pblBlockGetVictim BLOCK", sizeof( PBLBLOCK_t ));
        if( !block )
        {
            return( ( PBLBLOCK_t * ) 0 );
        }
        pblnblocks++;
        PBL_LIST_PUSH( blockListHead, blockListTail, block, next, prev );
    }
    else
    {
        /*
         * we reuse the last block in the LRU chain
         */
        if( blockListTail )
        {
            if( blockListTail->bf != -1 )
            {
                /*
                 * flush the block to disk if it is dirty
                 */
                if( blockListTail->dirty )
                {
                    rc = pblBlockFlush( blockListTail->bf, 0 );