/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 *      db.cc
 *
 * Copyright 2004 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include <stdio.h>
#include <string.h>
#ifdef TDRPC
#include <sysent.h>
#else
#include <unistd.h>
#endif

#include "nisdb_mt.h"
#include "db_headers.h"
#include "db.h"

extern db_result *empty_result(db_status);
extern int add_to_standby_list(db*);
extern int remove_from_standby_list(db*);

/* for db_next_desc */

#define LINEAR 1
#define CHAINED 2

struct db_next_info {
        int next_type;          /* linear or chained */
        void* next_value;       /* linear: entryp; */
                                /* chained: db_next_index_desc* */
};


/* Constructor:  Create a database using the given name, 'dbname.'
            The database is stored in a file named 'dbname'.
            The log file is stored in a file named 'dbname'.log.
            A temporary file 'dbname'.tmp is also used.   */
db::db(char* dbname)
{
        int len = strlen(dbname);
        dbfilename = new char[len+1];
        if (dbfilename == NULL)
                FATAL("db::db: cannot allocate space", DB_MEMORY_LIMIT);
        logfilename = new char[len+5];
        if (logfilename == NULL) {
                delete dbfilename;
                FATAL("db::db: cannot allocate space", DB_MEMORY_LIMIT);
        }
        tmpfilename = new char[len+5];
        if (tmpfilename == NULL) {
                delete dbfilename;
                delete logfilename;
                FATAL("db::db: cannot allocate space", DB_MEMORY_LIMIT);
        }
        sprintf(dbfilename, "%s", dbname);
        sprintf(logfilename, "%s.log", dbname);
        sprintf(tmpfilename, "%s.tmp", dbname);
        logfile = NULL;
        logfile_opened = FALSE;
        changed = FALSE;
        INITRW(db);
        READLOCKOK(db);

        internal_db.setDbPtr(this);
        (void) internal_db.configure(dbname);
}

/* destructor:  note that associated files should be removed separated  */
db::~db()
{
        (void)acqexcl();
        internal_db.reset();  /* clear any associated data structures */
        delete dbfilename;
        delete logfilename;
        delete tmpfilename;
        close_log();
        delete logfile;
        (void)destroylock();
}


static void
assign_next_desc(db_next_desc* desc, entryp value)
{
        db_next_info * store = new db_next_info;
        if (store == NULL) {
                desc->db_next_desc_val =  NULL;
                desc->db_next_desc_len = 0;
                FATAL("db::assign_next_desc: cannot allocate space",
                        DB_MEMORY_LIMIT);
        }

        store->next_type = LINEAR;
        store->next_value = (void*)value;
        desc->db_next_desc_val =  (char*) store;
        desc->db_next_desc_len = sizeof (db_next_info);
}

static void
assign_next_desc(db_next_desc* desc, db_next_index_desc * value)
{
        db_next_info * store = new db_next_info;
        if (store == NULL) {
                desc->db_next_desc_val =  NULL;
                desc->db_next_desc_len = 0;
                FATAL("db::assign_next_desc: cannot allocate space (2)",
                        DB_MEMORY_LIMIT);
        }
        store->next_type = CHAINED;
        store->next_value = (void*)value;
        desc->db_next_desc_val =  (char*) store;
        desc->db_next_desc_len = sizeof (db_next_info);
}

static entryp
extract_next_desc(db_next_desc* desc, int *next_type,
                db_next_index_desc** place2)
{
        entryp place;

        if (desc == NULL || desc->db_next_desc_len != sizeof (db_next_info)) {
                *next_type = 0;
                return (0);
        }
        *next_type = ((db_next_info*) desc->db_next_desc_val)->next_type;
        switch (*next_type) {
        case LINEAR:
                place = (entryp)
                        ((db_next_info*) desc->db_next_desc_val)->next_value;
                return (place);

        case CHAINED:
                *place2 = (db_next_index_desc*)
                        ((db_next_info*) desc->db_next_desc_val) ->next_value;
                return (0);
        default:
                *next_type = 0;   // invalid type
                return (0);
        }
}

/* Execute the specified action using the rest of the arguments as input.
            Return  a structure db_result containing the result. */
