dio.c

db.* (pronounced dee-be star) is an advanced, high performance, small footprint embedded database fo

/***************************************************************************
 *                                                                         *
 * db.*                                                                    *
 * open source database kernel                                             *
 *                                                                         *
 * Copyright (c) 2000 Centura Software Corporation. All rights reserved.   *
 *                                                                         *
 * Use of this software, whether in source code format, or in executable,  *
 * binary object code form, is governed by the CENTURA OPEN SOURCE LICENSE *
 * which is fully described in the LICENSE.TXT file, included within this  *
 * distribution of source code files.                                      * 
 *                                                                         *
 * Except as provided herein, the contents of this file are subject to the *
 * Centura Open Source Public License Version 1.0 (the "License"); you may *
 * not use this file except in compliance with the License.  A copy of the *
 * License will be provided to you by Club ITTIA.                          *
 *                                                                         *
 * Software distributed under the License is distributed on an "AS IS"     *
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the *
 * License for the specific language governing rights and limitations      *
 * under the License.                                                      *
 *                                                                         *
 * The Original Code is db.linux version 1.0, released February 29, 2000.  *
 *                                                                         *
 * The Initial Developer of the Original Code is Centura Software          *
 * Corporation. Portions created by Centura Software Corporation are       *
 * Copyright (C) 1984-2000 Centura Software Corporation. All Rights        *
 * Reserved.                                                               *
 *                                                                         *
 * This file contains modifications to the Original Code made by ITTIA.    *
 * This file may only be used in accordance with the ITTIA DB.* V.2        *
 * License Agreement which is available at WWW.ITTIA.COM.                  *
 *                                                                         *
 **************************************************************************/

#include "db.star.h"

#define MAX_SHORT_PRIME    32749    /* max prime that fits in a short */

static int  INTERNAL_FCN cache_init(PAGE_TABLE *, DB_TASK *);
static void INTERNAL_FCN cache_free(PAGE_TABLE *);
static int  INTERNAL_FCN pgtag_alloc(short *, DB_TASK *);
static void INTERNAL_FCN pgtag_free(short *, DB_TASK *);
static int  INTERNAL_FCN clear_cache(FILE_NO, FILE_NO, DB_TASK *);
static int  INTERNAL_FCN dio_findlru(FILE_NO, F_ADDR, short, PAGE_ENTRY * *, DB_TASK *);
static void INTERNAL_FCN dio_unhash(PAGE_ENTRY *, DB_TASK *);
static void INTERNAL_FCN dio_rehash(PAGE_ENTRY *, DB_TASK *);
static void INTERNAL_FCN dio_pzfree(int, DB_TASK *);
static int  INTERNAL_FCN dio_pzflush(DB_TASK *);
static int  INTERNAL_FCN isprime(long);

#define EXTENDFILES  0x00000010L    /* only data and key files are extended */

#define LU_HASH(f,p,sz)  ((short)(((f)+(p)) % (sz)))

/* ======================================================================
    Set the maximum number of open db.* files
*/
int INTERNAL_FCN dsetfiles(int num, DB_TASK *task)
{
    if (task->dbopen)
        dcloseall(task);

    if (num > 0 && num < (int) (FILEMASK + 1L))
        psp_fileSetHandleLimit(num);
    else
        dberr(S_INVNUM);

    return task->db_status;
}

/* ======================================================================
    Set number of virtual memory pages
*/
int INTERNAL_FCN dsetpages(
    int      dbpgs,             /* # of db cache pages */
    int      ixpgs,             /* # of index cache pages */
    DB_TASK *task)
{
    if ((dbpgs > MAX_SHORT_PRIME) || (ixpgs > MAX_SHORT_PRIME))
    {
        /* we can not allow numbers bigger than this because all the tables
           for the caches are using signed shorts as indices!
        */
        return (dberr(S_INVNUM));
    }

    if (task->cache->db_pgtab.pg_table)
    {
        /* There is already a cache - can't change size */
        return (dberr(S_SETPAGES));
    }

    task->cache->db_pgtab.pgtab_sz = (short) max(dbpgs, MINDBPAGES);
    task->cache->ix_pgtab.pgtab_sz = (short) max(ixpgs, MINIXPAGES);
    
    return (task->db_status);
}

void block_parms(off_t addr, size_t size, off_t *new_addr, size_t *new_size,
                 off_t *off, const SG *sg)
{
    size_t blocks;
    size_t blocksize = sg->blocksize;

    *off = addr % blocksize;
    *new_addr = addr - *off;
    blocks = (size - 1) / blocksize;
    if (*off >= (off_t) (blocksize - size % blocksize + 1))
        blocks++;

    *new_size = blocksize * blocks;
}

