/*
 pblkf.c - key file library implementation

 Copyright (C) 2002      Peter Graf

   This file is part of PBL - The Program Base Library.
   PBL is free software.

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.

    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

   For more information on the Program Base Library or Peter Graf,
   please see: http://www.mission-base.com/.

    $Log: pblkf.c,v $
    Revision 1.5  2003/04/20 00:26:49  peter
    removed a compiler warning with gcc version 3.3

    Revision 1.4  2003/02/19 22:26:49  peter
    made sure #defined values can be set by compiler switches

    Revision 1.3  2002/11/01 13:56:24  peter
    Truncation of the memory block list is now called
    at every file close.

    Revision 1.2  2002/11/01 13:27:30  peter
    The block reference hash table is deleted when the
    last block is deleted from the table, i.e. when the
    last file is closed

    Revision 1.1  2002/09/12 20:47:07  peter
    Initial revision


*/

/*
 * make sure "strings <exe> | grep Id | sort -u" shows the source file versions
 */
static char * rcsid = "$Id: pblkf.c,v 1.5 2003/04/20 00:26:49 peter Exp $";
static int    rcsid_fkt() { return( rcsid ? 0 : rcsid_fkt() ); }

#ifndef _WIN32

#include <unistd.h>

#endif

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/stat.h>

#include <fcntl.h>
#include <memory.h>
#include <string.h>


#include "pbl.h"                 /* program base library                      */

/******************************************************************************/
/* #defines                                                                   */
/******************************************************************************/

#ifndef PBL_KF_TEST_DEFINES

#ifndef PBLDATASIZE             /* value might be set by compiler switch      */
#define PBLDATASIZE      4096   /* size of entire block data                  */
#endif

#ifndef PBLBLOCKSPERFILE        /* value might be set by compiler switch      */
#define PBLBLOCKSPERFILE   64   /* default number of cache blocks per file    */
#endif

#ifndef PBLNEXPANDED            /* value might be set by compiler switch      */
#define PBLNEXPANDED       32   /* default number of expanded blocks          */
#endif

#ifndef PBLFILEBLOCKS           /* value might be set by compiler switch      */
#define PBLFILEBLOCKS  0xffff   /* number of blocks per big file  on disk     */
#endif

#ifndef PBL_NFILES              /* value might be set by compiler switch      */
#define PBL_NFILES        256   /* maximum number of open key files           */
#endif

#else /* PBL_TEST_DEFINES */

#define PBLDATASIZE        64   /* size of entire block data                  */
#define PBLBLOCKSPERFILE   10   /* default number of blocks in our cache      */
#define PBLNEXPANDED        5   /* default number of expanded blocks          */
#define PBLFILEBLOCKS       3   /* number of blocks per file                  */

#undef  PBLKEYLENGTH
#define PBLKEYLENGTH        8   /* maximum length of a key                  */

#undef  PBLDATALENGTH
#define PBLDATALENGTH      16   /* maximum length of data that is stored      */
                                /* with an item on the level 0 block          */
                                /* data that is longer is stored on data      */
                                /* blocks                                     */
#ifndef PBL_NFILES
#define PBL_NFILES        256   /* maximum number of open key files           */
#endif

#endif /* PBL_TEST_DEFINES */

#define PBLHEADERSIZE      13   /* offset of first free byte on a new block   */

#define PBLMAGIC          "1.00 Peter's B Tree"
                                /* during creation this string is written     */
                                /* to the files                               */

/******************************************************************************/
/* #macros                                                                    */
/******************************************************************************/

/*
 * macro to determine free space on an index block
 */
#define PBLINDEXBLOCKNFREE( B ) ((PBLDATASIZE - B->free) - (2 * B->nentries))

/*
 * macro to determine free space on a data block
 */
#define PBLDATABLOCKNFREE( B ) ( PBLDATASIZE - B->free )

/*
 * macros for getting pointers or buffer offsets from the index
 */
#define PBLINDEXTOPTR( B, I ) ( B->data\
                              + ( PBLDATASIZE - 2 * ( 1 + ( I ) )))

#define PBLINDEXTOBUFOFF( B, I ) pbl_BufToShort( PBLINDEXTOPTR( B, I ))

/******************************************************************************/
/* typedefs                                                                   */
/******************************************************************************/

/*
 * PBL key file descriptor
 */
typedef struct PBLKFILE_s
{
    char * magic;                /* magic string                              */
    int    bf;                   /* file handle from bigfile handling         */
    int    update;               /* flag: file open for update                */

    long   blockno;              /* of block current record is on             */
    int    index;                /* of current record on this block           */

    long   saveblockno;          /* block number of saved position            */
    int    saveindex;            /* item index of saved position              */

    void * filesettag;           /* file set tag attached to file             */

    int    transactions;         /* number of transactions active for file    */
    int    rollback;             /* next commit should lead to a rollback     */

    void * writeableListHead;    /* head and tail of writeable blocks         */
    void * writeableListTail;

                                 /* a user defined key compare function       */
    int (*keycompare)( void * left, size_t llen, void * right, size_t rlen );

} PBLKFILE_t;

/*
 * items are the things we store in the data array of the index blocks
 */
typedef struct PBLITEM_s
{
    unsigned int    level;       /* level where item is inserted              */

    int             keycommon;   /* number of initial bytes this item has in  */
                                 /* common with its predecessor on tbe block  */

    int             keylen;      /* length of the key                         */
    unsigned char * key;         /* pointer to the key                        */

    long            datalen;     /* length of the data                        */
    unsigned char * data;        /* the data of the item                      */

    long            datablock;   /* number of block the data is on            */
    long            dataoffset;  /* offset of data on block                   */

    struct PBLITEM_s * next;     /* we save items in a list                   */
    struct PBLITEM_s * prev;     /* during an insert                          */

} PBLITEM_t;

/*
 * memory blocks that have expanded keys are stored in an extra list
 */
typedef struct PBLBLOCKLINK_s
{
    struct PBLBLOCKLINK_s * next;
    struct PBLBLOCKLINK_s * prev;

} PBLBLOCKLINK_t;

/*
 * a file block in memory has this format
 */
typedef struct PBLBLOCK_s
{
    PBLBLOCKLINK_t   blocklink;         /* for linking blocks in a list       */

    unsigned int   level;               /* of block                           */

    long           blockno;             /* block number in file               */
    int            bf;                  /* big file handle block belongs to   */
    void         * filesettag;          /* file set tag attached              */

    unsigned char  data[ PBLDATASIZE ]; /* the real data                      */

    long           nblock;              /* next block of same level           */
    long           pblock;              /* previous block of same level       */
    unsigned int   nentries;            /* number of entries                  */
    int            free;                /* offset of first free byte          */

    long           parentblock;         /* number of parent block             */
    int            parentindex;         /* index on parentblock               */

    int            writeable;           /* block can be written to            */
    int            dirty;               /* has the block been written to ?    */

    unsigned char *expandedkeys;        /* pointer to expanded keys of        */
                                        /* the block                          */

    struct PBLBLOCK_s * next;           /* we keep the blocks we have         */
    struct PBLBLOCK_s * prev;           /* in memory in a LRU chain           */

} PBLBLOCK_t;

/*
 * all file handles of all open filesystem files of
 * all bigfiles are stored in a list
 */
typedef struct PBLBIGFILEHANDLE_s
{
    int    bf;                     /* bigfile handle of file                  */
    int    n;                      /* bigfile index of file                   */

    int    fh;                     /* filesystem handle                       */
    int    mode;                   /* open mode                               */

    struct PBLBIGFILEHANDLE_s * next;
    struct PBLBIGFILEHANDLE_s * prev;

} PBLBIGFILEHANDLE_t;

/*
 * a bigfile in memory
 */
typedef struct PBLBIGFILE_s
{
    char * name;                    /* name of first filesystem file          */

    int    mode;                    /* open mode                              */
    long   blocksperfile;           /* blocks per filesystem file             */
    long   blocksize;               /* size of one block                      */
    long   nextblockno;             /* block number of next free block        */

} PBLBIGFILE_t;

/*
 * references to blocks are stored in a hashtable
 */
typedef struct PBLBLOCKREF_s
{
    long           blockno;             /* block number in file               */
    long           bf;                  /* big file handle block belongs to   */

    PBLBLOCK_t   * block;               /* address of block                   */

} PBLBLOCKREF_t;

typedef struct PBLBLOCKHASHKEY_s
{
    long           blockno;             /* block number in file               */
    long           bf;                  /* big file handle block belongs to   */

} PBLBLOCKHASHKEY_t;

/******************************************************************************/
/* globals                                                                    */
/******************************************************************************/

long pblnreads  = 0;
long pblnwrites = 0;

static char * magic = PBLMAGIC;

/*
 * pool of open bigfiles
 */
static PBLBIGFILE_t pbf_pool[ PBL_NFILES ];

/*
 * headers for lists
 */
static PBLBIGFILEHANDLE_t     * pbf_ft_head;
static PBLBIGFILEHANDLE_t     * pbf_ft_tail;
static PBLBLOCKLINK_t         * linkListHead;
static PBLBLOCKLINK_t         * linkListTail;
static PBLBLOCK_t             * blockListHead;
static PBLBLOCK_t             * blockListTail;
static PBLITEM_t              * itemListHead;
static PBLITEM_t              * itemListTail;

/*
 * counters
 */
static int            pblnblocks;
static int            pblnlinks;
static int            pblblocksperfile = PBLBLOCKSPERFILE;
static int            pblexpandedperfile = PBLNEXPANDED;
static long           pblnfiles;

/*
 * block reference hash table
 */
static pblHashTable_t * pblblockhash;

/******************************************************************************/
/* declarations                                                               */
/******************************************************************************/
static void pblBlockKeysRelease( PBLBLOCK_t * block );
static int pblBlockKeysExpand( PBLBLOCK_t * block );

/******************************************************************************/
/* functions                                                                  */
/******************************************************************************/
/*
 * verify consistency of parameters
 */
static int pblParamsCheck(
unsigned char * key,
unsigned int    keylen,
unsigned char * data,
long            datalen
)
{
    if(( keylen < 1 ) || ( keylen > 255 ))
    {
        pbl_errno = PBL_ERROR_PARAM_KEYLEN;
        return( -1 );
    }

    if( datalen < 0 )
    {
        pbl_errno = PBL_ERROR_PARAM_DATALEN;
        return( -1 );
    }

    if( !key )
    {
        pbl_errno = PBL_ERROR_PARAM_KEY;
        return( -1 );
    }

    if( datalen && ( !data ))
    {
        pbl_errno = PBL_ERROR_PARAM_DATA;
        return( -1 );
    }

    return( 0 );
}

/*
 * functions on the block reference hash table
 */
/*
------------------------------------------------------------------------------
  FUNCTION:     pblBlockHashInsert

  DESCRIPTION:  inserts a new block reference into the hash table

  RETURNS:      int rc == 0: a new reference was inserted
                int rc == 1: the reference was already there, it was udpated
                int rc <  0: some error occured, see pbl_errno
                 PBL_ERROR_OUT_OF_MEMORY:        malloc failed, out of memory
------------------------------------------------------------------------------
*/
static int pblBlockHashInsert( long blockno, long bf, PBLBLOCK_t * block )
{
    PBLBLOCKHASHKEY_t   key;
    PBLBLOCKREF_t     * ref;
    int                 rc;

    memset( &key, 0, sizeof( key ));
    key.blockno = blockno;
    key.bf      = bf;

    if( !pblblockhash )
    {
        pblblockhash = pblHtCreate();
        if( !pblblockhash )
        {
            return( -1 );
        }
    }

    /*
     * see whether we have the reference already
     */
    ref = pblHtLookup( pblblockhash, &key, sizeof( key ));
    if( ref )
    {
        /*
         * the reference is already there, update the block pointer
         */
        ref->block = block;
        return( 1 );
    }

    if( pbl_errno == PBL_ERROR_NOT_FOUND )
    {
        pbl_errno = 0;
    }

    /*
     * allocate memory for new reference
     */
    ref = pbl_malloc0( "pblBlockHashInsert BLOCKREF", sizeof( PBLBLOCKREF_t ));
    if( !ref )
    {
        return( -1 );
    }

    /*
     * insert to the hash table
     */
    rc = pblHtInsert( pblblockhash, &key, sizeof( key ), ref );
    if( !rc )
    {
        ref->block   = block;
        ref->blockno = blockno;
        ref->bf      = bf;
    }
    else
    {
        PBL_FREE( ref );
        return( -1 );
    }

    return( 0 );
}