db_result *
db::exec_action(db_action action, db_query *query,
                entry_object *content, db_next_desc* previous)
{
        entryp where, prev;
        db_result *res = new db_result;
        long num_answers;
        entry_object_p * ans;
        entry_object * single;
        db_next_index_desc *index_desc;
        int next_type;
        db_next_index_desc *prev_desc;

        if (res == NULL)
                FATAL3("db::exec_action: cannot allocate space for result",
                        DB_MEMORY_LIMIT, NULL);

        res->objects.objects_len = 0; /* default */
        res->objects.objects_val = NULL;  /* default */

        switch (action) {
        case DB_LOOKUP:
                res->status = internal_db.lookup(query, &num_answers, &ans);
                res->objects.objects_len = (int) num_answers;
                res->objects.objects_val = ans;
                break;

        case DB_ADD:
                res->status = internal_db.add(query, content);
                break;

        case DB_REMOVE:
                res->status = internal_db.remove(query);
                break;

        case DB_FIRST:
                if (query == NULL) {
                        res->status = internal_db.first(&where, &single);
                        if (res->status == DB_SUCCESS)
                                assign_next_desc(&(res->nextinfo), where);
                }  else {
                        res->status = internal_db.first(query,
                                                        &index_desc,
                                                        &single);
                        if (res->status == DB_SUCCESS)
                                assign_next_desc(&(res->nextinfo), index_desc);
                }
                if (res->status == DB_SUCCESS) {
                        res->objects.objects_val = new entry_object_p;
                        if (res->objects.objects_val == NULL) {
                                res->objects.objects_len = 0;
                                delete res;
                                FATAL3(
                "db::exec_action: cannot allocate space for DB_FIRST result",
                DB_MEMORY_LIMIT, NULL);
                        }
                        res->objects.objects_len = 1;
                        res->objects.objects_val[0] = single;
                }
                break;

        case DB_NEXT:
                prev = extract_next_desc(previous, &next_type, &prev_desc);
                switch (next_type) {
                case LINEAR:
                        if (prev != 0) {
                                res->status = internal_db.next(prev, &where,
                                                                &single);
                                if (res->status == DB_SUCCESS)
                                        assign_next_desc(&(res->nextinfo),
                                                                where);
                        } else
                                        // invalid previous indicator
                                res->status = DB_NOTFOUND;
                        break;
                case CHAINED:
                        if (prev_desc != NULL) {
                                res->status = internal_db.next(prev_desc,
                                                        &index_desc, &single);
                                if (res->status == DB_SUCCESS)
                                        assign_next_desc(&(res->nextinfo),
                                                                index_desc);
                        } else
                                        // invalid previous indicator
                                res->status = DB_NOTFOUND;
                        break;
                default:
                        WARNING("db::exec_action: invalid previous indicator");
                        res->status = DB_BADQUERY;
                }
                if (previous && previous->db_next_desc_val) {
                        delete previous->db_next_desc_val;
                        previous->db_next_desc_len = 0;
                        previous->db_next_desc_val = NULL;
                }
                if (res->status == DB_SUCCESS) {
                        res->objects.objects_len = 1;
                        res->objects.objects_val = new entry_object_p;
                        if (res->objects.objects_val == NULL) {
                                res->objects.objects_len = 0;
                                delete res;
                                FATAL3(
                    "db::exec_action: cannot allocate space for DB_NEXT result",
                    DB_MEMORY_LIMIT, NULL);
                        }
                        res->objects.objects_val[0] = single;
                }
                break;

        case DB_RESET_NEXT:
                prev = extract_next_desc(previous, &next_type, &prev_desc);
                switch (next_type) {
                case LINEAR:
                        res->status = DB_SUCCESS;
                        if (previous->db_next_desc_val) {
        delete previous->db_next_desc_val;
        previous->db_next_desc_len = 0;
        previous->db_next_desc_val = NULL;
                        }
                        break;   // do nothing
                case CHAINED:
                        res->status = internal_db.reset_next(prev_desc);
                        if (previous->db_next_desc_val) {
        delete previous->db_next_desc_val;
        previous->db_next_desc_len = 0;
        previous->db_next_desc_val = NULL;
                        }
                        break;
                default:
                        WARNING("db::exec_action: invalid previous indicator");
                        res->status = DB_BADQUERY;
                }
                break;

        case DB_ALL:
                res->status = internal_db.all(&num_answers, &ans);
                res->objects.objects_len = (int) num_answers;
                res->objects.objects_val = ans;
                break;

        default:
                WARNING("unknown request");
                res->status = DB_BADQUERY;
                return (res);
        }
        return (res);
}