/* ======================================================================
    Open a database file
*/
int EXTERNAL_FCN dio_open(FILE_NO fno, DB_TASK *task)
{
    FILE_ENTRY *file_ptr = &task->file_table[fno];

    if (file_ptr->ft_status == OPEN)
        return (task->db_status);

    file_ptr->ft_desc = NULL;
    do
    {
        file_ptr->ft_desc = open_b(file_ptr->ft_name, O_RDWR, PSP_FLAG_SYNC,
                                   task);
        if (file_ptr->ft_desc == NULL)
        {
            int err = psp_errno();
            if (err == EACCES || err == EPIPE)
                return (dberr(S_EACCESS));    /* Sharing violation */

            return (dberr(S_NOFILE));
        }
    } while (file_ptr->ft_desc == NULL);

    file_ptr->ft_status = OPEN;
    ++task->cnt_open_files;

#ifdef DB_DEBUG
    if ((task->db_debug & LOCK_CHECK) && (file_ptr->ft_type != 'o'))
        task->pgzero[fno].file_mtime = psp_fileModTime(file_ptr->ft_desc);
#endif
    if (fno != task->ov_file)
        STAT_file_open(fno, task);
    else
        STAT_log_open(task);
    STAT_max_open(task->cnt_open_files, task);
    
    return (task->db_status);
}


/* ======================================================================
    Close a database file
    returns status, does not set task->db_status
*/
int EXTERNAL_FCN dio_close(FILE_NO fno, DB_TASK *task)
{
    FILE_ENTRY *file_ptr = &task->file_table[fno];

    if (file_ptr->ft_status == CLOSED)
        return (S_OKAY);  

    /* Commit file before closing only if needed */
    if (file_ptr->ft_flags & NEEDS_COMMIT)
    {
        /* Don't want commit_file to call dio_close recursively, so don't
           pass a file number ...
        */
        commit_file(file_ptr->ft_desc, task);
        file_ptr->ft_flags &= ~NEEDS_COMMIT;
    }

    if (file_ptr->ft_desc != NULL)
        psp_fileClose(file_ptr->ft_desc);

    file_ptr->ft_desc = NULL;
    file_ptr->ft_status = CLOSED;
    --task->cnt_open_files;

    return (task->db_status);
}

/* ======================================================================
    Initialize database I/O
*/
int INTERNAL_FCN dio_init(DB_TASK *task)
{
    /* init task on first open */
    if (!task->old_size_ft)
    {
        /* init file handling functions */
        task->last_file = 0;

        /* most recently used pages */
        task->last_datapage = NULL;
        task->last_keypage = NULL;
        task->last_ixpage = NULL;

        if (pgtag_alloc(&task->db_tag, task) != S_OKAY ||
            pgtag_alloc(&task->key_tag, task) != S_OKAY ||
            pgtag_alloc(&task->ix_tag, task) != S_OKAY)
            goto clean_exit;
    }

    if (task->cache->db_pgtab.pg_table)   /* re-init database cache */
    {
        if (dio_pzinit(task) != S_OKAY)
            goto clean_exit;

        if (task->page_size > task->cache->largest_page)
        {
            char *tempbuff;
            
            if (task->cache->prealloc)
            {
                /* must reallocate all cache pages */
                dio_clear(ALL_DBS, task);
                cache_free(&task->cache->db_pgtab);
                if (cache_init(&task->cache->db_pgtab, task) != S_OKAY)
                    return (task->db_status);
            }

            tempbuff = psp_cGetMemory(task->page_size, 0);
            if (tempbuff == NULL)
            {
                /* the cache has already been allocated, do not free it */
                return (dberr(S_NOMEMORY));
            }

            psp_freeMemory(task->cache->dbpgbuff, 0);
            task->cache->dbpgbuff = tempbuff;
            task->cache->largest_page = task->page_size;
        }

        return (S_OKAY);
    }

    /* the first task for a cache sets the rules */
    task->cache->prealloc = (task->dboptions & PREALLOC_CACHE) != 0;

    /* initialize database cache */
    if (cache_init(&task->cache->db_pgtab, task) != S_OKAY ||
        cache_init(&task->cache->ix_pgtab, task) != S_OKAY)
        goto clean_exit;

    /* allocate o_update buffer */
    task->cache->largest_page = task->page_size;
    task->cache->dbpgbuff = psp_cGetMemory(task->cache->largest_page, 0);
    if (task->cache->dbpgbuff == NULL)
    {
        dberr(S_NOMEMORY);
        goto clean_exit;
    }

    /* initialize the page zero table and return */
    dio_pzinit(task);

clean_exit:
    if (task->db_status != S_OKAY)
    {
        int stat = task->db_status;

        dio_free(ALL_DBS, task);
        task->db_status = stat;
    }
    return (task->db_status);
}