/*
------------------------------------------------------------------------------
  FUNCTION:     pblBlockHashRemove

  DESCRIPTION:  Remove a block reference from the hash table

  RETURNS:      int rc == 0: call went ok;
                otherwise:   block not found in hash table
------------------------------------------------------------------------------
*/
static int pblBlockHashRemove( long blockno, long bf )
{
    PBLBLOCKHASHKEY_t   key;
    PBLBLOCKREF_t     * ref;
    int                 rc;

    memset( &key, 0, sizeof( key ));
    key.blockno = blockno;
    key.bf      = bf;

    /*
     * if there is no hash table yet, the reference is not found
     */
    if( !pblblockhash )
    {
        return( 1 );
    }

    /*
     * see whether we have the reference
     */
    ref = pblHtLookup( pblblockhash, &key, sizeof( key ));
    if( !ref )
    {
        if( pbl_errno == PBL_ERROR_NOT_FOUND )
        {
            pbl_errno = 0;
        }

        return( 1 );
    }

    /*
     * remove the reference from the hash table
     */
    rc = pblHtRemove( pblblockhash, &key, sizeof( key ));
    if( rc )
    {
        return( 1 );
    }

    PBL_FREE( ref );

    /*
     * attempt to remove the hashtable
     */
    rc = pblHtDelete( pblblockhash );
    if( !rc )
    {
        /*
         * the hash table was empty and is deleted now
         */
        pblblockhash = 0;
    }
    else
    {
        /*
         * the hash table was not deleted because it is not
         * empty, clear the error
         */
        pbl_errno = 0;
    }

    return( 0 );
}

/*
------------------------------------------------------------------------------
  FUNCTION:     pblBlockHashFind

  DESCRIPTION:  Find a block reference in the hash table

  RETURNS:      PBLBLOCK_t * retptr == 0: block not found
                otherwise:                pointer to block in memory
------------------------------------------------------------------------------
*/
static PBLBLOCK_t * pblBlockHashFind( long blockno, long bf )
{
    PBLBLOCKHASHKEY_t   key;
    PBLBLOCKREF_t     * ref;

    memset( &key, 0, sizeof( key ));
    key.blockno = blockno;
    key.bf      = bf;

    /*
     * if there is no hash table yet, the reference is not found
     */
    if( !pblblockhash )
    {
        return( 0 );
    }

    /*
     * see whether we have the reference
     */
    ref = pblHtLookup( pblblockhash, &key, sizeof( key ));
    if( !ref )
    {
        if( pbl_errno == PBL_ERROR_NOT_FOUND )
        {
            pbl_errno = 0;
        }

        return( 0 );
    }

    return( ref->block );
}

/*
 * BIGFILE functions
 */
/*
 * find a filehandle of a filesystem file
 */
static PBLBIGFILEHANDLE_t * pbf_fh_find( int bf, int n )
{
    PBLBIGFILEHANDLE_t * entry;

    /*
     * file system handles are kept in an LRU list
     */
    for( entry = pbf_ft_head; entry; entry = entry->next )
    {
        if( bf == entry->bf && n == entry->n )
        {
            if( entry != pbf_ft_head )
            {
                PBL_LIST_UNLINK( pbf_ft_head, pbf_ft_tail, entry, next, prev );
                PBL_LIST_PUSH( pbf_ft_head, pbf_ft_tail, entry, next, prev );
            }
            return( entry );
        }
    }

    return( 0 );
}

/*
 * close files with the filesystem
 */
static void pbf_fh_close( int bf, int n )
{
    PBLBIGFILEHANDLE_t * entry;
    PBLBIGFILEHANDLE_t * tmp;

    if( n < 0 )
    {
        /*
         * close all file handles of this big file
         */
        for( entry = pbf_ft_head; entry; )
        {
            tmp = entry;
            entry = entry->next;

            if( bf == tmp->bf )
            {
                close( tmp->fh );
                PBL_LIST_UNLINK( pbf_ft_head, pbf_ft_tail, tmp, next, prev );
                PBL_FREE( tmp );
            }
        }

        return;
    }

    /*
     * close a particular file handle
     */
    entry = pbf_fh_find( bf, n );
    if( !entry )
    {
        return;
    }

    close( entry->fh );

    PBL_LIST_UNLINK( pbf_ft_head, pbf_ft_tail, entry, next, prev );
    PBL_FREE( entry );
}

/*
 * create the bigfile path of a file
 * bigfiles are stored in multiple filesystem files
 * if the first one is called "file.ext",
 * the others have names like "file_0002.ext" etc.
 */
static char * pbf_fh_path( char * name, long n )
{
    char * path;
    char * dotptr;

    if( n < 1 )
    {
        /*
         * the name of the first file does not change
         */
        path = strdup( name );
        if( !path )
        {
            pbl_errno = PBL_ERROR_OUT_OF_MEMORY;
            return( 0 );
        }
        return( path );
    }

    path = pbl_malloc( "pbf_fh_path path", 6 + strlen( name ));
    if( !path )
    {
        return( 0 );
    }

    /*
     * see if there is a ".ext" after the last slash
     */
    dotptr = strrchr( name, '.' );
    if( dotptr )
    {
        if( strchr( dotptr, '/' ) || strchr( dotptr, '\\' ))
        {
            dotptr = 0;
        }
    }

    /*
     * build the filename, start counting at one
     */
    n++;
    if( dotptr )
    {
        memcpy( path, name, dotptr - name );
        snprintf( path + ( dotptr - name ), 6, "_%04lx", 0xffff & n );
        strcat( path, dotptr );
    }
    else
    {
        strcpy( path, name );
        snprintf( path + strlen( path ), 6, "_%04lx", 0xffff & n );
    }

    return( path );
}

/*
 * open a file with the file system
 */
static int pbf_fh_open( char * name, int mode, int bf, int n )
{
    PBLBIGFILEHANDLE_t * entry;
    int                  fh = -1;
    int                  i;
    char               * path;

    /*
     * look in LRU list of open filesystem files
     */
    entry = pbf_fh_find( bf, n );
    if( entry )
    {
        if( entry->mode != mode )
        {
            /*
             * the file was found, but the open mode is wrong, close the file
             */
            pbf_fh_close( bf, n );
            entry = 0;
        }
        else
        {
            /*
             * the file is open in the right mode, use the file handle
             */
            return( entry->fh );
        }
    }

    /*
     * open the file
     */
    path = pbf_fh_path( name, n );
    if( !path )
    {
        return( -1 );
    }

    for( i = 0; i < 3; i++ )
    {
        fh = open( path, mode, S_IREAD | S_IWRITE );
        if( -1 == fh && pbf_ft_tail )
        {
            /*
             * maybe the process or the system ran out of filehandles
             * close one file and try again
             */
            pbf_fh_close( pbf_ft_tail->bf, pbf_ft_tail->n );
            continue;
        }

        break;
    }
    PBL_FREE( path );

    if( -1 == fh )
    {
        pbl_errno = PBL_ERROR_OPEN;
        return( -1 );
    }

    /*
     * create and link the file handle list entry
     */
    entry = pbl_malloc0( "pbf_fh_open *entry", sizeof( *entry ));
    if( !entry )
    {
        close( fh );
        return( 0 );
    }

    entry->fh   = fh;
    entry->mode = mode;
    entry->bf   = bf;
    entry->n    = n;

    PBL_LIST_PUSH( pbf_ft_head, pbf_ft_tail, entry, next, prev );

    return( entry->fh );
}

/*
 * close a bigfile
 */
static int pbf_close( int bf )
{
    if( bf < 0 || bf >= PBL_NFILES )
    {
        return( -1 );
    }

    /*
     * close all filesystem files
     */
    pbf_fh_close( bf, -1 );

    PBL_FREE( pbf_pool[ bf ].name );
    pbf_pool[ bf ].name = 0;

    return( 0 );
}

/*
 * open a bigfile
 */
static int pbf_open(
char * name,
int    update,
long   blocksperfile,
long   blocksize
)
{
    int        fh = -1;
    int        mode;
    int        i;
    char     * path;

    path = pbf_fh_path( name, 0 );
    if( !path )
    {
        return( -1 );
    }

    if( update )
    {
        mode = O_CREAT | O_BINARY | O_RDWR;
    }
    else
    {
        mode = O_BINARY | O_RDONLY;
    }

    /*
     * find a slot in the big file pool that is free
     */
    for( i = 0; i < PBL_NFILES; i++ )
    {
        if( pbf_pool[ i ].name )
        {
            continue;
        }

        /*
         * reserve the slot
         */
        pbf_pool[ i ].mode = mode;
        pbf_pool[ i ].nextblockno = -1;
        pbf_pool[ i ].blocksperfile = blocksperfile;
        pbf_pool[ i ].blocksize = blocksize;

        /*
         * open the first filesystem file
         */
        fh = pbf_fh_open( path, mode, i, 0 );
        if( -1 == fh )
        {
            PBL_FREE( path );
            pbl_errno = PBL_ERROR_OPEN;
            return( -1 );
        }

        pbf_pool[ i ].name = path;
        return( i );
    }

    PBL_FREE( path );
    pbl_errno = PBL_ERROR_OPEN;
    return( -1 );
}

/*
 * file io for a bigfile
 */
static int pbf_blockio(
int             bf,
int             blockwrite,
long            blockno,
unsigned char * buffer
)
{
    long       rc = -1;
    long       n;
    int        fh;
    int        i;
    int        j;
    long       offset;

    if( bf < 0 || bf >= PBL_NFILES || !pbf_pool[ bf ].name )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    /*
     * make sure the n'th filesystem file is open
     */
    n = blockno / pbf_pool[ bf ].blocksperfile;

    fh = pbf_fh_open( pbf_pool[ bf ].name, pbf_pool[ bf ].mode, bf, n );
    if( -1 == fh )
    {
        pbl_errno = PBL_ERROR_READ;
        return( -1 );
    }

    blockno %= pbf_pool[ bf ].blocksperfile;
    offset = blockno * pbf_pool[ bf ].blocksize;

    if( blockwrite )
    {
        /*
         * try the write more than once if needed
         */
        for( i = 0; i < 3; i++ )
        {
            /*
             * try the seek more than once if needed
             */
            for( j = 0; j < 3; j++ )
            {
                rc = lseek( fh, offset, SEEK_SET );
                if( offset != rc )
                {
                    continue;
                }
                break;
            }
            if( offset != rc )
            {
                pbl_errno = PBL_ERROR_SEEK;
                return( -1 );
            }

            rc = write( fh, buffer, (unsigned int) pbf_pool[ bf ].blocksize );
            if( rc != pbf_pool[ bf ].blocksize )
            {
                if( errno == EINTR )
                {
                    continue;
                }
                pbl_errno = PBL_ERROR_WRITE;
                return( -1 );
            }
            break;
        }
        if( i >= 3 )
        {
            pbl_errno = PBL_ERROR_WRITE;
            return( -1 );
        }

        pblnwrites++;
    }
    else
    {
        /*
         * try the read more than once if needed
         */
        for( i = 0; i < 3; i++ )
        {
            /*
             * try the seek more than once if needed
             */
            for( j = 0; j < 3; j++ )
            {
                rc = lseek( fh, offset, SEEK_SET );
                if( offset != rc )
                {
                    continue;
                }
                break;
            }
            if( offset != rc )
            {
                pbl_errno = PBL_ERROR_SEEK;
                return( -1 );
            }

            rc = read( fh, buffer, (unsigned int) pbf_pool[ bf ].blocksize );
            if( rc < 0 )
            {
                if( errno == EINTR )
                {
                    continue;
                }
                pbl_errno = PBL_ERROR_READ;
                return( -1 );
            }
            pblnreads++;

            if( rc != pbf_pool[ bf ].blocksize )
            {
                if( errno == EINTR )
                {
                    continue;
                }
                pbl_errno = PBL_ERROR_BAD_FILE;
                return( -1 );
            }
            break;
        }
        if( i >= 3 )
        {
            pbl_errno = PBL_ERROR_READ;
            return( -1 );
        }
    }

    return( 0 );
}

/*
 * read a block from a bigfile
 */
static int pbf_blockread( int bf, long blockno, unsigned char * buffer )
{
    int        rc;

    rc = pbf_blockio( bf, 0, blockno, buffer );

    return( rc );
}

/*
 * write a block to a bigfile
 */
static int pbf_blockwrite( int bf, long blockno, unsigned char * buffer )
{
    int        rc;

    rc = pbf_blockio( bf, 1, blockno, buffer );

    return( rc );
}

/*
 * append a block to a bigfile
 */