/*
 * Log the given action and execute it.
 * The minor version of the database is updated after the action has
 * been executed and the database is flagged as being changed.
 * Return the structure db_result, or NULL if the logging failed or the
 * action is unknown.
*/
db_result *
db::log_action(db_action action, db_query *query, entry_object *content)
{
        vers *v = internal_db.get_version()->nextminor();
        db_result * res;
        db_log_entry le(action, v, query, content);
        bool_t copylog = FALSE;

        WRITELOCK(this, empty_result(DB_LOCK_ERROR), "w db::log_action");
        /*
         * If this is a synchronous operation on the master we should
         * not copy the log for each operation.  Doing so causes
         * massive disk IO that hampers the performance of these operations.
         * Where as on the replica these operations are not synchronous
         * (batched) and don't affect the performance as much.
         */

        if ((action == DB_ADD_NOSYNC) || (action == DB_REMOVE_NOSYNC))
                copylog = TRUE;

        if (open_log(copylog) < 0)  {
                delete v;
                WRITEUNLOCK(this, empty_result(DB_LOCK_ERROR),
                                "wu db::log_action DB_STORAGE_LIMIT");
                return (empty_result(DB_STORAGE_LIMIT));
        }

        if (logfile->append(&le) < 0) {
                close_log();
                WARNING_M("db::log_action: could not add log entry: ");
                delete v;
                WRITEUNLOCK(this, empty_result(DB_LOCK_ERROR),
                                "wu db::log_action DB_STORAGE_LIMIT");
                return (empty_result(DB_STORAGE_LIMIT));
        }

        switch (action) {
        case DB_ADD_NOSYNC:
                action = DB_ADD;
                break;
        case DB_REMOVE_NOSYNC:
                action = DB_REMOVE;
                break;
        default:
                if (logfile->sync_log() < 0) {
                        close_log();
                        WARNING_M("db::log_action: could not add log entry: ");
                        delete v;
                        WRITEUNLOCK(this, empty_result(DB_LOCK_ERROR),
                                        "wu db::log_action DB_STORAGE_LIMIT");
                        return (empty_result(DB_STORAGE_LIMIT));
                }
                break;
        }
        res = exec_action(action, query, content, NULL);
        internal_db.change_version(v);
        delete v;
        changed = TRUE;
        WRITEUNLOCK(this, empty_result(DB_LOCK_ERROR), "wu db::log_action");

        return (res);
}

/*
 * Execute 'action' using the rest of the arguments as input.
 * Return the result of the operation in a db_result structure;
 * Return NULL if the request is unknown.
 * If the action involves updates (ADD and REMOVE), it is logged first.
 */
db_result *
db::execute(db_action action, db_query *query,
                entry_object *content, db_next_desc* previous)
{
        db_result       *res;

        switch (action) {
        case DB_LOOKUP:
        case DB_FIRST:
        case DB_NEXT:
        case DB_ALL:
        case DB_RESET_NEXT:
                READLOCK(this, empty_result(DB_LOCK_ERROR), "r db::execute");
                res = exec_action(action, query, content, previous);
                READUNLOCK(this, empty_result(DB_LOCK_ERROR),
                                "ru db::execute");
                return (res);

        case DB_ADD_NOLOG:
                WRITELOCK(this, empty_result(DB_LOCK_ERROR), "w db::execute");
                changed = TRUE;
                res = exec_action(DB_ADD, query, content, previous);
                WRITEUNLOCK(this, empty_result(DB_LOCK_ERROR),
                                "wu db::execute");
                return (res);

        case DB_ADD:
        case DB_REMOVE:
        case DB_ADD_NOSYNC:
        case DB_REMOVE_NOSYNC:
                /* log_action() will do the locking */
                return (log_action(action, query, content));

        default:
                WARNING("db::execute: unknown request");
                return (empty_result(DB_INTERNAL_ERROR));
        }
}