/* ======================================================================
*/
static int INTERNAL_FCN cache_init(PAGE_TABLE *pgtab, DB_TASK *task)
{
    short       pg_no;
    PAGE_ENTRY *pg_ptr = pgtab->pg_table;

    /* lookup_sz is prime number >= pgtab_sz */
    pgtab->lookup_sz = pgtab->pgtab_sz;
    pgtab->lookup_sz |= 1;
    while (! isprime(pgtab->lookup_sz))
        pgtab->lookup_sz += 2;

    pgtab->lookup = (LOOKUP_ENTRY *)psp_cGetMemory(pgtab->lookup_sz * sizeof(LOOKUP_ENTRY), 0);
    pgtab->pg_table = (PAGE_ENTRY *)psp_cGetMemory(pgtab->pgtab_sz * sizeof(PAGE_ENTRY), 0);
    if (!pgtab->lookup || !pgtab->pg_table)
        return (dberr(S_NOMEMORY));
    memset(pgtab->pg_table, '\0', pgtab->pgtab_sz * sizeof(PAGE_ENTRY));

    /* this code assumes that the number of pages <= size of lookup pgtab */
    for (pg_no = 0, pg_ptr = pgtab->pg_table;
         pg_no < pgtab->pgtab_sz; ++pg_no, ++pg_ptr)
    {
        pg_ptr->file = -1;         /* page not in use */

        if (task->cache->prealloc)
        {
            /* allocate the cache pages now using the largest page size to make
               sure that there is enough memory.
            */
            pg_ptr->buff = psp_cGetMemory(task->page_size, 0);
            if (!pg_ptr->buff)
                return (dberr(S_NOMEMORY));

            pg_ptr->buff_size = (short) task->page_size;
#ifdef DBSTAT
            STAT_mem_alloc(pgtab, task->page_size);
#endif
        }
    }

    return (task->db_status);
}

/* ======================================================================
    Free the memory allocated for pages
*/
void INTERNAL_FCN dio_free(int dbn, DB_TASK *task)
{
    task->last_datapage = NULL;
    task->last_keypage = NULL;
    task->last_ixpage = NULL;

    dio_pzfree(dbn, task);

    if (dbn != ALL_DBS)
    {
        /* Files being free'd have already had their cache pages cleared,
           so just update the file reference numbers of files being moved
           down in file table.  The ix-cache should be empty at this point.
        */
        int i, high;
        PAGE_ENTRY *pg_ptr;

        high = task->db_table[dbn].ft_offset + task->db_table[dbn].Size_ft - 1;
        for (i = 0, pg_ptr = task->cache->db_pgtab.pg_table; i < task->cache->db_pgtab.pgtab_sz; ++i, ++pg_ptr)
        {
            if (pg_ptr->file > high)
            {
                dio_unhash(pg_ptr, task);
                pg_ptr->file -= task->db_table[dbn].Size_ft;
                dio_rehash(pg_ptr, task);
            }
        }
    }
    else
    {
        pgtag_free(&task->db_tag, task);
        pgtag_free(&task->key_tag, task);
        pgtag_free(&task->ix_tag, task);

#ifdef DBSTAT
        if (task->cache->db_pgtab.pg_table)
        {
            sync_MEM_STATS(&task->gen_stats.dbmem_stats, &task->cache->db_pgtab.mem_stats);
            sync_CACHE_STATS(&task->gen_stats.db_stats, &task->cache->db_pgtab.cache_stats);
        }
        if (task->cache->ix_pgtab.pg_table)
        {
            sync_MEM_STATS(&task->gen_stats.ixmem_stats, &task->cache->ix_pgtab.mem_stats);
            sync_CACHE_STATS(&task->gen_stats.ix_stats, &task->cache->ix_pgtab.cache_stats);
        }
#endif
        cache_free(&task->cache->db_pgtab);
        cache_free(&task->cache->ix_pgtab);

        if (task->cache->dbpgbuff)
            psp_freeMemory(task->cache->dbpgbuff, 0);

        task->cache->dbpgbuff = NULL;
    }
}

/* ======================================================================
*/
static void INTERNAL_FCN cache_free(PAGE_TABLE *pgtab)
{
    int i;
    PAGE_ENTRY *pg_ptr;

    if (pgtab->pg_table)   /* null if dio_init() failed */
    {
        for (i = 0, pg_ptr = pgtab->pg_table; i < pgtab->pgtab_sz; ++i, ++pg_ptr)
        {
            if (pg_ptr->buff)
                psp_freeMemory(pg_ptr->buff, 0);
        }

        psp_freeMemory(pgtab->pg_table, 0);
        pgtab->pg_table = NULL;
    }

    if (pgtab->lookup)
    {
        psp_freeMemory(pgtab->lookup, 0);
        pgtab->lookup = NULL;
    }

#ifdef DBSTAT
    memset(&pgtab->mem_stats,   '\0', sizeof(MEM_STATS));
    memset(&pgtab->cache_stats, '\0', sizeof(CACHE_STATS));
#endif
}

/* ======================================================================
    Allocate a page tag
*/
static int INTERNAL_FCN pgtag_alloc(short *tag, DB_TASK *task)
{
    short tag_idx;
    PAGE_TAG *tag_ptr;

    if (!task->cache->tag_table)
        task->cache->num_tags = task->cache->max_tags = 0;

    for (tag_idx = 0, tag_ptr = task->cache->tag_table; tag_idx < task->cache->max_tags; ++tag_idx, ++tag_ptr)
    {
        if (tag_ptr->lru_page == -1)
            break;
    }
    *tag = tag_idx;

    if (tag_idx >= task->cache->max_tags)