static long pbf_blockappend( int bf, unsigned char * buffer )
{
    int           fh;
    long          offset;
    int           i;
    int           j;
    long          rc = -1;
    unsigned char c;

    if( bf < 0 || bf >= PBL_NFILES || !pbf_pool[ bf ].name )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    if( pbf_pool[ bf ].nextblockno == 0x3fff )
    {
        rc = -1;
    }

    /*
     * if we don't know the next free block of the file yet
     */
    if( pbf_pool[ bf ].nextblockno < 0 )
    {
        /*
         * find the next free block of a bigfile
         */
        for( i = 0; 1; i++ )
        {
            /*
             * create and open next filesytem file used for the bigfile
             */
            fh = pbf_fh_open( pbf_pool[ bf ].name, pbf_pool[ bf ].mode, bf, i );
            if( -1 == fh )
            {
                pbl_errno = PBL_ERROR_WRITE;
                return( -1 );
            }

            /*
             * reopen the filesystem file read only
             */
            fh = pbf_fh_open( pbf_pool[ bf ].name, O_BINARY | O_RDONLY, bf, i );
            if( -1 == fh )
            {
                pbl_errno = PBL_ERROR_WRITE;
                return( -1 );
            }

            /*
             * seek to the end of the file
             */
            offset = pbf_pool[ bf ].blocksperfile * pbf_pool[ bf ].blocksize -1;

            for( j = 0; j < 3; j++ )
            {
                rc = lseek( fh, offset, SEEK_SET );
                if( offset != rc )
                {
                    if( errno == EINTR )
                    {
                        continue;
                    }
                    rc = -1;
                    break;
                }

                /*
                 * check whether the block exist, by reading its last byte
                 */
                rc = read( fh, &c, 1 );
                if( rc < 1 )
                {
                    if( errno == EINTR )
                    {
                        continue;
                    }
                }
                break;
            }
                
            if( rc == 1 )
            {
                /*
                 * we can read the last byte of the block, it is not free
                 */
                continue;
            }

            /*
             * we found the last filesystem file used for the bigfile
             * seek to the end of the file
             */
            for( j = 0; j < 3; j++ )
            {
                rc = lseek( fh, 0, SEEK_END );
                if( rc < 0 )
                {
                    if( errno == EINTR )
                    {
                        continue;
                    }

                    pbl_errno = PBL_ERROR_WRITE;
                    return( -1 );
                }
                break;
            }

            if( rc % pbf_pool[ bf ].blocksize )
            {
                pbl_errno = PBL_ERROR_BAD_FILE;
                return( -1 );
            }

            pbf_pool[ bf ].nextblockno = rc / pbf_pool[ bf ].blocksize;
            break;
        }
    }

    /*
     * append the block to the bigfile
     */
    rc = pbf_blockio( bf, 1, pbf_pool[ bf ].nextblockno, buffer );
    if( rc )
    {
        return( rc );
    }

    return( pbf_pool[ bf ].nextblockno++ );
}

/*
 * ITEM functions
 *
 * Layout of an item in the data of a block with a level greater than 0
 *
 * LENGTH    NAME        SEMANTICS
 *
 * 1        KEYLEN        length of the key of the item
 * 1        KEYCOMMON     number of bytes the key has in common with
 *                        the key of the predecessor of the item on the block
 * VARLONG  DATABLOCK     block number of block this item points to
 * KEYLEN   KEY           the key itself, only the last KEYLEN - KEYCOMMON
 *                        bytes are stored
 *
 * Layout of an item in the data of a block with a level 0 and
 * with the datalen of the item smaller than or equal to PBLDATALENGTH
 *
 * LENGTH    NAME        SEMANTICS
 *
 * 1         KEYLEN       length of the key of the item
 * 1         KEYCOMMON    number of bytes the key has in common with
 *                        the key of the predecessor of the item on the block
 * VARLONG   DATALEN      length of the data stored on the block
 * KEYLEN    KEY          the key itself, only the last KEYLEN - KEYCOMMON
 *                        bytes are stored
 * DATALEN   DATA         the data is stored on the block
 *
 * Layout of an item in the data of a block with a level 0 and
 * with the datalen of the item greater than PBLDATALENGTH
 *
 * LENGTH    NAME        SEMANTICS
 *
 * 1         KEYLEN       length of the key of the item
 * 1         KEYCOMMON    number of bytes the key has in common with
 *                        the key of the predecessor of the item on the block
 * VARLONG   DATALEN      length of the data stored on the datablock
 * KEYLEN    KEY          the key itself, only the last KEYLEN - KEYCOMMON
 *                        bytes are stored
 * VARLONG   DATABLOCK    block number of block data is stored on
 * VARLONG   DATAOFFSET   offset of the data on that block
 *
 * The long values stored for an item, DATALEN, DATABLOCK and DATAOFFSET
 * are stored as variable length buffers, i.e. the number of bytes
 * used in the file for the values depends on their numerical value.
 * See the VarBuf* functions for details.
 *
 * The smallest item of an inner block always has KEYLEN 0, which makes
 * it's key logically smaller than any other possible key of an item.
 *
 * The items stored on a block start at address ( block->data + PBLHEADERSIZE ).
 * The items are always stored immediately one after the other starting at
 * at this address, we leave no space in between.
 *
 * As the keys of the items and therefore the items themselves can have
 * different lengths, we store the relative offsets of the items in the
 * data of a block also in the data of the block. Those relative offsets
 * are stored as two byte unsigned shorts at the end of the data of the block.
 * The relative offsets are called slots in the code below.
 *
 * The slots in the slot array are kept in order.
 *
 * For every block the short block->free gives the relative offset of the first
 * free byte of the block, immediately after the end of the last item stored.
 *
 * Before we append an item we check if there is enough space for the item
 * and its slot between the end of the last item stored and the beginning
 * of the slot array of the items stored. If not, PBL_ERROR_NOFIT is given
 * to the pblinsert procedure which then has to split the block.
 *
 * During deletion we pack the ordered array of the slots
 * and the items themselves by shifting them on the block.
 *
 * The number of bytes the key of an item has in common with the key
 * of the predecessor of the item is stored with each item.
 * This is done in order to minimize the space needed for keys.
 *
 * EG: If a key is "New York" and its predecessor is "New Haven",
 *     only the string "York" is stored for the second key together
 *     with one byte indicating that the key has the first 4 bytes
 *     with its predecessor, the key "New Haven".
 * 
 * Whenever the full keys are needed for the items of a block,
 * the keys of the items of the block have to be "expanded" first.
 */

/*
 * The following three static procedures are the only ones that 'know' the
 * layout of an item stored on a block
 */
static void pblItemToBuf( PBLBLOCK_t * block, int bufoff, PBLITEM_t * item )
{
    unsigned char * ptr = &( block->data[ bufoff ] );

    *ptr++ = ( unsigned char ) item->keylen;
    *ptr++ = ( unsigned char ) item->keycommon;

    if( block->level > 0 )
    {
        ptr += pbl_LongToVarBuf( ptr, item->datablock );
    }
    else
    {
        ptr += pbl_LongToVarBuf( ptr, item->datalen );
    }

    if( item->keylen - item->keycommon > 0 )
    {
        /*
         * make sure we don't copy in place
         */
        if( ptr != item->key + item->keycommon )
        {
            pbl_memlcpy( ptr, PBLKEYLENGTH,
                         item->key + item->keycommon,
                         item->keylen - item->keycommon );
        }
        ptr += item->keylen - item->keycommon;
    }

    /*
     * the block needs to be written to the filesystem
     */
    block->dirty = 1;

    if( block->level > 0 )
    {
        return;
    }

    if( item->datalen <= PBLDATALENGTH )
    {
        if( item->datalen > 0 )
        {
            if( ptr != item->data )
            {
                memcpy( ptr, item->data, item->datalen );
            }
        }
    }
    else
    {
        ptr += pbl_LongToVarBuf( ptr, item->datablock );
        pbl_LongToVarBuf( ptr, item->dataoffset );
    }
}

static int pblBufToItem( PBLBLOCK_t * block, int bufoff, PBLITEM_t * item )
{
    unsigned char * ptr;
    unsigned char * endptr;

    item->level = block->level;

    /*
     * make sure the offset is in bounds
     */
    if( bufoff < 0 || bufoff >= (int)sizeof( block->data ))
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }
    endptr = &( block->data[ sizeof( block->data ) ] );
    ptr = &( block->data[ bufoff ] );

    /*
     * parse the item
     */
    item->keylen = *ptr++;
    item->keycommon = *ptr++;

    if( block->level > 0 )
    {
        ptr += pbl_VarBufToLong( ptr, &( item->datablock ));
        item->datalen = 0;
    }
    else
    {
        ptr += pbl_VarBufToLong( ptr, &( item->datalen ));
        item->datablock = 0;
    }

    if( ptr >= endptr )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    if( item->keylen - item->keycommon > 0 )
    {
        item->key = ptr;
        ptr += item->keylen - item->keycommon;
        if( ptr >= endptr )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }
    }
    else
    {
        item->key = 0;
    }

    if( block->level > 0 )
    {
        item->dataoffset = 0;
        item->data = 0;

        return( 0 );
    }

    if( item->datalen <= PBLDATALENGTH )
    {
        item->dataoffset = 0;
        if( item->datalen > 0 )
        {
            item->data = ptr;
            ptr += item->datalen;
            if( ptr >= endptr )
            {
                pbl_errno = PBL_ERROR_BAD_FILE;
                return( -1 );
            }
        }
        else
        {
            item->data = 0;
        }
    }
    else
    {
        ptr += pbl_VarBufToLong( ptr, &( item->datablock ));
        pbl_VarBufToLong( ptr, &( item->dataoffset ));
        item->data = 0;
    }

    if( ptr >= endptr )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }
    return( 0 );
}

static int pblItemSize( PBLBLOCK_t * block, PBLITEM_t * item )
{
    int rc = 2;

    if( block->level > 0 )
    {
        rc += pbl_LongSize( item->datablock );
    }
    else
    {
        rc += pbl_LongSize( item->datalen );
    }

    if( item->keylen - item->keycommon > 0 )
    {
        rc += item->keylen - item->keycommon;
    }

    if( block->level > 0 )
    {
        return( rc );
    }

    if( item->datalen <= PBLDATALENGTH )
    {
        if( item->datalen > 0 )
        {
            rc += item->datalen;
        }
    }
    else
    {
        rc += pbl_LongSize( item->datablock );
        rc += pbl_LongSize( item->dataoffset );
    }

    return( rc );
}

/*
 * compare two items
 */
static int pblItemCompare( PBLKFILE_t *kf, PBLITEM_t *left, PBLITEM_t *right )
{
    int rc;

    if( kf->keycompare )
    {
        /*
         * there is a specific compare function for the key file
         */
        rc = (*kf->keycompare)( left->key, left->keylen,
                                right->key, right->keylen );
    }
    else
    {
        /*
         * use the default key compare function
         */
        rc = pbl_memcmp( left->key, left->keylen,
                         right->key, right->keylen );
    }

    return( rc );
}

static int pblItemGet( PBLBLOCK_t * block, unsigned int index, PBLITEM_t *item )
{
    /*
     * if there is no item with required index
     */
    if( index >= (unsigned int)block->nentries )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    /*
     * copy the item with the required index
     */
    if( pblBufToItem( block, PBLINDEXTOBUFOFF( block, index ), item ))
    {
        return( -1 );
    }

    /*
     * if we have the expanded keys for the block we
     * actually use those keys
     */
    if( item->keycommon > 0 && block->expandedkeys )
    {
        item->key = block->expandedkeys + index * PBLKEYLENGTH;
    }

    return( 0 );
}

/*
 * append an item to a block
 */
static int pblItemAppend(
PBLBLOCK_t      * block,
unsigned char   * predkey,
unsigned int      predkeylen,
PBLITEM_t       * item
)
{
    unsigned char * ptr;
    unsigned int    itemsize;

    if( !block->writeable )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }

    if( !predkey && block->nentries > 0 )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }

    /*
     * calculate how many bytes the key of the item has in common
     * with the key of the predecessor on the block
     */
    if( predkey && ( predkeylen > 0 ))
    {
        item->keycommon = pbl_memcmplen( predkey, predkeylen,
                                         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 );
    }

    /*
     * put item to data part of block
     */
    pblItemToBuf( block, block->free, item );

    /*
     * put the slot to the block
     */
    ptr = PBLINDEXTOPTR( block, block->nentries );
    pbl_ShortToBuf( ptr, block->free );

    block->free     += itemsize;
    block->nentries += 1;

    if( block->expandedkeys )
    {
        pblBlockKeysRelease( block );
    }

    return( 0 );
}

/*
 * delete an item from a block
 */
static int pblItemDelete( PBLBLOCK_t * block, int index )
{
    PBLITEM_t       item;
    int             ntomove;
    unsigned int    i;
    int             offset;
    int             itemoffset;
    unsigned char * ptr;
    unsigned int    itemsize;

    /*
     * read the item to delete
     */
    if( pblItemGet( block, index, &item ))
    {
        return( -1 );
    }

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

    /*
     * calculate the number of items that have to be moved
     */
    itemoffset = PBLINDEXTOBUFOFF( block, index );
    ptr = block->data + itemoffset;
    ntomove = block->free - ( itemoffset + itemsize );
    if( ntomove < 0 )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    if( ntomove > 0 )
    {
        /*
         * move the other items to the left
         */
        memmove( ptr, ptr + itemsize, ntomove );

        /*
         * the slots who's offsets are to the right of the deleted item
         * have to change, because the items where moved
         */
        for( i = 0; i < block->nentries; i++ )
        {
            offset = PBLINDEXTOBUFOFF( block, i );
            if( offset > itemoffset )
            {
                offset -= itemsize;
                ptr = PBLINDEXTOPTR( block, i );
                pbl_ShortToBuf( ptr, offset );
            }
        }
    }

    /*
     * blank out the bytes deleted
     */
    memset(( block->data + block->free ) - itemsize, 0, itemsize );

    /*
     * there is more free space on the block now
     */
    block->free -= itemsize;

    /*
     * delete the slot from the slots array
     */
    ptr = PBLINDEXTOPTR( block, block->nentries - 1 );
    if( index < (int)block->nentries - 1 )
    {
        ntomove = ( block->nentries - 1 ) - index;
        memmove( ptr + 2, ptr, ntomove * 2 );
    }

    /*
     * blank out the last slot
     */
    *ptr++ = 0;
    *ptr   = 0;

    /*
     * there is one less slot on the block
     */
    block->nentries -= 1;

    if( block->expandedkeys )
    {
        pblBlockKeysRelease( block );
    }

    block->dirty = 1;
    return( 0 );
}