/* close existing logfile and delete its structure */
int
db::reset_log()
{
        WRITELOCK(this, -1, "w db::reset_log");
        /* try to close old log file */
        /* doesnot matter since we do synchronous writes only */
        if (logfile != NULL) {
            if (logfile_opened == TRUE) {
                    logfile->sync_log();
                    if (logfile->close() < 0) {
                        WARNING_M("db::reset_log: could not close log file: ");
                    }
                    remove_from_standby_list(this);
            }
            delete logfile;
            logfile = NULL;
        }
        logfile_opened = FALSE;
        WRITEUNLOCK(this, -1, "wu db::reset_log");
        return (0);
}

/* close existing logfile, but leave its structure if exists */
int
db::close_log(int bypass_standby)
{
        WRITELOCK(this, -1, "w db::close_log");
        if (logfile != NULL && logfile_opened == TRUE) {
                logfile->sync_log();
                logfile->close();
                if (!bypass_standby)
                    remove_from_standby_list(this);
        }
        logfile_opened = FALSE;
        WRITEUNLOCK(this, -1, "wu db::close_log");
        return (0);
}

/* open logfile, creating its structure if it does not exist */
int
db::open_log(bool_t copylog)
{
        WRITELOCK(this, -1, "w db::open_log");
        if (logfile == NULL) {
                if ((logfile = new db_log(logfilename, PICKLE_APPEND))
                    == NULL)
                        FATAL3("db::reset_log: cannot allocate space",
                            DB_MEMORY_LIMIT, -1);
        }

        if (logfile_opened == TRUE) {
                WRITEUNLOCK(this, -1, "wu db::open_log");
                return (0);
        }

        logfile->copylog = copylog;

        if ((logfile->open()) == FALSE){
                WARNING_M("db::open_log: could not open log file: ");
                delete logfile;
                logfile = NULL;
                WRITEUNLOCK(this, -1, "wu db::open_log");
                return (-1);
        }
        add_to_standby_list(this);
        logfile_opened = TRUE;
        WRITEUNLOCK(this, -1, "wu db::open_log");
        return (0);
}

/*
 * Execute log entry 'j' on the database identified by 'dbchar' if the
 * version of j is later than that of the database.  If 'j' is executed,
 * 'count' is incremented and the database's verison is updated to that of 'j'.
 * Returns TRUE always for valid log entries; FALSE otherwise.
 */
static bool_t
apply_log_entry(db_log_entry * j, char * dbchar, int *count)
{
        db_mindex * db = (db_mindex *) dbchar;
        bool_t status = TRUE;

        WRITELOCK(db, FALSE, "db::apply_log_entry");

        if (db->get_version()->earlier_than(j->get_version())) {
                ++ *count;
#ifdef DEBUG
                j->print();
#endif /* DEBUG */
                switch (j->get_action()) {
                case DB_ADD:
                case DB_ADD_NOSYNC:
                        db->add(j->get_query(), j->get_object());
                        break;

                case DB_REMOVE:
                case DB_REMOVE_NOSYNC:
                        db->remove(j->get_query());
                        break;

                default:
                        WARNING("db::apply_log_entry: unknown action_type");
                        WRITEUNLOCK(db, FALSE, "db::apply_log_entry");
                        return (FALSE);
                }
                db->change_version(j->get_version());
        }

        WRITEUNLOCK(db, FALSE, "db::apply_log_entry");

        return (TRUE);  /* always want to TRUE if action valid ? */
}

/*
 * Execute log entry 'j' on this db.  'j' is executed if its version is
 * later than that of the database; if executed, the database's version
 * will be changed to that of 'j', regardless of the status of the operation.
 * Returns TRUE if 'j' was executed;   FALSE if it was not.
 * Log entry is added to this database's log if log_entry is applied.
 */
bool_t
db::execute_log_entry(db_log_entry *j)
{
        int count = 0;
        apply_log_entry (j, (char *) &internal_db, &count);
        bool_t copylog = FALSE;
        db_action action;

        /*
         * If this is a synchronous operation on the master we should
         * not copy the log for each operation.  Doing so causes
         * massive disk IO that hampers the performance of these operations.
         * Where as on the replica these operations are not synchronous
         * (batched) and don't affect the performance as much.
         */

        action = j->get_action();
        if ((action == DB_ADD_NOSYNC) || (action == DB_REMOVE_NOSYNC))
                copylog = TRUE;

        /*
         * should really record the log entry first, but can''t do that without
         * knowing whether the log entry is applicable.
         */
        WRITELOCK(this, FALSE, "w db::execute_log_entry");
        if (count == 1) {
                if (open_log(copylog) < 0) {
                        WRITEUNLOCK(this, FALSE, "wu db::execute_log_entry");
                        return (FALSE);
                }

                if (logfile->append(j) < 0) {
                        close_log();
                        WARNING_M(
                        "db::execute_log_entry: could not add log entry: ");
                        WRITEUNLOCK(this, FALSE, "wu db::execute_log_entry");
                        return (FALSE);
                }
//        close_log();  /* do this asynchronously */
        }
        WRITEUNLOCK(this, FALSE, "wu db::execute_log_entry");

        return (count == 1);
}