/*
 * insert an item on a block before the item with a given index
 */
static int pblItemInsert( PBLBLOCK_t * block, PBLITEM_t * item, int index )
{
    int             ntomove;
    unsigned char * ptr;
    unsigned int    itemsize;

    /*
     * 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 );
    }

    /*
     * put item to data part of block
     */
    pblItemToBuf( block, block->free, item );

    /*
     * move the slots of the items after index
     */
    ptr = PBLINDEXTOPTR( block, block->nentries );
    ntomove = block->nentries - index;
    if( ntomove > 0 )
    {
        memmove( ptr, ptr + 2, 2 * ntomove );
    }

    /*
     * put the slot to the slots array
     */
    ptr = PBLINDEXTOPTR( block, index );
    pbl_ShortToBuf( ptr, block->free );

    block->free     += itemsize;
    block->nentries += 1;

    if( block->expandedkeys )
    {
        pblBlockKeysRelease( block );
    }

    block->dirty = 1;
    return( 0 );
}

/*
 * remove an item from a block
 */
static int pblItemRemove( PBLBLOCK_t *block, unsigned int index )
{
    PBLITEM_t         peeritem;
    PBLITEM_t         previtem;
    unsigned char     data[ PBLDATALENGTH ];
    unsigned char     savekey[ PBLKEYLENGTH ];
    int               rc;
    int               keycommon;
    int               dodelete = 0;

    if( !block->writeable )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }

    /*
     * if there is no item with required index
     */
    if( index >= block->nentries )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    /*
     * create the expanded keys of the block
     */
    if( pblBlockKeysExpand( block ))
    {
        return( -1 );
    }

    if( index == block->nentries - 1 )
    {
        /*
         * delete the last item on the block
         */
        rc = pblItemDelete( block, index );
        return( rc );
    }

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

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

    if(( index > 0 ) && ( previtem.keylen > 0 ))
    {
        keycommon = pbl_memcmplen( previtem.key, previtem.keylen,
                                   peeritem.key, peeritem.keylen );
    }
    else
    {
        keycommon = 0;
    }

    /*
     * if the next item has to change
     */
    if( keycommon != peeritem.keycommon )
    {
        /*
         * delete and reinsert the next item, because its keycommon changed
         */
        dodelete = 1;

        /*
         * set the new keycommon value for the next item
         */
        peeritem.keycommon = keycommon;

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

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

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

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

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

/*
 * find an item that has a key equal to the search key
 *
 * - uses a binary search to position to an item on a block
 */
static int pblItemFind(
PBLKFILE_t * kf,
PBLBLOCK_t * block,
PBLITEM_t  * item,
int          which
)
{
    int       found = -1;
    int       index = 0;
    int       left  = 0;
    int       right = block->nentries - 1;
    int       rc    = 1;
    PBLITEM_t curitem;

    while( left <= right )
    {
        index = ( left + right ) / 2;

        if( pblItemGet( block, index, &curitem ))
        {
            return( -1 );
        }

        rc = pblItemCompare( kf, &curitem, item );
        if( rc == 0 )
        {
            switch( which )
            {
              case PBLLT:
                right = index - 1;
                break;

              case PBLLE:
              case PBLFI:
                found = index;
                right = index - 1;
                break;

              case PBLEQ:
                found = index;
                return( found );

              case PBLLA:
              case PBLGE:
                found = index;
                left  = index + 1;
                break;

              case PBLGT:
                left  = index + 1;
                break;
            }
        }
        else if ( rc < 0 )
        {
            switch( which )
            {
              case PBLLT:
              case PBLLE:
                found = index;
                left  = index + 1;
                break;

              case PBLFI:
              case PBLEQ:
              case PBLLA:
              case PBLGE:
              case PBLGT:
                left  = index + 1;
                break;
            }
        }
        else
        {
            switch( which )
            {
              case PBLLT:
              case PBLLE:
              case PBLFI:
              case PBLEQ:
              case PBLLA:
                right = index - 1;
                break;

              case PBLGE:
              case PBLGT:
                found = index;
                right = index - 1;
                break;
            }
        }
    }

    if( found < 0 )
    {
        pbl_errno = PBL_ERROR_NOT_FOUND;
    }

    return( found );
}

static int pblItemAdd( PBLKFILE_t * kf, PBLBLOCK_t * block, PBLITEM_t * item )
{
    PBLITEM_t         previtem;
    PBLITEM_t         peeritem;
    int               rc;
    int               index;
    unsigned char     savekey[ PBLKEYLENGTH ];
    unsigned int      savekeylen;
    unsigned char     data[ PBLDATALENGTH ];
    unsigned int      itemsize;
    int               keycommon;

    if( !block->writeable )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }

    if( block->nentries < 1 )
    {
        if( pblItemAppend( block, 0, 0, item ))
        {
            return( -1 );
        }
        return( 0 );
    }

    /*
     * create the expanded keys of the block
     */
    if( pblBlockKeysExpand( block ))
    {
        return( -1 );
    }

    /*
     * find the first item that is bigger than the one we insert
     */
    index = pblItemFind( kf, block, item, PBLGT );
    if( index < 0 )
    {
        if( pbl_errno != PBL_ERROR_NOT_FOUND )
        {
            return( -1 );
        }

        /*
         * append to the block
         */
        pbl_errno = 0;
        index = block->nentries;

        if( pblItemGet( block, index - 1, &peeritem ))
        {
            return( -1 );
        }

        savekeylen = peeritem.keylen;
        if( savekeylen > 0 )
        {
            pbl_memlcpy( savekey, sizeof( savekey ), peeritem.key, savekeylen );
        }

        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 );
                    if( rc )
                    {
                        return( ( PBLBLOCK_t * ) 0 );
                    }
                }

                /*
                 * remove the reference to the block from the hash table
                 */
                pblBlockHashRemove( blockListTail->blockno, blockListTail->bf );
            }

            if(( block = blockListTail ))
            {
                PBL_LIST_UNLINK( blockListHead, blockListTail,
                                 block, next, prev );
                PBL_LIST_PUSH(   blockListHead, blockListTail,
                                 block, next, prev );
            }
        }
        else
        {
            pbl_errno = PBL_ERROR_PROGRAM;
            return( 0 );
        }

        /*
         * make sure the expanded keys of the block are freed
         */
        pblBlockKeysRelease( block );
    }

    block->parentblock  = -1;
    block->parentindex  = -1;
    block->dirty        = 0;
    block->bf           = -1;
    block->filesettag   = NULL;

    return( block );
}

static PBLBLOCK_t * pblBlockFind( PBLKFILE_t * file, long blockno )
{
    PBLBLOCK_t * block;

    /*
     * check if the block is in the LRU list of writeable blocks
     */
    for( block = file->writeableListHead; block; block = block->next )
    {
        if(( block->blockno == blockno )
         &&( block->bf      == file->bf ))
        {
            /*
             * the block is already there, make it the first of the LRU chain
             */
            if( block != file->writeableListHead )
            {
                PBL_LIST_UNLINK( file->writeableListHead,
                                 file->writeableListTail, block, next, prev );
                PBL_LIST_PUSH( file->writeableListHead,
                               file->writeableListTail, block, next, prev );
            }

            return( block );
        }
    }

    /*
     * check if the block is the head of the LRU list of blocks cached
     */
    if( blockListHead
     && blockListHead->blockno == blockno
     && blockListHead->bf      == file->bf )
    {
        return( blockListHead );
    }

    /*
     * lookup the block in the LRU list of blocks cached
     */
    block = pblBlockHashFind( blockno, file->bf );
    if( block )
    {
        /*
         * the block is there, make it the first of the LRU chain
         */
        if( block != blockListHead )
        {
            PBL_LIST_UNLINK( blockListHead, blockListTail, block, next, prev );
            PBL_LIST_PUSH( blockListHead, blockListTail, block, next, prev );
        }

        return( block );
    }

    return( 0 );
}

static PBLBLOCK_t * pblBlockGet( PBLKFILE_t * file, long blockno )
{
    PBLBLOCK_t * block;
    long       rc;

    /*
     * check if block is in memory
     */
    block = pblBlockFind( file, blockno );
    if( block )
    {
        return( block );
    }

    /*
     * block is not in memory, we have to load it; get an empty block
     */
    block = pblBlockGetVictim( file );
    if( !block )
    {
        return( block );
    }

    /*
     * read the data from file
     */
    rc = pbf_blockread( file->bf, blockno, block->data );
    if( rc < 0 )
    {
        return( ( PBLBLOCK_t * ) 0 );
    }

    /*
     * block has been read successfully, so it belongs to this file from now on
     */
    pblBlockToData( block );

    block->blockno    = blockno;
    block->bf         = file->bf;
    block->filesettag = file->filesettag;

    /*
     * insert the reference into the hash table
     */
    if( pblBlockHashInsert( block->blockno, block->bf, block ) )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( 0 );
    }
    return( block );
}

/*
 * get a version of block we can write to to the writeable list of the file
 */
static PBLBLOCK_t * pblBlockGetWriteable( PBLKFILE_t * file, long blockno )
{
    PBLBLOCK_t * newblock;
    PBLBLOCK_t * block;

    /*
     * get the block to memory
     */
    block = pblBlockGet( file, blockno );
    if( !block || block->writeable )
    {
        return( block );
    }

    if( !block->dirty )
    {
        /*
         * move the block over to the writeable list
         */
        pblnblocks--;
        PBL_LIST_UNLINK( blockListHead, blockListTail, block, next, prev );
        pblBlockHashRemove( block->blockno, block->bf );

        block->writeable = 1;
        PBL_LIST_PUSH( file->writeableListHead,
                       file->writeableListTail, block, next, prev );

        return( block );
    }

    /*
     * create a copy of the block in the writeable block list
     */
    newblock = pbl_memdup( "pblBlockGetWriteable block",
                           block, sizeof( *block ) );
    if( !newblock )
    {
        return( newblock );
    }

    newblock->writeable = 1;
    PBL_LIST_PUSH( file->writeableListHead, file->writeableListTail,
                   newblock, next, prev );

    /*
     * make sure the expanded keys are only stored once
     */
    if( block->expandedkeys )
    {
        PBLBLOCKLINK_t  * blink;

        blink = ( PBLBLOCKLINK_t * ) block;
        PBL_LIST_UNLINK( linkListHead, linkListTail, blink, next, prev );

        blink = ( PBLBLOCKLINK_t * ) newblock;
        PBL_LIST_PUSH( linkListHead, linkListTail, blink, next, prev );

        block->expandedkeys = 0;
    }

    return( newblock );
}

static int pblBlockFree( PBLKFILE_t * file, long blockno )
{
    PBLBLOCK_t * rootblock;
    PBLBLOCK_t * block;
    PBLBLOCK_t * nblock = 0;
    PBLBLOCK_t * pblock = 0;

    /*
     * get the root block to memory
     */
    rootblock = pblBlockGetWriteable( file, 0 );
    if( !rootblock )
    {
        return( -1 );
    }

    /*
     * read the block to be freed
     */
    block = pblBlockGetWriteable( file, blockno );
    if( !block )
    {
        return( -1 );
    }

    /*
     * read the previous and next block if they exists
     */
    if( block->nblock )
    {
        nblock = pblBlockGetWriteable( file, block->nblock );
        if( !nblock )
        {
            return( -1 );
        }
    }

    if( block->pblock )
    {
        pblock = pblBlockGetWriteable( file, block->pblock );
        if( !pblock )
        {
            return( -1 );
        }
    }

    if( nblock )
    {
        nblock->pblock = block->pblock;
        nblock->dirty = 1;
    }

    if( pblock )
    {
        pblock->nblock = block->nblock;
        pblock->dirty = 1;
    }

    pblBlockKeysRelease( block );

    /*
     * set the values of the free block
     */
    block->level  = 254;
    block->nblock = rootblock->pblock;

    /*
     * blocks freed always have their predecessor set to 0
     */
    block->pblock   = 0;
    block->nentries = 0;
    block->free     = PBLHEADERSIZE;           /* offset of first free byte   */

    memset( block->data, 0, PBLDATASIZE );

    block->dirty = 1;

    /*
     * set the link from the rootblock to the block
     */
    rootblock->pblock = blockno;
    rootblock->dirty  = 1;

    return( 0 );
}

static int pblBlockConcat(
PBLKFILE_t      * file,
PBLBLOCK_t      * block,
PBLBLOCK_t      * from,
unsigned char   * key,
unsigned int      keylen
)
{
    PBLBLOCK_t      tmpblock;
    unsigned char   predkey[ PBLKEYLENGTH ];
    unsigned int    predkeylen = 0;
    PBLITEM_t       item;
    int             rc;
    unsigned int    i;
    int             nentries;

    if( !block->writeable || !from->writeable )
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }

    /*
     * read the last item of left block
     */
    nentries = block->nentries;
    if( block->nentries > 0 )
    {
        if( pblBlockKeysExpand( block ))
        {
            return( -1 );
        }

        pblItemGet( block, block->nentries - 1, &item );

        predkeylen = item.keylen;
        if( predkeylen )
        {
            pbl_memlcpy( predkey, sizeof( predkey ), item.key, predkeylen );
        }
    }

    /*
     * we do not need the expanded keys of the block anymore
     */
    pblBlockKeysRelease( block );

    /*
     * create a local copy to concatenate to
     */
    tmpblock = *block;

    /*
     * expand the keys of the right block
     */
    if( pblBlockKeysExpand( from ))
    {
        return( -1 );
    }

    /*
     * copy all items to be merged to the temporary block
     */
    for( i = 0; i < from->nentries; i++ )
    {
        pblItemGet( from, i, &item );

        /*
         * the first item can have an empty key, if so we use
         * key given as parameter
         */
        if( i == 0 && keylen > 0 && item.keylen < 1 )
        {
            item.key = key;
            item.keylen = keylen;
        }
        rc = pblItemAppend( &tmpblock, predkey, predkeylen, &item );
        if( rc )
        {
            if( pbl_errno == PBL_ERROR_NOFIT )
            {
                pbl_errno = 0;
                return( 0 );
            }

            return( rc );
        }

        predkeylen = item.keylen;
        if( predkeylen > 0 )
        {
            pbl_memlcpy( predkey, sizeof( predkey ), item.key, predkeylen );
        }
    }

    /*
     * copy the values back to our original block
     */
    block->nentries = tmpblock.nentries;
    block->free     = tmpblock.free;
    memcpy( block->data, tmpblock.data, PBLDATASIZE );

    block->dirty = 1;

    /*
     * change values to the current record if they point to the right block
     */
    if( file->blockno == from->blockno )
    {
        /*
         * set the current record values to the left block
         */
        file->blockno = block->blockno;
        file->index  += nentries;
    }

    return( 1 );
}

static long pblBlockAppend(
PBLKFILE_t * file,
int          level,
long         nblock,
long         pblock
)
{
    PBLBLOCK_t      * rootblock = 0;
    PBLBLOCK_t      * block = 0;
    long              freeblockno = 0;

    /*
     * if nblock is 1, we are called to create the first block of a file
     * no need to try to read any block of the file
     */
    if( nblock != 1 && pblock != 1 )
    {
        /*
         * get the root block to memory
         */
        rootblock = pblBlockGetWriteable( file, 0 );
        if( !rootblock )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        /*
         * read block number of first free block of file
         */
        freeblockno = rootblock->pblock;
        if( freeblockno )
        {
            /*
             * read the free block
             */
            block = pblBlockGetWriteable( file, freeblockno );
            if( block )
            {
                if( block->level != 254 )
                {
                    pbl_errno = PBL_ERROR_BAD_FILE;
                    return( -1 );
                }

                /*
                 * set the next free block to the rootblock
                 */
                rootblock->pblock = block->nblock;
                if( !rootblock->pblock && block->pblock )
                {
                    /*
                     * the block read is a new block,
                     * set the next block of the root block to a big value
                     */
                    rootblock->pblock = block->pblock;
                }

                rootblock->dirty = 1;
            }
            else
            {
                /*
                 * no error, append a new block
                 */
                pbl_errno = 0;
                freeblockno += 1;
            }
        }
    }

    if( !block )
    {
        PBLBLOCK_t newblock;

        /*
         * init the new block
         */
        memset( &newblock, 0, sizeof( newblock ));

        /*
         * add a free block
         */
        newblock.level = ( char ) 254 & 0xff;

        /*
         * new blocks appended have their next block set to 0
         * and their previous block set to the highest free block
         *
         * blocks freed always set the pblock to 0
         */
        newblock.nblock = 0;
        newblock.pblock = freeblockno;;
        newblock.free   = PBLHEADERSIZE;

        /*
         * prepare the new block for writing
         */
        pblDataToBlock( &newblock );

        /*
         * append a new block to the file
         */
        freeblockno = pbf_blockappend( file->bf, newblock.data );
        if( freeblockno < 0 )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        /*
         * read the free block
         */
        block = pblBlockGetWriteable( file, freeblockno );
        if( !block || block->level != 254 )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        if( rootblock )
        {
            /*
             * set the next free block to the rootblock
             */
            rootblock->pblock = block->pblock;
            if( !rootblock->pblock )
            {
                rootblock->pblock = 1;
            }
            rootblock->dirty = 1;
        }
    }

    /*
     * init the new block
     */
    memset( block->data, 0, PBLDATASIZE );

    block->level      = ( char ) level & 0xff;
    block->nentries   = 0;
    block->nblock     = nblock;
    block->pblock     = pblock;
    block->free       = PBLHEADERSIZE;
    block->dirty      = 1;

    return( block->blockno );
}

static int pblBlockMerge(
PBLKFILE_t * file,
long         parentblockno,
int          parentindex,
long         blockno
)
{
    int             merged = 0;
    PBLBLOCK_t    * parentblock;
    PBLBLOCK_t    * block;
    PBLBLOCK_t    * peerblock;
    PBLITEM_t       item;
    int             rc;
    unsigned char   key[ PBLKEYLENGTH ];
    unsigned int    keylen;

    /*
     * read the parentblock
     */
    parentblock = pblBlockGet( file, parentblockno );
    if( !parentblock )
    {
        /*
         * no error because the parent block might have been split
         */
        pbl_errno = 0;
        return( merged );
    }

    /*
     * check the parentindex because the parentblock might have been
     * split without the child knowing about it
     */
    if( parentindex >= (int)parentblock->nentries )
    {
        return( merged );
    }

    /*
     * read the item pointing to blockno
     */
    if( pblItemGet( parentblock, parentindex, &item ))
    {
        return( -1 );
    }

    /*
     * check the pointer to the child, because the parentblock might have been
     * split without the child knowing about it
     */
    if( blockno != item.datablock )
    {
        return( merged );
    }

    /*
     * if there is a block to the left
     */
    while( parentindex > 0 )
    {
        /*
         * check the parentindex because the parentblock might have been
         * split without the child knowing about it
         */
        if( parentindex >= (int)parentblock->nentries )
        {
            return( merged );
        }

        /*
         * read the item pointing to blockno
         */
        if( pblItemGet( parentblock, parentindex, &item ))
        {
            return( -1 );
        }

        /*
         * set the pointer to the child
         */
        blockno = item.datablock;

        /*
         * read the child block
         */
        block = pblBlockGet( file, blockno );
        if( !block )
        {
            return( -1 );
        }

        /*
         * read the item pointing to the peer
         */
        if( pblItemGet( parentblock, parentindex - 1, &item ))
        {
            return( -1 );
        }

        /*
         * read the peerblock
         */
        peerblock = pblBlockGet( file, item.datablock );
        if( !peerblock )
        {
            return( -1 );
        }

        /*
         * see how much empty space we have on the two blocks
         */
        rc = PBLINDEXBLOCKNFREE( block ) + PBLINDEXBLOCKNFREE( peerblock );
        if( rc < ( PBLDATASIZE + 6 + PBLKEYLENGTH ))
        {
            /*
             * we do not merge
             */
            break;
        }

        /*
         * read the child block
         */
        block = pblBlockGetWriteable( file, blockno );
        if( !block )
        {
            return( -1 );
        }

        /*
         * read the peerblock
         */
        peerblock = pblBlockGetWriteable( file, item.datablock );
        if( !peerblock )
        {
            return( -1 );
        }

        /*
         * read the first key of the right block to merge
         */
        parentblock = pblBlockGetWriteable( file, parentblockno );
        if( !parentblock )
        {
            return( -1 );
        }

        /*
         * check the parentindex
         */
        if( parentindex >= (int)parentblock->nentries )
        {
            return( merged );
        }

        if( pblBlockKeysExpand( parentblock ))
        {
            return( -1 );
        }

        /*
         * read the item pointing to blockno
         */
        if( pblItemGet( parentblock, parentindex, &item ))
        {
            return( -1 );
        }

        if( item.keylen < 1 )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        keylen = item.keylen;
        pbl_memlcpy( key, sizeof( key ), item.key, keylen );

        /*
         * concatenate the two blocks
         */
        rc = pblBlockConcat( file, peerblock, block, key, keylen );
        if( rc < 0 )
        {
            return( rc );
        }
        else if( rc == 0 )
        {
            /*
             * we could not merge, break the loop
             */
            break;
        }

        /*
         * the two blocks were merged
         */
        merged += 1;

        /*
         * free the block
         */
        rc = pblBlockFree( file, blockno );
        if( rc )
        {
            return( rc );
        }

        rc = pblItemRemove( parentblock, parentindex );
        if( rc )
        {
            return( rc );
        }
    }

    /*
     * if there is a block to the left
     */
    while( parentindex < (int)parentblock->nentries - 1 )
    {
        /*
         * read the item pointing to blockno
         */
        if( pblItemGet( parentblock, parentindex, &item ))
        {
            return( -1 );
        }

        /*
         * set the pointer to the child
         */
        blockno = item.datablock;

        /*
         * read the child block
         */
        block = pblBlockGet( file, blockno );
        if( !block )
        {
            return( -1 );
        }

        /*
         * read the item pointing to the peer
         */
        if( pblItemGet( parentblock, parentindex + 1, &item ))
        {
            return( -1 );
        }

        /*
         * read the peerblock
         */
        peerblock = pblBlockGet( file, item.datablock );
        if( !peerblock )
        {
            return( -1 );
        }

        /*
         * see how much empty space we have on the two blocks
         */
        rc = PBLINDEXBLOCKNFREE( block ) + PBLINDEXBLOCKNFREE( peerblock );
        if( rc < ( PBLDATASIZE + 6 + PBLKEYLENGTH ))
        {
            /*
             * we do not merge
             */
            break;
        }

        /*
         * read the child block
         */
        block = pblBlockGetWriteable( file, blockno );
        if( !block )
        {
            return( -1 );
        }

        /*
         * read the peerblock
         */
        blockno = item.datablock;
        peerblock = pblBlockGetWriteable( file, blockno );
        if( !peerblock )
        {
            return( -1 );
        }

        /*
         * read the first key of the right block to merge
         */
        parentblock = pblBlockGetWriteable( file, parentblockno );
        if( !parentblock )
        {
            return( -1 );
        }

        /*
         * check the parentindex
         */
        if( parentindex >= (int)parentblock->nentries - 1 )
        {
            return( merged );
        }

        if( pblBlockKeysExpand( parentblock ))
        {
            return( -1 );
        }

        /*
         * read the item pointing to blockno
         */
        if( pblItemGet( parentblock, parentindex + 1, &item ))
        {
            return( -1 );
        }

        if( item.keylen < 1 )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        keylen = item.keylen;
        pbl_memlcpy( key, sizeof( key ), item.key, keylen );

        /*
         * concatenate the two blocks
         */
        rc = pblBlockConcat( file, block, peerblock, key, keylen );
        if( rc < 0 )
        {
            return( rc );
        }
        else if( rc == 0 )
        {
            /*
             * we could not merge, break the loop
             */
            break;
        }

        /*
         * the two blocks were merged
         */
        merged += 1;

        /*
         * free the block
         */
        rc = pblBlockFree( file, blockno );
        if( rc )
        {
            return( rc );
        }

        rc = pblItemRemove( parentblock, parentindex + 1 );
        if( rc )
        {
            return( rc );
        }
    }

    return( merged );
}

/*
 * truncate the blocklist to the number of blocks allowed
 */
static int pblBlockListTruncate( void )
{
    PBLBLOCK_t * block;
    int          rc;

    /*
     * truncate the list of blocks we have in memory
     */
    while( pblnblocks >= 8 + ( pblblocksperfile * pblnfiles ) )
    {
        block = blockListTail;
        if( !block )
        {
            pblnblocks = 0;
            break;
        }

        if( block->bf != -1 )
        {
            if( block->dirty )
            {
                /*
                 * if one block of a file is dirty, all blocks are flushed
                 */
                rc = pblBlockFlush( block->bf, 0 );
                if( rc )
                {
                    return( rc );
                }
            }

            pblBlockHashRemove( block->blockno, block->bf );
        }

        PBL_LIST_UNLINK( blockListHead, blockListTail, block, next, prev );
        pblBlockKeysRelease( block );
        PBL_FREE( block );

        pblnblocks--;
    }

    return( 0 );
}