/* Incorporate updates in log to database already loaded.
            Does not affect "logfile" */
int
db::incorporate_log(char* filename)
{
        db_log f(filename, PICKLE_READ);
        int ret;

        WRITELOCK(this, -1, "w db::incorporate_log");
        WRITELOCK2((&internal_db), -1, "w internal_db db::incorporate_log",
                        this);
        internal_db.setNoWriteThrough();
        ret = f.execute_on_log(&(apply_log_entry), (char *) &internal_db);
        internal_db.clearNoWriteThrough();
        WRITEUNLOCK2(this, (&internal_db), ret, ret,
                        "wu db::incorporate_log",
                        "wu mindex db::incorporate_log");
        return (ret);
}

/* Load database and incorporate any logged updates into the loaded copy.
            Return TRUE if load succeeds; FALSE otherwise. */
bool_t
db::load()
{
        int count;
        int load_status;

        WRITELOCK(this, FALSE, "w db::load");
        if (changed == TRUE)
                syslog(LOG_ERR,
        "WARNING: the current db '%s' has been changed but not checkpointed",
                        dbfilename);

        unlink(tmpfilename);  /* get rid of partial checkpoints */

        if ((load_status = internal_db.load(dbfilename)) != 0) {
            if (load_status < 0)
                    syslog(LOG_ERR, "Load of db '%s' failed", dbfilename);
            /* otherwise, there was just nothing to load */
            WRITEUNLOCK(this, FALSE, "wu db::load");
            return (FALSE);
        }

        changed = FALSE;
        reset_log();
        WRITELOCK2((&internal_db), FALSE, "w internal_db db::load", this);
        internal_db.setInitialLoad();
        if ((count = incorporate_log(logfilename)) < 0)
                syslog(LOG_ERR, "incorporation of db logfile '%s' load failed",
            logfilename);
        changed = (count > 0);
        internal_db.clearInitialLoad();
        WRITEUNLOCK2(this, (&internal_db),
                        (changed ? TRUE : FALSE), (changed ? TRUE : FALSE),
                        "wu db::load", "wu internal_db db::load");
        return (TRUE);
}

/*
 * Initialize the database using table scheme 's'.
 * Because the 'scheme' must be 'remembered' between restarts,
 * after the initialization, the empty database is checkpointed to record
 * the scheme. Returns TRUE if initialization succeeds; FALSE otherwise.
 */
bool_t
db::init(db_scheme * s)
{
        bool_t  ret = FALSE;

        WRITELOCK(this, FALSE, "w db::init");
        internal_db.init(s);
        if (internal_db.good()) {
                unlink(tmpfilename);    /* delete partial checkpoints */
                unlink(logfilename);    /* delete previous logfile */
                reset_log();
                changed = TRUE;         /* force dump to get scheme stored. */
                ret = checkpoint();
        }
        WRITEUNLOCK(this, FALSE, "wu db::init");
        return (ret);
}