/**
 * change the number of cache blocks used per open key file
 *
 * the default number is 64, a memory block uses about 4096 bytes of heap memory
 *
 * @return int rc: the number of blocks used after the call
 *
 *
 */

int pblKfInit(
int nblocks            /* number of blocks used per open file */
)
{
    pbl_errno = 0;

    if( nblocks < 1 )
    {
        return( pblnblocks );
    }

    if( nblocks < 8 )
    {
        nblocks = 8;
    }

    pblblocksperfile = nblocks;
    pblexpandedperfile = pblblocksperfile / 2;

    return( pblblocksperfile );
}

/*
 * FILE functions
 */
static long pblDataAppend(
PBLKFILE_t * file,
char       * data,
long         datalen,
long       * offset
)
{
    long blockno;
    long returnoffset;
    long returnblock;
    long diff;
    long bytesWritten = 0;
    int  nbytes;
    PBLBLOCK_t * rootblock = 0;
    PBLBLOCK_t * datablock = 0;

    rootblock = pblBlockGet( file, 0 );
    if( !rootblock )
    {
        return( -1 );
    }

    /*
     * rootblock->nblock always contains the number of the last datablock
     */
    datablock = pblBlockGetWriteable( file, rootblock->nblock );
    if( !datablock )
    {
        return( -1 );
    }

    if( datablock->level != 255 )
    {
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( -1 );
    }

    returnoffset = datablock->free;
    returnblock  = datablock->blockno;

    while( bytesWritten < datalen )
    {
        diff = datalen - bytesWritten;
        if( diff > PBLDATABLOCKNFREE( datablock ))
        {
            nbytes = PBLDATABLOCKNFREE( datablock );
        }
        else
        {
            nbytes = (( int )( diff % PBLDATASIZE));
        }

        if( nbytes > 0 )
        {
            memcpy((void *) &(datablock->data[ datablock->free ]),
                   (void *) data,
                   nbytes );
            datablock->dirty = 1;
        }

        bytesWritten        += nbytes;
        data                += nbytes;
        datablock->free += nbytes;

        if( PBLDATABLOCKNFREE( datablock ) < 1 )
        {
            /*
             * make a new data block
             */
            blockno = pblBlockAppend( file, -1, 0, datablock->blockno );
            if( blockno < 0 )
            {
                return( -1 );
            }

            datablock->nblock = blockno;
            datablock->dirty  = 1;

            /*
             * get the new datablock to memory
             */
            datablock = pblBlockGetWriteable( file, blockno );
            if( !datablock )
            {
                return( -1 );
            }

            /*
             * set address to rootblock
             */
            rootblock = pblBlockGetWriteable( file, 0 );
            if( !rootblock )
            {
                return( -1 );
            }

            rootblock->nblock = blockno;
            rootblock->dirty  = 1;
        }
    }

    *offset = returnoffset;

    return( returnblock );
}

/**
 * create a key file with the name specified by path.
 *
 * a file set tag can be attached to the file,
 * if a file having a non NULL file set tag is flushed
 * to disk all files having the same file set tag attached
 * are flushed as well.
 *
 * <P>
 * <B>RESTRICTIONS</B>:
 * <BR> - the file to create must not exists.
 * <BR> - the current record of the file will not be set
 *
 * @return  pblKeyFile_t * retptr == NULL: an error occured, see pbl_errno
 * @return  pblKeyFile_t * retptr != NULL: a pointer to a key file descriptor
 */

pblKeyFile_t * pblKfCreate(
char * path,       /** path of file to create                                 */
void * filesettag  /** file set tag, for flushing multiple files consistently */
)
{
    pblKeyFile_t * k         = NULL;
    PBLKFILE_t   * kf        = NULL;
    PBLBLOCK_t   * rootblock = NULL;
    PBLITEM_t      item;
    int            fh;
    int            bf;
    long           blockno;
    int            rc;

    pbl_errno = 0;

    /*
     * make sure we have one filehandle for the create, close one file
     */
    if( pbf_ft_tail )
    {
        pbf_fh_close( pbf_ft_tail->bf, pbf_ft_tail->n );
    }

    /*
     * do a exclusive create open, make sure the file does not exist yet
     */
    fh = open( path, O_CREAT | O_EXCL | O_BINARY | O_RDWR, S_IREAD | S_IWRITE );
    if( -1 == fh )
    {
        pbl_errno = PBL_ERROR_CREATE;
        return( 0 );
    }
    close( fh );

    /*
     * open the file
     */
    bf = pbf_open( path, 1, PBLFILEBLOCKS, PBLDATASIZE );
    if( bf < 0 )
    {
        return( 0 );
    }

    /*
     * get and init file structure
     */
    kf = pbl_malloc0( "pblKfCreate FILE", sizeof( PBLKFILE_t ));
    if( !kf )
    {
        goto errout;
    }

    /*
     * we have a key file set the return value
     */
    k              = ( pblKeyFile_t * )kf;
    kf->magic      = rcsid;
    kf->bf         = bf;
    kf->update     = 1;
    kf->filesettag = filesettag;

    /*
     * start a transaction on the file
     */
    pblKfStartTransaction( k );

    /*
     * make the root block, next offset is first data block
     */
    blockno = pblBlockAppend( kf, 0, 1, 1 );
    if( blockno < 0 )
    {
        goto errout;
    }

    rootblock = pblBlockGet( kf, 0 );
    if( !rootblock )
    {
        goto errout;
    }

    /*
     * make the first data block
     */
    blockno = pblBlockAppend( kf, -1, 0, 0 );
    if( blockno != 1 )
    {
        goto errout;
    }

    /*
     * init the first item we insert into each file
     */
    item.level     = 0;
    item.key       = ( char * ) 0;
    item.keylen    = 0;
    item.keycommon = 0;
    item.datalen   = strlen( magic ) + 1;
    item.data      = magic;

    /*
     * append the magic string as first data item
     */
    if( item.datalen > PBLDATALENGTH )
    {
        item.datablock = pblDataAppend( kf, item.data,
                                        item.datalen, &item.dataoffset );
        if( item.datablock < 1 )
        {
            goto errout;
        }
        item.data = 0;
    }

    /*
     * insert the first item into the root block
     */
    rootblock = pblBlockGetWriteable( kf, 0 );
    if( !rootblock )
    {
        goto errout;
    }

    rc = pblItemAdd( kf, rootblock, &item );
    if( rc )
    {
        goto errout;
    }

    /*
     * no current record yet
     */
    kf->blockno = -1;
    kf->index   = -1;

    /*
     * commit the changes
     */
    if( pblKfCommit( k, 0 ))
    {
        goto errout;
    }

    if( pblBlockFlush( kf->bf, 0 ))
    {
        goto errout;
    }

    pblnfiles++;

    return( k );

errout:

    if( kf )
    {
        if( kf->bf >= 0 )
        {
            pblKfCommit( k, 1 );
        }
        PBL_FREE( kf );
    }

    if( -1 != bf )
    {
        pblBlocksRelease( bf );
        close( fh );
        unlink( path );
    }

    return( 0 );
}

/**
 * open an existing key file
 * 
 * if update is 0, the file is opened for read access only,
 * if update is not 0 the file is opened for reading and writing
 *
 * a file set tag can be attached to the file,
 * if a file having a non NULL file set tag is flushed
 * to disk all files having the same file set tag attached
 * are flushed as well.
 *
 * <P>
 * <B>RESTRICTIONS</B>:
 * <BR> - the file must exist already
 * <BR> - the current record of the file will not be set
 *
 * @return  pblKeyFile_t * retptr == NULL: an error occured, see pbl_errno
 * @return  pblKeyFile_t * retptr != NULL: a pointer to a key file descriptor
 */

pblKeyFile_t * pblKfOpen(
char * path,       /** path of file to create                                 */
int    update,     /** flag: should file be opened for update?                */
void * filesettag  /** file set tag, for flushing multiple files consistently */
)
{
    PBLKFILE_t  * kf;
    PBLBLOCK_t  * datablock;
    int           bf;

    pbl_errno = 0;

    bf = pbf_open( path, update, PBLFILEBLOCKS, PBLDATASIZE );
    if( -1 == bf )
    {
        return( 0 );
    }

    /*
     * get and init file structure
     */
    kf = pbl_malloc0( "pblKfOpen FILE", sizeof( PBLKFILE_t ));
    if( !kf )
    {
        pbf_close( bf );
        return( 0 );
    }
    kf->magic      = rcsid;
    kf->bf         = bf;
    kf->update     = update;
    kf->filesettag = filesettag;
    kf->blockno    = -1;
    kf->index      = -1;

    /*
     * read and check the first datablock
     */
    datablock = pblBlockGet( kf, 1 );
    if( !datablock || ( datablock->level != 255 ))
    {
        pblBlocksRelease( bf );
        pbf_close( bf );
        PBL_FREE( kf );
        pbl_errno = PBL_ERROR_BAD_FILE;
        return( 0 );
    }

    pblnfiles++;

    return( ( pblKeyFile_t * ) kf );
}

/**
 * close a key file
 *
 * all changes are flushed to disk before, 
 * all memory allocated for the file is released. 
 *
 * @return int rc == 0: call went ok, file is closed
 * @return int rc != 0: some error, see pbl_errno
 */

int pblKfClose(
pblKeyFile_t * k   /** key file to close */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;
    int rc = 0;

    pbl_errno = 0;

    if( kf->update )
    {
        rc = pblBlockFlush( kf->bf, 1 );
    }
    else
    {
        pblBlocksRelease( kf->bf );
    }

    pbf_close( kf->bf );
    PBL_FREE( kf );

    pblnfiles--;

    if( pblBlockListTruncate())
    {
        rc = -1;
    }

    return( rc );
}

/**
 * set an application specific compare function for the keys of a key file
 *
 * an application specific compare function can be used in order to
 * implement special orderings of the values of an index, e.g.
 * because of the use of european "umlauts" in names
 *
 * the default compare function is the c-library memcmp function
 * the keycompare function should behave like memcmp
 *
 * @return void
 */

void pblKfSetCompareFunction(
pblKeyFile_t * k,             /** key file to set compare function for  */
int ( *keycompare )           /** compare function to set               */
    (
                void* left,   /** "left" buffer for compare             */
                size_t llen,  /** length of that buffer                 */
                void* right,  /** "right" buffer for compare            */
                size_t rlen   /** length of that buffer                 */
    )
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;

    kf->keycompare = keycompare;
}

/**
 * flush a key file
 *
 * all changes are flushed to disk,
 *
 * @return int rc == 0: call went ok
 * @return int rc != 0: some error, see pbl_errno
 */

int pblKfFlush(
pblKeyFile_t * k   /** key file to flush */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;
    int rc;

    pbl_errno = 0;

    rc = pblBlockFlush( kf->bf, 0 );
    if( rc )
    {
        return( rc );
    }

    rc = pblBlockListTruncate();

    return( rc );
}

/**
 * commit or rollback changes done during a transaction.
 *
 * transactions can be nested, if so the commit
 * only happens when the outermost transaction
 * calls a commit.
 *
 * the commit only happens to process space buffer cache,
 * call \Ref{pblKfFlush}() after \Ref{pblKfCommit}() if you want to
 * flush to kernel space buffer cache.
 *
 * @return int rc == 0: the commit went ok
 * @return int rc >  0: a rollback happened, either because the caller
 *                      requested it or because an inner transaction resulted
 *                      in a rollback
 * @return int rc <  0: some error, see pbl_errno
 */

int pblKfCommit(
pblKeyFile_t * k, /** key file to commit                                      */
int rollback      /** != 0: roll back the changes, == 0: commit the changes   */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;
    PBLBLOCK_t * block;
    PBLBLOCK_t * b;

    pbl_errno = 0;

    /*
     * if there is no transaction active for the file
     */
    if( !rollback && ( kf->transactions < 1 ))
    {
        pbl_errno = PBL_ERROR_PROGRAM;
        return( -1 );
    }
    kf->transactions--;

    /*
     * if a rollback is requested for this commit
     */
    if( rollback )
    {
        /*
         * remember that at least one rollback is requested for the file
         */
        kf->rollback = 1;
    }

    /*
     * if there is an outer transaction active for the file
     */
    if( kf->transactions > 0 )
    {
        return( kf->rollback );
    }

    /*
     * there is no more transaction active, rollback or commit
     */
    if( kf->rollback )
    {
        /*
         * release all blocks that were changed without putting
         * them back into the blocklist buffer cache
         */
        while(( block = kf->writeableListTail ))
        {
            PBL_LIST_UNLINK( kf->writeableListHead, kf->writeableListTail,
                             block, next, prev );
            pblBlockKeysRelease( block );
            PBL_FREE( block );
            continue;
        }

        /*
         * reset the transaction and the rollback value
         */
        kf->transactions = 0;
        kf->rollback = 0;
        return( 1 );
    }

    /*
     * commit the changed blocks
     */
    while(( block = kf->writeableListTail ))
    {
        /*
         * commit all blocks that were changed by rechaining
         * them into the blocklist buffer cache
         */
        PBL_LIST_UNLINK( kf->writeableListHead, kf->writeableListTail,
                         block, next, prev );

        /*
         * find a potential copy of the block in the LRU list
         */
        b = pblBlockFind( kf, block->blockno );
        if( b )
        {
            /*
             * delete the copy from the LRU list
             */
            PBL_LIST_UNLINK( blockListHead, blockListTail, b, next, prev );
            pblBlockKeysRelease( b );
            PBL_FREE( b );
        }
        else
        {
            /*
             * we add a block to the LRU list
             */
            pblnblocks++;
        }

        /*
         * blocks in the buffer cache are not writeable
         */
        block->writeable = 0;
        PBL_LIST_PUSH( blockListHead, blockListTail, block, next, prev );

        /*
         * insert or update the reference in the hash table
         */
        if( pblBlockHashInsert( block->blockno, block->bf, block ) < 0 )
        {
            pbl_errno = PBL_ERROR_PROGRAM;
            return( 0 );
        }
    }

    /*
     * reset the transaction and the rollback value
     */
    kf->transactions = 0;
    kf->rollback = 0;

    return( 0 );
}

/**
 * start a transaction on a key file
 *
 * transactions can be nested
 *
 * @return int rc == 0: the transaction was started successfully
 * @return int rc >  0: the transaction was started
 *                      but another transaction has resulted in
 *                      a rollback request on the file already
 */

int pblKfStartTransaction(
pblKeyFile_t * k           /** key file to start transaction on               */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;
    pbl_errno = 0;

    /*
     * if there is no transaction active for the file
     */
    if( kf->transactions < 1 )
    {
        kf->transactions = 1;
        kf->rollback = 0;
        return( 0 );
    }
    kf->transactions++;

    return( kf->rollback );
}

/**
 * save the position of the current record for later restore
 *
 * @return int rc == 0: the position was saved
 * @return int rc  < 0: an error, see pbl_errno
 */
int pblKfSavePosition(
pblKeyFile_t * k       /** key file to save position for                     */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;

    pbl_errno = 0;

    /*
     * save the block number and the index
     */
    kf->saveblockno = kf->blockno;
    kf->saveindex   = kf->index;

    return( 0 );
}

/**
 * restore the position of the current record saved by the
 * last previous call to \Ref{pblKfSavePosition}().
 *
 * @return int rc == 0: the position was restored
 * @return int rc  < 0: an error, see pbl_errno
 */

int pblKfRestorePosition(
pblKeyFile_t * k       /** key file to restore position for                   */
)
{
    PBLKFILE_t * kf = ( PBLKFILE_t * ) k;

    /*
     * restore the block number and the index
     */
    kf->blockno = kf->saveblockno;
    kf->index   = kf->saveindex;

    return( 0 );
}

/*
 * INSERT functions
 */
static int pblItemSave( PBLITEM_t * item )
{
    PBLITEM_t * newitem  = ( PBLITEM_t * ) 0;

    newitem = pbl_malloc0( "pblItemSave ITEM", sizeof( PBLITEM_t ));
    if( !newitem )
    {
        return( -1 );
    }

    /*
     * save the values of the item
     */
    *newitem = *item;

    /*
     * save the key
     */
    if( newitem->keylen > 0 )
    {
        newitem->key = pbl_memdup( "pblItemSave item->key",
                                   item->key, newitem->keylen );
        if( !newitem->key )
        {
            PBL_FREE( newitem );
            return( -1 );
        }
    }
    else
    {
        newitem->key = 0;
    }

    /*
     * save the data
     */
    if( newitem->datalen < 1 || newitem->datalen > PBLDATALENGTH )
    {
        newitem->data = 0;
    }

    /*
     * push the new item to top of list
     */
    PBL_LIST_PUSH( itemListHead, itemListTail, newitem, next, prev );

    return( 0 );
}

static void pblItemRelease( void )
{
    PBLITEM_t * item;

    if(( item = itemListHead ))
    {
        PBL_LIST_UNLINK( itemListHead, itemListTail, item, next, prev );

        if( item->key )
        {
            PBL_FREE( item->key );
        }
        PBL_FREE( item );
    }
}

static void pblItemReleaseAll( void )
{
    while( itemListHead )
    {
        pblItemRelease( );
    }
}

#if 1

static int pblSplit( PBLKFILE_t *file, PBLBLOCK_t * block )
{
    unsigned int   index;
    PBLITEM_t      splititem;
    PBLITEM_t      item;
    PBLBLOCK_t     tmpblock;
    PBLBLOCK_t    *newblock;
    PBLBLOCK_t    *target;
    unsigned char *predkey = "";
    unsigned int   predkeylen = 0;
    long           newblockno;
    int            rc;

    /*
     * create a new block
     */
    newblockno = pblBlockAppend( file, block->level,
                                 block->nblock, block->blockno );
    if( newblockno < 0 )
    {
        return( -1 );
    }

    /*
     * set backward pointer in successor block of block
     */
    if( block->nblock )
    {
        newblock = pblBlockGetWriteable( file, block->nblock );
        if( !newblock )
        {
            return( -1 );
        }

        newblock->pblock = newblockno;
        newblock->dirty  = 1;
    }

    /*
     * get the new block to memory
     */
    newblock = pblBlockGetWriteable( file, newblockno );
    if( !newblock )
    {
        return( -1 );
    }

    /*
     * copy the block to split onto the stack
     */
    pblBlockKeysRelease( block );
    tmpblock = *block;

    /*
     * prepare the block for split
     */
    block->nblock   = newblockno;
    block->nentries = 0;
    block->free     = PBLHEADERSIZE;
    block->dirty    = 1;
    memset( block->data, 0, PBLDATASIZE );

    tmpblock.expandedkeys = 0;
    if( pblBlockKeysExpand( &tmpblock ))
    {
       return( -1 );
    }

    /*
     * copy the items from tmpblock to our two real blocks
     */
    for( target = block, index = 0; index < tmpblock.nentries; index++ )
    {
        rc = pblItemGet( &tmpblock, index, &item );
        if( rc )
        {
            pblBlockKeysRelease( &tmpblock );
            return( -1 );
        }

        /*
         * if first block we copy to is more than half full
         */
        if(  ( target == block )
         && (( PBLINDEXBLOCKNFREE( block ) < ( PBLDATASIZE / 2 ))
          || ( index == tmpblock.nentries - 1 )))
        {
            /*
             * the first item that goes to the new second block has to be
             * saved for later insert into the father block of the new block
             */
            splititem = item;
            splititem.datalen    = 0;
            splititem.datablock  = newblockno;
            splititem.dataoffset = 0;
            splititem.level++;

            rc = pblItemSave( &splititem );
            if( rc )
            {
                pblBlockKeysRelease( &tmpblock );
                return( -1 );
            }

            /*
             * for blocks of level greater than 0 the first item on each block
             * has keylength 0
             */
            if( tmpblock.level > 0 )
            {
                item.keylen = 0;
            }

            /*
             * from now on copy to the second block
             */
            target = newblock;
            predkeylen = 0;
        }

        rc = pblItemAppend( target, predkey, predkeylen, &item );
        if( rc < 0 )
        {
            pblBlockKeysRelease( &tmpblock );
            return( -1 );
        }

        predkeylen = item.keylen;
        if( predkeylen > 0 )
        {
            predkey = item.key;
        }
        else
        {
            predkey = "";
        }
    }

    pblBlockKeysRelease( &tmpblock );

    /*
     * set the parent pointers to the new block
     */
    newblock->parentblock = block->parentblock;
    newblock->parentindex = block->parentindex + 1;

    return( 0 );
}

#else

static int pblSplit( PBLKFILE_t * file, PBLBLOCK_t * block )
{
    unsigned int   index;
    int            splitindex = 0;
    PBLITEM_t      splititem;
    PBLITEM_t      item;
    PBLBLOCK_t    *newblock;
    PBLBLOCK_t    *target;
    unsigned char *predkey = "";
    unsigned int   predkeylen = 0;
    long           newblockno;
    int            rc;
    int            blockfree = PBLHEADERSIZE;

    /*
     * create a new block
     */
    newblockno = pblBlockAppend( file, block->level,
                                 block->nblock, block->blockno );
    if( newblockno < 0 )
    {
        return( -1 );
    }

    /*
     * set backward pointer in successor block of block
     */
    if( block->nblock )
    {
        newblock = pblBlockGetWriteable( file, block->nblock );
        if( !newblock )
        {
            return( -1 );
        }

        newblock->pblock = newblockno;
        newblock->dirty  = 1;
    }

    /*
     * get the new block to memory
     */
    newblock = pblBlockGetWriteable( file, newblockno );
    if( !newblock )
    {
        return( -1 );
    }

    /*
     * expand the keys of the block to split
     */
    if( pblBlockKeysExpand( block ))
    {
       return( -1 );
    }

    /*
     * prepare the block for split
     */
    block->nblock = newblockno;

    /*
     * copy the items from the block to our new block
     */
    for( target = block, index = 0; index < block->nentries; index++ )
    {
        rc = pblItemGet( block, index, &item );
        if( rc )
        {
            return( -1 );
        }

        if( target == block )
        {
            /*
             * calculate the size the items need on the block so far
             */
            blockfree += pblItemSize( block, &item );

            /*
             * if first block we copy to is more than half full
             */
            if( ( blockfree > ( PBLDATASIZE / 2 ))
             || ( index == block->nentries - 1 ))
            {
                /*
                 * the first item that goes to the new second block has to be
                 * saved for later insert into the father block of the new block
                 */
                splitindex = index;
                splititem  = item;
                splititem.datalen    = 0;
                splititem.datablock  = newblockno;
                splititem.dataoffset = 0;
                splititem.level++;

                rc = pblItemSave( &splititem );
                if( rc )
                {
                    return( -1 );
                }

                /*
                 * for blocks of level greater than 0 the first item
                 * on each block has keylength 0
                 */
                if( block->level > 0 )
                {
                    item.keylen = 0;
                }

                /*
                 * from now on copy to the second block
                 */
                target = newblock;
                predkeylen = 0;
            }
        }

        /*
         * only copy to the new block if the target is right
         */
        if( target == newblock )
        {
            rc = pblItemAppend( target, predkey, predkeylen, &item );
            if( rc < 0 )
            {
                return( -1 );
            }

            /*
             * remember the key of the predecessor item
             */
            predkeylen = item.keylen;
            if( predkeylen > 0 )
            {
                predkey = item.key;
            }
            else
            {
                predkey = "";
            }
        }
    }
    
    /*
     * delete the items that were copied from the block
     */
    for( index = block->nentries - 1; index >= splitindex; index-- )
    {
        rc = pblItemDelete( block, index );
        if( rc )
        {
            return( -1 );
        }
    }

    /*
     * set the parent pointers to the new block
     */
    newblock->parentblock = block->parentblock;
    newblock->parentindex = block->parentindex + 1;

    return( 0 );
}

#endif

static int pblSplitRoot( PBLKFILE_t *file )
{
    PBLBLOCK_t * rootblock;
    PBLITEM_t    item;
    PBLBLOCK_t * newblock;
    long         newblockno;
    int          rc;

    /*
     * get the root block to memory
     */
    rootblock = pblBlockGetWriteable( file, 0 );
    if( !rootblock )
    {
        return( -1 );
    }
    pblBlockKeysRelease( rootblock );

    /*
     * create a new block and get it to memory
     */
    newblockno = pblBlockAppend( file, rootblock->level, 0, 0 );
    if( newblockno < 0 )
    {
        return( -1 );
    }

    newblock = pblBlockGetWriteable( file, newblockno );
    if( !newblock )
    {
        return( -1 );
    }

    /*
     * copy some data of the root block to the new block
     */
    newblock->nentries = rootblock->nentries;
    newblock->free     = rootblock->free;
    memcpy( newblock->data, rootblock->data, PBLDATASIZE );

    newblock->dirty    = 1;

    /*
     * get the root block to memory
     */
    rootblock = pblBlockGetWriteable( file, 0 );
    if( !rootblock )
    {
        return( -1 );
    }

    /*
     * clear the root block
     */
    rootblock->level   += 1;
    rootblock->nentries = 0;
    rootblock->free     = PBLHEADERSIZE;
    rootblock->dirty    = 1;
    memset( rootblock->data, 0, PBLDATASIZE );

    newblock = pblBlockGetWriteable( file, newblockno );
    if( !newblock )
    {
        return( -1 );
    }

    /*
     * copy the first item from new block to the root block
     */
    rc = pblItemGet( newblock, 0, &item );
    if( rc )
    {
        return( -1 );
    }

    item.level      = rootblock->level;
    item.keylen     = 0;
    item.datalen    = 0;
    item.datablock  = newblockno;
    item.dataoffset = 0;

    rc = pblItemAppend( rootblock, 0, 0, &item );
    if( rc < 0 )
    {
        return( -1 );
    }

    /*
     * set the parent pointers to the new block
     */
    newblock->parentblock = 0;
    newblock->parentindex = 0;

    /*
     * split the new block
     */
    return( pblSplit( file, newblock ));
}