/*
    Write out in-memory copy of database to file.
            1.  Update major version.
            2.  Dump contents to temporary file.
            3.  Rename temporary file to real database file.
            4.  Remove log file.
    A checkpoint is done only if it has changed since the previous checkpoint.
    Returns TRUE if checkpoint was successful; FALSE otherwise.
*/
bool_t
db::checkpoint()
{
        WRITELOCK(this, FALSE, "w db::checkpoint");
        if (changed == FALSE) {
                WRITEUNLOCK(this, FALSE, "wu db::checkpoint");
                return (TRUE);
        }

        vers *oldversion = new vers(internal_db.get_version()); /* copy */
        vers *nextversion = oldversion->nextmajor();    /* get next version */
        internal_db.change_version(nextversion);        /* change version */

        if (internal_db.dump(tmpfilename) < 0) {        /* dump to tempfile */
                WARNING_M("db::checkpoint: could not dump database: ");
                internal_db.change_version(oldversion); /* rollback */
                delete nextversion;
                delete oldversion;
                WRITEUNLOCK(this, FALSE, "wu db::checkpoint");
                return (FALSE);
        }
        if (rename(tmpfilename, dbfilename) < 0){       /* rename permanently */
                WARNING_M(
                    "db::checkpoint: could not rename temp file to db file: ");
                internal_db.change_version(oldversion); /* rollback */
                delete nextversion;
                delete oldversion;
                WRITEUNLOCK(this, FALSE, "wu db::checkpoint");
                return (FALSE);
        }
        reset_log();            /* should check for what? */
        unlink(logfilename);    /* should do atomic rename and log delete */
        delete nextversion;
        delete oldversion;
        changed = FALSE;
        WRITEUNLOCK(this, FALSE, "wu db::checkpoint");
        return (TRUE);
}


/* For generating log_list */

struct traverse_info {
        vers *version;          // version to check for
        db_log_entry * head;    // head of list of log entries found
        db_log_entry * tail;    // tail of list of log entries found
};

/*
 * For the given entry determine, if it is later than the version supplied,
 *          1.  increment 'count'.
 *          2.  add the entry to the list of log entries found.
 *
 * Since traversal happens on an automatic (struct traverse_info) in
 * db::get_log_entries_since(), no locking is necessary.
 */
static bool_t entry_since(db_log_entry * j, char * tichar, int *count)
{
        traverse_info *ti = (traverse_info*) tichar;

        if (ti->version->earlier_than(j->get_version())) {
                ++ *count;
//    j->print();   // debug
                if (ti->head == NULL)
                        ti->head = j;
                else {
                        ti->tail->setnextptr(j); // make last entry point to j
                }
                ti->tail = j;                   // make j new last entry
        }

        return (TRUE);
}

/* Return structure db_log_list containing entries that are later
            than the version 'v' given.  */
db_log_list*
db::get_log_entries_since(vers * v)
{
        int count;
        struct traverse_info ti;
        db_log f(logfilename, PICKLE_READ);

        ti.version = v;
        ti.head = ti.tail = NULL;

        count = f.execute_on_log(&(entry_since), (char *) &ti, FALSE);

        db_log_list * answer = new db_log_list;

        if (answer == NULL)
                FATAL3("db::get_log_entries_since: cannot allocate space",
                        DB_MEMORY_LIMIT, NULL);

        answer->list.list_len = count;

        if (count > 0) {
                db_log_entry_p *entries;
                db_log_entry_p currentry, nextentry;
                int i;

                entries = answer->list.list_val = new db_log_entry_p[count];
                if (entries == NULL) {
                        delete answer;
                        FATAL3(
                "db::get_log_entries_since: cannot allocate space for entries",
                DB_MEMORY_LIMIT, NULL);
                        }
                currentry = ti.head;
                for (i = 0, currentry = ti.head;
                        i < count && currentry != NULL;
                        i++) {
                        entries[i] = currentry;
                        nextentry = currentry->getnextptr();
                        currentry->setnextptr(NULL);
                        currentry = nextentry;
                }
        } else
                answer->list.list_val = NULL;

        return (answer);
}

/* Delete all files associated with database. */
int
db::remove_files()
{
        WRITELOCK(this, -1, "w db::remove_files");
        unlink(tmpfilename);  /* delete partial checkpoints */
        reset_log();
        unlink(logfilename);  /* delete logfile */
        unlink(dbfilename);   /* delete database file */
        WRITEUNLOCK(this, -1, "wu db::remove_files");
        return (0);
}

db_status
db::sync_log() {

        db_status       ret;

        WRITELOCK(this, DB_LOCK_ERROR, "w db::sync_log");
        if (logfile == 0) {
                ret = DB_BADTABLE;
        } else {
                if (logfile_opened == FALSE || logfile->sync_log())
                        ret = DB_SUCCESS;
                else
                        ret = DB_SYNC_FAILED;
        }
        WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db::sync_log");
        return (ret);
}

/* Pass configuration information to the db_mindex */
bool_t
db::configure(char *objName) {
        return (internal_db.configure(objName));
}

db_mindex *
db::mindex(void) {
        return (&internal_db);
}