/**
 * insert a new record with the given key and data into a key file,
 *
 * multiple records with the same key are allowed,
 * if there are already records with the same key the new
 * record will be inserted behind all records with the same key,
 *
 * the current record of the file will be set to the new record
 *
 * <P>
 * <B>RESTRICTIONS</B>:
 * <BR> - the file must be open for update,
 * <BR> - key must point to the key to be inserted,
 * <BR> - keylen must be bigger than 0 and smaller than 256,
 * <BR> - data must point to the data be inserted,
 * <BR> - datalen must not be negative,
 * <BR> - if datalen == 0, the pointer data is not evaluated at all
 *
 * @return int rc == 0: call went ok
 * @return int rc != 0: some error occured, see pbl_errno
 */

int pblKfInsert(
pblKeyFile_t  * k,      /** key file to insert to                             */
unsigned char * key,    /** key to insert                                     */
int             keylen, /** length of the key                                 */
unsigned char * data,   /** data to insert                                    */
long            datalen /** length of the data                                */
)
{
    PBLKFILE_t *  kf = ( PBLKFILE_t * ) k;
    PBLITEM_t     item;
    PBLITEM_t   * insertitem;
    PBLBLOCK_t  * block;
    long          blockno;
    int           index;
    int           saveerrno;
    int           rc;

    long          parentblock = -1;
    int           parentindex = -1;
    int           split = 0;

    pbl_errno = 0;

    /*
     * start a transaction on the key file
     */
    pblKfStartTransaction( k );
    if( pblBlockListTruncate())
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    if( !kf->update )
    {
        pblKfCommit( k, 1 );
        pbl_errno = PBL_ERROR_NOT_ALLOWED;
        return( -1 );
    }

    rc = pblParamsCheck( key, keylen, data, datalen );
    if( rc )
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    /*
     * only data that is longer than PBLDATALENGTH
     * bytes is written to data blocks
     */
    if( datalen > PBLDATALENGTH )
    {
        /*
         * append the data to the file
         */
        item.datablock = pblDataAppend( kf, data, datalen, &item.dataoffset );
        if( item.datablock < 1 )
        {
            pblKfCommit( k, 1 );
            return( -1 );
        }
        item.data = 0;
    }
    else
    {
        item.datablock = 0;
        item.dataoffset = 0;
        item.data = data;
    }

    /*
     * prepare the data item to be inserted
     */
    item.level      = 0;
    item.keylen     = keylen;
    item.keycommon  = 0;
    item.key        = key;
    item.datalen    = datalen;

    /*
     * push the item to the insert stack of the file
     */
    rc = pblItemSave( &item );
    if( rc )
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    /*
     * insert all items that are on the insert stack
     */
    while( itemListHead )
    {
        insertitem = itemListHead;

        /*
         * we always start the insert at the root block
         */
        blockno = 0;
        parentblock = -1;
        parentindex = -1;

        /*
         * handle all levels of the tree
         */
        while( !pbl_errno )
        {
            block = pblBlockGet( kf, blockno );
            if( !block )
            {
                break;
            }

            /*
             * set the links to the parentblock of the block
             */
            block->parentblock = parentblock;
            block->parentindex = parentindex;

            /*
             * if the item has to be inserted in this level
             */
            if( block->level <= insertitem->level )
            {
                block = pblBlockGetWriteable( kf, blockno );
                if( !block )
                {
                    break;
                }

                index = pblItemAdd( kf, block, insertitem );
                if( index < 0 )
                {
                    if( pbl_errno == PBL_ERROR_NOFIT )
                    {
                        pbl_errno = 0;

                        /*
                         * split the root block or a normal block
                         */
                        if( blockno )
                        {
                            rc = pblSplit( kf, block );
                        }
                        else
                        {
                            rc = pblSplitRoot( kf );
                        }
                        if( !rc )
                        {
                            split = 1;
                        }
                    }
                    break;
                }

                /*
                 * insert was successful
                 */
                if( block->level == 0 )
                {
                    /*
                     * set values of current record
                     */
                    kf->blockno = block->blockno;
                    kf->index   = index;
                }

                /*
                 * release the item that was inserted
                 */
                pblItemRelease( );

                break;
            }

            for(;;)
            {
                block = pblBlockGet( kf, blockno );
                if( !block )
                {
                    break;
                }

                /*
                 * set the links to the parentblock of the block
                 */
                block->parentblock = parentblock;
                block->parentindex = parentindex;

                if( pblBlockKeysExpand( block ))
                {
                    break;
                }

                /*
                 * item has to be inserted on a lower level, find out where
                 *
                 * we either insert into the last subtree, or into the
                 * greatest smaller subtree
                 */
                index = pblItemFind( kf, block, insertitem, PBLLA );
                if( index < 0 )
                {
                    if( pbl_errno == PBL_ERROR_NOT_FOUND )
                    {
                        pbl_errno = 0;
                        index = pblItemFind( kf, block, insertitem, PBLLT );
                    }

                    if( index < 0 )
                    {
                        break;
                    }
                }

                rc = pblItemGet( block, index, &item );
                if( rc )
                {
                    break;
                }

                /*
                 * see if we can merge blocks before the insert
                 */
                if( !split )
                {
                    rc = pblBlockMerge( kf, blockno,
                                        index, item.datablock );
                    if( rc > 0 )
                    {
                        continue;
                    }
                    else if( rc < 0 )
                    {
                        break;
                    }
                }

                /*
                 * get the blockno of the relevant child block
                 */
                blockno = item.datablock;
                parentblock = block->blockno;
                parentindex = index;

                pbl_errno = 0;
                break;
            }
        }

        /*
         * if an error occurred during this insert
         */
        if( pbl_errno )
        {
            break;
        }
    }

    saveerrno = pbl_errno;

    pblItemReleaseAll( );

    if( saveerrno )
    {
        pbl_errno = saveerrno;
        kf->blockno = -1;
        kf->index = -1;
        pblKfCommit( k, 1 );
        return( -1 );
    }

    pblKfCommit( k, 0 );
    return( 0 );
}

/*
 * UPDATE functions
 */
static PBLBLOCK_t * pblPositionCheck( PBLKFILE_t *kf )
{
    PBLBLOCK_t * block;
    int          index;

    /*
     * check if the current block is set for the file
     */
    if( kf->blockno < 0 )
    {
        pbl_errno = PBL_ERROR_POSITION;
        return( 0 );
    }

    /*
     * get the current block to memory
     */
    block = pblBlockGet( kf, kf->blockno );
    if( !block )
    {
        return( 0 );
    }
    index = kf->index;

    /*
     * if we are positioned on our pseudo magic item, we set to next item
     */
    if(( index == 0 ) && ( !block->pblock || !block->blockno ))
    {
        index = 1;
    }

    /*
     * if the index is negative, we actually are set on the last item of
     * the predecessor of the current block, or if the there is no predecessor
     * on the first item of the current block
     */
    while( index < 0 )
    {
        /*
         * if we are on the first block, we need the second item, because the
         * the first item is our pseudo magic item
         */
        if( !block->pblock || !block->blockno )
        {
            index = 1;
        }
        else
        {
            block = pblBlockGet( kf, block->pblock );
            if( !block )
            {
                return( 0 );
            }

            if( block->nentries )
            {
                index = block->nentries - 1;
            }
        }
    }

    while( ( int )index >= ( int )block->nentries )
    {
        /*
         * if there is no successor of the current block, we have to stay here
         * the rootblock never has a successor !
         */
        if( !block->nblock || !block->blockno )
        {
            break;
        }
        else
        {
            block = pblBlockGet( kf, block->nblock );
            if( !block )
            {
                return( 0 );
            }

            if( block->nentries )
            {
                index = 0;
            }
        }
    }

    while( ( int )index >= ( int )block->nentries )
    {
        /*
         * if the block has entries, we take the last one
         */
        if( block->nentries )
        {
            index = block->nentries - 1;
        }
        else if( !block->pblock || !block->blockno )
        {
            /*
             * this is a structure error, because the first block always has
             * at least one item, our pseudo magic item
             */
            pbl_errno = PBL_ERROR_BAD_FILE;
            kf->blockno = -1;
            return( 0 );
        }
        else
        {
            block = pblBlockGet( kf, block->pblock );
            if( !block )
            {
                return( 0 );
            }

            if( block->nentries )
            {
                index = block->nentries - 1;
            }
        }
    }

    /*
     * if we ended up positioning at our pseudo item, the file does not
     * have any other items
     */
    if(( index == 0 ) && ( !block->pblock || !block->blockno ))
    {
        pbl_errno = PBL_ERROR_NOT_FOUND;
        kf->blockno = -1;
        return( 0 );
    }

    kf->blockno = block->blockno;
    kf->index = index;

    return( block );
}

static long pblDataWrite(
PBLKFILE_t * file,
char       * data,
long         blockno,
long         blockoffset,
long         datalen
)
{
    long         diff;
    long         bytesWritten = 0;
    int          nbytes;
    PBLBLOCK_t * datablock = (PBLBLOCK_t *) 0;

    while( bytesWritten < datalen )
    {
        datablock = pblBlockGetWriteable( file, blockno );
        if( !datablock )
        {
            return( -1 );
        }

        if( datablock->level != 255 )
        {
            pbl_errno = PBL_ERROR_BAD_FILE;
            return( -1 );
        }

        diff = datalen - bytesWritten;
        if( diff > ( PBLDATASIZE - blockoffset ) )
        {
            nbytes = PBLDATASIZE - blockoffset;
        }
        else
        {
            nbytes = ((int ) ( diff % PBLDATASIZE));
        }

        if( nbytes > 0 )
        {
            memcpy((void *) &(datablock->data[ blockoffset ]),
                   (void *) data,
                   nbytes );
            datablock->dirty = 1;

            bytesWritten += nbytes;
            data         += nbytes;
        }

        if( bytesWritten < datalen )
        {
            /*
             * get offset of next block and set blockoffset to beginning of
             * data
             */
            blockno     = datablock->nblock;
            blockoffset = PBLHEADERSIZE;
        }
    }

    return( bytesWritten );
}

/**
 * delete the current record of the key file.
 *
 * the current record of the file is set to the next record or
 * if the last record is deleted, to the previous record,
 *
 * if there are no more records in the file after the delete
 * the current record is of course unpositioned
 *
 * <P>
 * <B>RESTRICTIONS</B>:
 * <BR> - the file must be open for update,
 * <BR> - no space will be given back to the file system,
 * <BR> - if an index block and its successor or its predeccessor
 *        together use less than half of a block the two blocks are merged
 *
 * @return int rc == 0: call went ok
 * @return int rc != 0: some error occured, see pbl_errno
 */

int pblKfDelete(
pblKeyFile_t  * k       /** key file to delete record from                    */
)
{
    PBLKFILE_t *    kf = ( PBLKFILE_t * ) k;
    long            parentblock;
    PBLBLOCK_t    * block;
    int             rc;

    pbl_errno = 0;

    /*
     * start a transaction on the key file
     */
    pblKfStartTransaction( k );
    if( pblBlockListTruncate())
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    if( !kf->update )
    {
        pblKfCommit( k, 1 );
        pbl_errno = PBL_ERROR_NOT_ALLOWED;
        return( -1 );
    }

    /*
     * make sure current record of the file is positioned
     */
    block = pblPositionCheck( kf );
    if( !block )
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    /*
     * read the block the current item is on
     */
    block = pblBlockGetWriteable( kf, kf->blockno );
    if( !block )
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    /*
     * delete the current item
     */
    rc = pblItemRemove( block, kf->index );
    if( rc )
    {
        pblKfCommit( k, 1 );
        return( rc );
    }

    /*
     * we deleted an item, now see if we can merge some blocks
     */
    for(;;)
    {
        if( block->parentblock < 0 || block->parentindex < 0 )
        {
            rc = 0;
            break;
        }

        /*
         * remember the blocknumber of the parent block
         */
        parentblock = block->parentblock;

        /*
         * see whether some blocks can be merged because of the delete
         */
        rc = pblBlockMerge( kf, block->parentblock,
                            block->parentindex, block->blockno );
        if( rc < 1 )
        {
            break;
        }

        /*
         * the merge deleted an item on the parent block, read that block
         */
        block = pblBlockGetWriteable( kf, parentblock );
        if( !block )
        {
            pblKfCommit( k, 1 );
            return( -1 );
        }
    }

    if( rc )
    {
        pblKfCommit( k, 1 );
        return( -1 );
    }

    pblKfCommit( k, 0 );
    return( rc );
}

/**
 * update the data of the current record
 *
 * the current record of the file is updated with the new data given
 *
 *
 * <P>
 * <B>RESTRICTIONS</B>:
 * <BR> - the file must be open for update,
 * <BR> - if the new datalen of the record is not bigger than the old datalen,
 *        the data will be updated in place, otherwise the new data of the
 *        record will be appended to the file, the space previously used for 
 *        the data of