lib/libnisdb/db_dictionary.cc

	db_dictionary.cc revision 8d0852b7f8176b2c281cc944010af2b0fecf9fb2
/*
 * 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_dictionary.cc
 *
 * Copyright 2004 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#include "db_headers.h"
#include "db_entry.h"
#include "db_dictionary.h"
#include "db_dictlog.h"
#include "db_vers.h"
#include "nisdb_mt.h"
#include "nisdb_rw.h"
#include "ldap_parse.h"
#include "ldap_map.h"
#include "nis_hashitem.h"
#include "ldap_util.h"
#include "nis_db.h"
#include <rpcsvc/nis.h>

#include <stdio.h>
#include <string.h>
#include <malloc.h>
#ifdef TDRPC
#include <sysent.h>
#endif
#include <unistd.h>
#include <syslog.h>
#include <rpc/rpc.h>

typedef bool_t  (*db_func)(XDR *, db_table_desc *);

extern db_dictionary *InUseDictionary;
extern db_dictionary *FreeDictionary;

/* *************** dictionary version ****************** */

#define DB_MAGIC 0x12340000
#define DB_MAJOR 0
#define DB_MINOR 10
#define DB_VERSION_0_9  (DB_MAGIC|(DB_MAJOR<<8)|9)
#define DB_ORIG_VERSION DB_VERSION_0_9
#define DB_CURRENT_VERSION (DB_MAGIC|DB_MAJOR<<8|DB_MINOR)

vers db_update_version;   /* Note 'global' for all dbs. */

#define INMEMORY_ONLY 1

/*
 * Checks for valid version.  For now, there are two:
 * DB_VERSION_ORIG was the ORIGINAL one with major = 0, minor = 9
 * DB_CURRENT_VERSION is the latest one with changes in the database format
 *  for entry objects and the change in the dictionary format.
 *
 * Our current implementation can support both versions.
 */
static inline bool_t
db_valid_version(u_int vers)
{
    return ((vers == DB_CURRENT_VERSION) || (vers == DB_ORIG_VERSION));
}

static char *
db_version_str(u_int vers)
{
    static char vstr[128];
    u_int d_major =  (vers&0x0000ff00)>>8;
    u_int d_minor =  (vers&0x000000ff);

    sprintf(vstr, "SunSoft, SSM, Version %d.%d", d_major, d_minor);
    return (vstr);
}

/*
 * Special XDR version that checks for a valid version number.
 * If we don't find an acceptable version, exit immediately instead
 * of continuing to xdr rest of dictionary, which might lead to
 * a core dump if the formats between versions are incompatible.
 * In the future, there might be a switch to determine versions
 * and their corresponding XDR routines for the rest of the dictionary.
 */
extern "C" {
bool_t
xdr_db_dict_version(XDR *xdrs, db_dict_version *objp)
{
    if (xdrs->x_op == XDR_DECODE) {
        if (!xdr_u_int(xdrs, (u_int*) objp) ||
            !db_valid_version(((u_int) *objp))) {
            syslog(LOG_ERR,
    "db_dictionary: invalid dictionary format! Expecting %s",
                db_version_str(DB_CURRENT_VERSION));
            fprintf(stderr,
    "db_dictionary: invalid dictionary format! Expecting %s\n",
                db_version_str(DB_CURRENT_VERSION));
            exit(1);
        }
    } else if (!xdr_u_int(xdrs, (u_int*) objp))
        return (FALSE);
    return (TRUE);
}

void
make_zero(vers* v)
{
    v->zero();
}


};


/* ******************* dictionary data structures *************** */

/* Delete contents of single db_table_desc pointed to by 'current.' */
static void
delete_table_desc(db_table_desc *current)
{
    if (current->table_name != NULL) delete current->table_name;
    if (current->scheme != NULL) delete current->scheme;
    if (current->database != NULL) delete current->database;
    delete current;
}

/* Create new table descriptor using given table name and table_object. */
db_status
db_dictionary::create_table_desc(char *tab, table_obj* zdesc,
                db_table_desc** answer)
{
    db_table_desc *newtab;
    if ((newtab = new db_table_desc) == NULL) {
        FATAL3(
        "db_dictionary::add_table: could not allocate space for new table",
        DB_MEMORY_LIMIT, DB_MEMORY_LIMIT);
    }

    newtab->database = NULL;
    newtab->table_name = NULL;
    newtab->next = NULL;

    if ((newtab->scheme = new db_scheme(zdesc)) == NULL) {
        delete_table_desc(newtab);
        FATAL3(
    "db_dictionary::add_table: could not allocate space for scheme",
        DB_MEMORY_LIMIT, DB_MEMORY_LIMIT);
    }

    if (newtab->scheme->numkeys() == 0) {
        WARNING(
    "db_dictionary::add_table: could not translate table_obj to scheme");
        delete_table_desc(newtab);
        return (DB_BADOBJECT);
    }

    if ((newtab->table_name = strdup(tab)) == NULL) {
        delete_table_desc(newtab);
        FATAL3(
        "db_dictionary::add_table: could not allocate space for table name",
        DB_MEMORY_LIMIT, DB_MEMORY_LIMIT);
    }

    if (answer)
        *answer = newtab;
    return (DB_SUCCESS);
}


/* Delete list of db_table_desc pointed to by 'head.' */
static void
delete_bucket(db_table_desc *head)
{
    db_table_desc * nextone, *current;

    for (current = head; current != NULL; current = nextone) {
        nextone = current->next;    // remember next
        delete_table_desc(current);
    }
}

static void
delete_dictionary(db_dict_desc *dict)
{
    db_table_desc* bucket;
    int i;
    if (dict) {
        if (dict->tables.tables_val) {
            /* delete each bucket */
            for (i = 0; i < dict->tables.tables_len; i++)
                bucket = dict->tables.tables_val[i];
                if (bucket)
                    delete_bucket(bucket);
            /* delete table */
            delete dict->tables.tables_val;
        }
        /* delete dictionary */
        delete dict;
    }
}

/* Relocate bucket starting with this entry to new hashtable 'new_tab'. */
static void
relocate_bucket(db_table_desc* bucket,
        db_table_desc_p *new_tab, unsigned long hashsize)
{
    db_table_desc_p np, next_np, *hp;

    for (np = bucket; np != NULL; np = next_np) {
        next_np = np->next;
        hp = &new_tab[np->hashval % hashsize];
        np->next = *hp;
        *hp = np;
    }
}

/*
 * Return pointer to entry with same hash value and table_name
 * as those supplied.  Returns NULL if not found.
 */
static db_status
enumerate_bucket(db_table_desc* bucket, db_status(*func)(db_table_desc *))
{
    db_table_desc_p np;
    db_status status;

    for (np = bucket; np != NULL; np = np->next) {
        status = (func)(np);
        if (status != DB_SUCCESS)
            return (status);
    }
    return (DB_SUCCESS);
}


/*
 * Return pointer to entry with same hash value and table_name
 * as those supplied.  Returns NULL if not found.
 */
static db_table_desc_p
search_bucket(db_table_desc* bucket, unsigned long hval, char *target)
{
    db_table_desc_p np;

    for (np = bucket; np != NULL; np = np->next) {
        if (np->hashval == hval &&
            strcmp(np->table_name, target) == 0) {
            break;
        }
    }
    return (np);
}


/*
 * Remove entry with the specified hashvalue and target table name.
 * Returns 'TRUE' if successful, FALSE otherwise.
 * If the entry being removed is at the head of the list, then
 * the head is updated to reflect the removal. The storage for the
 * entry is freed if desired.
 */
static bool_t
remove_from_bucket(db_table_desc_p bucket,
        db_table_desc_p *head, unsigned long hval, char *target,
        bool_t free_storage)
{
    db_table_desc_p np, dp;

    /* Search for it in the bucket */
    for (dp = np = bucket; np != NULL; np = np->next) {
        if (np->hashval == hval &&
            strcmp(np->table_name, target) == 0) {
            break;
        } else {
            dp = np;
        }
    }

    if (np == NULL)
        return (FALSE); // cannot delete if it is not there

    if (dp == np) {
        *head = np->next;   // deleting head of bucket
    } else {
        dp->next = np->next;    // deleting interior link
    }
    if (free_storage)
        delete_table_desc(np);

    return (TRUE);
}


/*
 * Add given entry to the bucket pointed to by 'bucket'.
 * If an entry with the same table_name is found, no addition
 * is done.  The entry is added to the head of the bucket.
 */
static bool_t
add_to_bucket(db_table_desc_p bucket, db_table_desc **head, db_table_desc_p td)
{
    db_table_desc_p curr, prev;
    register char *target_name;
    unsigned long target_hval;
    target_name = td->table_name;
    target_hval = td->hashval;

    /* Search for it in the bucket */
    for (prev = curr = bucket; curr != NULL; curr = curr->next) {
        if (curr->hashval == target_hval &&
            strcmp(curr->table_name, target_name) == 0) {
            break;
        } else {
            prev = curr;
        }
    }

    if (curr != NULL)
        return (FALSE);  /* duplicates not allowed */

    curr = *head;
    *head = td;
    td->next = curr;
    return (TRUE);
}


/* Print bucket starting with this entry. */
static void
print_bucket(db_table_desc *head)
{
    db_table_desc *np;
    for (np = head; np != NULL; np = np->next) {
        printf("%s: %d\n", np->table_name, np->hashval);
    }
}

static db_status
print_table(db_table_desc *tbl)
{
    if (tbl == NULL)
        return (DB_BADTABLE);
    printf("%s: %d\n", tbl->table_name, tbl->hashval);
    return (DB_SUCCESS);
}


static int hashsizes[] = {      /* hashtable sizes */
    11,
    53,
    113,
    337,
    977,
    2053,
    4073,
    8011,
    16001,
    0
};

// prevents wrap around numbers from being passed
#define CALLOC_LIMIT 536870911

/* Returns the next size to use for the hash table */
static unsigned int
get_next_hashsize(long unsigned oldsize)
{
    long unsigned newsize, n;
    if (oldsize == 0)
        newsize = hashsizes[0];
    else {
        for (n = 0; newsize = hashsizes[n++]; )
            if (oldsize == newsize) {
                newsize = hashsizes[n]; /* get next size */
                break;
            }
        if (newsize == 0)
            newsize = oldsize * 2 + 1;  /* just double */
    }
    return (newsize);
}

/*
 * Grow the current hashtable upto the next size.
 * The contents of the existing hashtable is copied to the new one and
 * relocated according to its hashvalue relative to the new size.
 * Old table is deleted after the relocation.
 */
static void
grow_dictionary(db_dict_desc_p dd)
{
    unsigned int oldsize, i, new_size;
    db_table_desc_p * oldtab, *newtab;

    oldsize = dd->tables.tables_len;
    oldtab = dd->tables.tables_val;

    new_size = get_next_hashsize(oldsize);

    if (new_size > CALLOC_LIMIT) {
        FATAL("db_dictionary::grow: table size exceeds calloc limit",
            DB_MEMORY_LIMIT);
    }

    if ((newtab = (db_table_desc_p*)
        calloc((unsigned int) new_size,
            sizeof (db_table_desc_p))) == NULL) {
        FATAL("db_dictionary::grow: cannot allocate space",
            DB_MEMORY_LIMIT);
    }

    if (oldtab != NULL) {       // must transfer contents of old to new
        for (i = 0; i < oldsize; i++) {
            relocate_bucket(oldtab[i], newtab, new_size);
        }
        delete oldtab;      // delete old hashtable
    }

    dd->tables.tables_val = newtab;
    dd->tables.tables_len = new_size;
}

#define HASHSHIFT   3
#define HASHMASK    0x1f

static u_int
get_hashval(char *value)
{
    int i, len;
    u_int hval = 0;

    len = strlen(value);
    for (i = 0; i < len; i++) {
        hval = ((hval<<HASHSHIFT)^hval);
        hval += (value[i] & HASHMASK);
    }

    return (hval);
}

static db_status
enumerate_dictionary(db_dict_desc *dd, db_status (*func) (db_table_desc*))
{
    int i;
    db_table_desc *bucket;
    db_status status;

    if (dd == NULL)
        return (DB_SUCCESS);

    for (i = 0; i < dd->tables.tables_len; i++) {
        bucket = dd->tables.tables_val[i];
        if (bucket) {
            status = enumerate_bucket(bucket, func);
            if (status != DB_SUCCESS)
                return (status);
        }
    }

    return (DB_SUCCESS);
}


/*
 * Look up target table_name in hashtable and return its db_table_desc.
 * Return NULL if not found.
 */
static db_table_desc *
search_dictionary(db_dict_desc *dd, char *target)
{
    register unsigned long hval;
    unsigned long bucket;

    if (target == NULL || dd == NULL || dd->tables.tables_len == 0)
        return (NULL);

    hval = get_hashval(target);
    bucket = hval % dd->tables.tables_len;

    db_table_desc_p fst = dd->tables.tables_val[bucket];

    if (fst != NULL)
        return (search_bucket(fst, hval, target));
    else
        return (NULL);
}

/*
 * Remove the entry with the target table_name from the dictionary.
 * If successful, return DB_SUCCESS; otherwise DB_NOTUNIQUE if target
 * is null; DB_NOTFOUND if entry is not found.
 * If successful, decrement count of number of entries in hash table.
 */
static db_status
remove_from_dictionary(db_dict_desc *dd, char *target, bool_t remove_storage)
{
    register unsigned long hval;
    unsigned long bucket;
    register db_table_desc *fst;

    if (target == NULL)
        return (DB_NOTUNIQUE);
    if (dd == NULL || dd->tables.tables_len == 0)
        return (DB_NOTFOUND);
    hval = get_hashval(target);
    bucket = hval % dd->tables.tables_len;
    fst = dd->tables.tables_val[bucket];
    if (fst == NULL)
        return (DB_NOTFOUND);
    if (remove_from_bucket(fst, &dd->tables.tables_val[bucket],
            hval, target, remove_storage)) {
        --(dd->count);
        return (DB_SUCCESS);
    } else
        return (DB_NOTFOUND);
}

/*
 * Add a new db_table_desc to the dictionary.
 * Return DB_NOTUNIQUE, if entry with identical table_name
 * already exists.  If entry is added, return DB_SUCCESS.
 * Increment count of number of entries in index table and grow table
 * if number of entries equals size of table.
 *
 * Inputs: db_dict_desc_p dd    pointer to dictionary to add to.
 *     db_table_desc *td    pointer to table entry to be added. The
 *              db_table_desc.next field will be altered
 *              without regard to it's current setting.
 *              This means that if next points to a list of
 *              table entries, they may be either linked into
 *              the dictionary unexpectly or cut off (leaked).
 */
static db_status
add_to_dictionary(db_dict_desc_p dd, db_table_desc *td)
{
    register unsigned long hval;
    char *target;

    if (dd == NULL)
        return (DB_NOTFOUND);

    if (td == NULL)
        return (DB_NOTFOUND);
    target = td->table_name;
    if (target == NULL)
        return (DB_NOTUNIQUE);

    hval = get_hashval(target);

    if (dd->tables.tables_val == NULL)
        grow_dictionary(dd);

    db_table_desc_p fst;
    unsigned long bucket;
    bucket = hval % dd->tables.tables_len;
    fst = dd->tables.tables_val[bucket];
    td->hashval = hval;
    if (fst == NULL)  { /* Empty bucket */
        dd->tables.tables_val[bucket] = td;
    } else if (!add_to_bucket(fst, &dd->tables.tables_val[bucket], td)) {
            return (DB_NOTUNIQUE);
        }

    /* increase hash table size if number of entries equals table size */
    if (++(dd->count) > dd->tables.tables_len)
        grow_dictionary(dd);

    return (DB_SUCCESS);
}

/* ******************* pickling routines for dictionary ******************* */


/* Does the actual writing to/from file specific for db_dict_desc structure. */
static bool_t
transfer_aux(XDR* x, pptr tbl)
{
    return (xdr_db_dict_desc_p(x, (db_dict_desc_p *) tbl));
}

class pickle_dict_desc: public pickle_file {
    public:
    pickle_dict_desc(char *f, pickle_mode m) : pickle_file(f, m) {}

    /* Transfers db_dict_desc structure pointed to by dp to/from file. */
    int transfer(db_dict_desc_p * dp)
        { return (pickle_file::transfer((pptr) dp, &transfer_aux)); }
};

/* ************************ dictionary methods *************************** */

db_dictionary::db_dictionary()
{
    dictionary = NULL;
    initialized = FALSE;
    filename = NULL;
    tmpfilename = NULL;
    logfilename = NULL;
    logfile = NULL;
    logfile_opened = FALSE;
    changed = FALSE;
    INITRW(dict);
    READLOCKOK(dict);
}

/*
 * This routine clones an entire hash bucket chain. If you clone a
 * data dictionary entry with the ->next pointer set, you will get a
 * clone of that entry, as well as the entire linked list. This can cause
 * pain if you then pass the cloned bucket to routines such as
 * add_to_dictionary(), which do not expect nor handle dictionary hash
 * entries with the ->next pointer set. You might either get duplicate
 * entires or lose entries. If you wish to clone the entire bucket chain
 * and add it to a new dictionary, loop through the db_table_desc->next list
 * and call add_to_dictionary() for each item.
 */
int
db_dictionary::db_clone_bucket(db_table_desc *bucket, db_table_desc **clone)
{
    u_long      size;
    XDR     xdrs;
    char        *bufin = NULL;

    READLOCK(this, DB_LOCK_ERROR, "r db_dictionary::db_clone_bucket");
    db_func use_this = xdr_db_table_desc;
    size = xdr_sizeof((xdrproc_t) use_this, (void *) bucket);
    bufin = (char *) calloc(1, (size_t) size * sizeof (char));
    if (!bufin) {
        READUNLOCK(this, DB_MEMORY_LIMIT,
            "db_dictionary::insert_modified_table: out of memory");
        FATAL3("db_dictionary::insert_modified_table: out of memory",
            DB_MEMORY_LIMIT, 0);
    }
    xdrmem_create(&xdrs, bufin, (size_t) size, XDR_ENCODE);
    if (!xdr_db_table_desc(&xdrs, bucket)) {
        free(bufin);
        xdr_destroy(&xdrs);
        READUNLOCK(this, DB_MEMORY_LIMIT,
        "db_dictionary::insert_modified_table: xdr encode failed");
        FATAL3(
        "db_dictionary::insert_modified_table: xdr encode failed.",
        DB_MEMORY_LIMIT, 0);
    }
    *clone = (db_table_desc *) calloc(1, (size_t) size * sizeof (char));
    if (!*clone) {
        xdr_destroy(&xdrs);
        free(bufin);
        READUNLOCK(this, DB_MEMORY_LIMIT,
            "db_dictionary::insert_modified_table: out of memory");
        FATAL3("db_dictionary::insert_modified_table: out of memory",
            DB_MEMORY_LIMIT, 0);
    }

    xdrmem_create(&xdrs, bufin, (size_t) size, XDR_DECODE);
    if (!xdr_db_table_desc(&xdrs, *clone)) {
        free(bufin);
        free(*clone);
        xdr_destroy(&xdrs);
        READUNLOCK(this, DB_MEMORY_LIMIT,
        "db_dictionary::insert_modified_table: xdr encode failed");
        FATAL3(
        "db_dictionary::insert_modified_table: xdr encode failed.",
        DB_MEMORY_LIMIT, 0);
    }
    free(bufin);
    xdr_destroy(&xdrs);
    READUNLOCK(this, DB_LOCK_ERROR, "ru db_dictionary::db_clone_bucket");
    return (1);
}


int
db_dictionary::change_table_name(db_table_desc *clone, char *tok, char *repl)
{
    char    *newname;
    char    *loc_end, *loc_beg;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::change_table_name");
    while (clone) {
        /*
         * Special case for a tok="". This is used for the
         * nisrestore(1M), when restoring a replica in another
         * domain. This routine is used to change the datafile
         * names in the data.dict (see bugid #4031273). This will not
         * effect massage_dict(), since it never generates an empty
         * string for tok.
         */
        if (strlen(tok) == 0) {
            strcat(clone->table_name, repl);
            clone = clone->next;
            continue;
        }
        newname = (char *) calloc(1, sizeof (char) *
                strlen(clone->table_name) +
                strlen(repl) - strlen(tok) + 1);
        if (!newname) {
            WRITEUNLOCK(this, DB_MEMORY_LIMIT,
            "db_dictionary::change_table_name: out of memory");
            FATAL3("db_dictionary::change_table_name: out of memory.",
                DB_MEMORY_LIMIT, 0);
        }
        if (loc_beg = strstr(clone->table_name, tok)) {
            loc_end = loc_beg + strlen(tok);
            int s = loc_beg - clone->table_name;
            memcpy(newname, clone->table_name, s);
            strcat(newname + s, repl);
            strcat(newname, loc_end);
            free(clone->table_name);
            clone->table_name = newname;
        } else {
            free(newname);
        }
        clone = clone->next;
    }
    WRITEUNLOCK(this, DB_LOCK_ERROR,
            "wu db_dictionary::change_table_name");
    return (1);
}


#ifdef  curdict
#undef  curdict
#endif
/*
 * A function to initialize the temporary dictionary from the real
 * dictionary.
 */
bool_t
db_dictionary::inittemp(char *dictname, db_dictionary& curdict)
{
    int status;
    db_table_desc_p *newtab;

    db_shutdown();

    WRITELOCK(this, FALSE, "w db_dictionary::inittemp");
    if (initialized) {
        /* Someone else got in between db_shutdown() and lock() */
        WRITEUNLOCK(this, FALSE, "wu db_dictionary::inittemp");
        return (TRUE);
    }

    pickle_dict_desc f(dictname, PICKLE_READ);
    filename = strdup(dictname);
    if (filename == NULL) {
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::inittemp: could not allocate space");
        FATAL3("db_dictionary::inittemp: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }
    int len = strlen(filename);
    tmpfilename = new char[len+5];
    if (tmpfilename == NULL) {
        delete filename;
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::inittemp: could not allocate space");
        FATAL3("db_dictionary::inittemp: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }
    logfilename = new char[len+5];
    if (logfilename == NULL) {
        delete filename;
        delete tmpfilename;
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::inittemp: cannot allocate space");
        FATAL3("db_dictionary::inittemp: cannot allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }

    sprintf(tmpfilename, "%s.tmp", filename);
    sprintf(logfilename, "%s.log", filename);
    unlink(tmpfilename);  /* get rid of partial checkpoints */
    dictionary = NULL;

    if ((status = f.transfer(&dictionary)) < 0) {
        initialized = FALSE;
    } else if (status == 1) { /* no dictionary exists, create one */
        dictionary = new db_dict_desc;
        if (dictionary == NULL) {
            WRITEUNLOCK(this, FALSE,
            "db_dictionary::inittemp: could not allocate space");
            FATAL3(
            "db_dictionary::inittemp: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
        }
        dictionary->tables.tables_len =
                curdict.dictionary->tables.tables_len;
        if ((newtab = (db_table_desc_p *) calloc(
            (unsigned int) dictionary->tables.tables_len,
            sizeof (db_table_desc_p))) == NULL) {
            WRITEUNLOCK(this, FALSE,
            "db_dictionary::inittemp: cannot allocate space");
            FATAL3(
            "db_dictionary::inittemp: cannot allocate space",
            DB_MEMORY_LIMIT, 0);
        }
        dictionary->tables.tables_val = newtab;
        dictionary->count = 0;
        dictionary->impl_vers = curdict.dictionary->impl_vers;
        initialized = TRUE;
    } else  /* dictionary loaded successfully */
        initialized = TRUE;

    if (initialized == TRUE) {
        changed = FALSE;
        reset_log();
    }

    WRITEUNLOCK(this, FALSE, "wu db_dictionary::inittemp");
    return (initialized);
}


/*
 * This method replaces the token string specified with the replacment
 * string specified. It assumes that at least one and only one instance of
 * the token exists. It is the responsibility of the caller to ensure that
 * the above assumption stays valid.
 */
db_status
db_dictionary::massage_dict(char *newdictname, char *tok, char *repl)
{
    int     retval;
    u_int       i, tbl_count;
    db_status   status;
    db_table_desc   *bucket, *np, *clone, *next_np;
    char        tail[NIS_MAXNAMELEN];
    db_dictionary   *tmpptr;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::massage_dict");
    if (dictionary == NULL) {
        WRITEUNLOCK(this, DB_INTERNAL_ERROR,
        "db_dictionary::massage_dict: uninitialized dictionary file");
        FATAL3(
        "db_dictionary::massage_dict: uninitialized dictionary file.",
        DB_INTERNAL_ERROR, DB_INTERNAL_ERROR);
    }

    if ((tbl_count = dictionary->count) == 0) {
        WRITEUNLOCK(this, DB_SUCCESS,
                "wu db_dictionary::massage_dict");
        return (DB_SUCCESS);
    }

    /* First checkpoint */
    if ((status = checkpoint()) != DB_SUCCESS) {
        WRITEUNLOCK(this, status, "wu db_dictionary::massage_dict");
        return (status);
    }

#ifdef DEBUG
    enumerate_dictionary(dictionary, &print_table);
#endif

    /* Initialize the free dictionary so that we can start populating it */
    FreeDictionary->inittemp(newdictname, *this);

    for (i = 0; i < dictionary->tables.tables_len; i++) {
        bucket = dictionary->tables.tables_val[i];
        if (bucket) {
            np = bucket;
            while (np != NULL) {
                next_np = np->next;
                retval = db_clone_bucket(np, &clone);
                if (retval != 1) {
                    WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::massage_dict");
                    return (DB_INTERNAL_ERROR);
                }
                if (change_table_name(clone, tok, repl) == -1) {
                    delete_table_desc(clone);
                    WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::massage_dict");
                    return (DB_INTERNAL_ERROR);
                }
                /*
                 * We know we don't have a log file, so we will
                 * just add to the in-memory database and dump
                 * all of it once we are done.
                 */
                status = add_to_dictionary
                        (FreeDictionary->dictionary,
                        clone);
                if (status != DB_SUCCESS) {
                    delete_table_desc(clone);
                    WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::massage_dict");
                    return (DB_INTERNAL_ERROR);
                }
                status = remove_from_dictionary(dictionary,
                            np->table_name, TRUE);
                if (status != DB_SUCCESS) {
                    delete_table_desc(clone);
                    WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::massage_dict");
                    return (DB_INTERNAL_ERROR);
                }
                np = next_np;
            }
        }
    }

    if (FreeDictionary->dump() != DB_SUCCESS) {
        WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                "wu db_dictionary::massage_dict");
        FATAL3(
        "db_dictionary::massage_dict: Unable to dump new dictionary.",
        DB_INTERNAL_ERROR, DB_INTERNAL_ERROR);
    }

    /*
     * Now, shutdown the inuse dictionary and update the FreeDictionary
     * and InUseDictionary pointers as well. Also, delete the old dictionary
     * file.
     */
    unlink(filename); /* There shouldn't be a tmpfile or logfile */
    db_shutdown();
    tmpptr = InUseDictionary;
    InUseDictionary = FreeDictionary;
    FreeDictionary = tmpptr;
    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::massage_dict");
    return (DB_SUCCESS);
}


db_status
db_dictionary::merge_dict(db_dictionary& tempdict, char *tok, char *repl)
{

    db_status   dbstat = DB_SUCCESS;

    db_table_desc   *tbl = NULL, *clone = NULL, *next_td = NULL;
    int     retval, i;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::merge_dict");

    for (i = 0; i < tempdict.dictionary->tables.tables_len; ++i) {
        tbl = tempdict.dictionary->tables.tables_val[i];
        if (!tbl)
            continue;
        retval = db_clone_bucket(tbl, &clone);
        if (retval != 1) {
            WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::merge_dict");
            return (DB_INTERNAL_ERROR);
        }
        while (clone) {
            next_td = clone->next;
            clone->next = NULL;
            if ((tok) &&
                (change_table_name(clone, tok, repl) == -1)) {
                delete_table_desc(clone);
                if (next_td)
                    delete_table_desc(next_td);
                WRITEUNLOCK(this, DB_INTERNAL_ERROR,
                    "wu db_dictionary::merge_dict");
                return (DB_INTERNAL_ERROR);
            }

            dbstat = add_to_dictionary(dictionary, clone);
            if (dbstat == DB_NOTUNIQUE) {
                /* Overide */
                dbstat = remove_from_dictionary(dictionary,
                        clone->table_name, TRUE);
                if (dbstat != DB_SUCCESS) {
                    WRITEUNLOCK(this, dbstat,
                    "wu db_dictionary::merge_dict");
                    return (dbstat);
                }
                dbstat = add_to_dictionary(dictionary,
                                clone);
            } else {
                if (dbstat != DB_SUCCESS) {
                    WRITEUNLOCK(this, dbstat,
                    "wu db_dictionary::merge_dict");
                    return (dbstat);
                }
            }
            clone = next_td;
        }
    }
/*
 * If we were successful in merging the dictionaries, then mark the
 * dictionary changed, so that it will be properly checkpointed and
 * dumped to disk.
 */
    if (dbstat == DB_SUCCESS)
        changed = TRUE;
    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::merge_dict");
    return (dbstat);
}

int
db_dictionary::copyfile(char *infile, char *outfile)
{
    db_table_desc   *tbl = NULL;
    db  *dbase;
    int ret;

    READLOCK(this, DB_LOCK_ERROR, "r db_dictionary::copyfile");
    /*
     * We need to hold the read-lock until the dump() is done.
     * However, we must avoid the lock migration (read -> write)
     * that would happen in find_table() if the db must be loaded.
     * Hence, look first look for an already loaded db.
     */
    dbase  = find_table(infile, &tbl, TRUE, TRUE, FALSE);
    if (dbase == NULL) {
        /* Release the read-lock, and try again, allowing load */
        READUNLOCK(this, DB_LOCK_ERROR, "ru db_dictionary::copyfile");
        dbase  = find_table(infile, &tbl, TRUE, TRUE, TRUE);
        if (dbase == NULL)
            return (DB_NOTFOUND);
        /*
         * Read-lock again, and get a 'tbl' we can use since we're
         * still holding the lock.
         */
        READLOCK(this, DB_LOCK_ERROR, "r db_dictionary::copyfile");
        dbase  = find_table(infile, &tbl, TRUE, TRUE, FALSE);
        if (dbase == NULL) {
            READUNLOCK(this, DB_NOTFOUND,
                    "ru db_dictionary::copyfile");
            return (DB_NOTFOUND);
        }
    }
    ret = tbl->database->dump(outfile) ? DB_SUCCESS : DB_INTERNAL_ERROR;
    READUNLOCK(this, ret, "ru db_dictionary::copyfile");
    return (ret);
}


bool_t
db_dictionary::extract_entries(db_dictionary& tempdict, char **fs, int fscnt)
{
    int     i, retval;
    db_table_desc   *tbl, *clone;
    db_table_desc   tbl_ent;
    db_status   dbstat;

    READLOCK(this, FALSE, "r db_dictionary::extract_entries");
    for (i = 0; i < fscnt; ++i) {
        tbl = find_table_desc(fs[i]);
        if (!tbl) {
            syslog(LOG_DEBUG,
                "extract_entries: no dictionary entry for %s",
                fs[i]);
            READUNLOCK(this, FALSE,
                    "ru db_dictionary::extract_entries");
            return (FALSE);
        } else {
            tbl_ent.table_name = tbl->table_name;
            tbl_ent.hashval = tbl->hashval;
            tbl_ent.scheme = tbl->scheme;
            tbl_ent.database = tbl->database;
            tbl_ent.next = NULL;
        }
        retval = db_clone_bucket(&tbl_ent, &clone);
        if (retval != 1) {
            syslog(LOG_DEBUG,
            "extract_entries: unable to clone entry for %s",
            fs[i]);
            READUNLOCK(this, FALSE,
                    "ru db_dictionary::extract_entries");
            return (FALSE);
        }
        dbstat = add_to_dictionary(tempdict.dictionary, clone);
        if (dbstat != DB_SUCCESS) {
            delete_table_desc(clone);
            READUNLOCK(this, FALSE,
                    "ru db_dictionary::extract_entries");
            return (FALSE);
        }
    }
    if (tempdict.dump() != DB_SUCCESS) {
        READUNLOCK(this, FALSE,
                "ru db_dictionary::extract_entries");
        return (FALSE);
    }
    READUNLOCK(this, FALSE,
            "ru db_dictionary::extract_entries");
    return (TRUE);
}


/*
 * Initialize dictionary from contents in 'file'.
 * If there is already information in this dictionary, it is removed.
 * Therefore, regardless of whether the load from the file succeeds,
 * the contents of this dictionary will be altered.  Returns
 * whether table has been initialized successfully.
 */
bool_t
db_dictionary::init(char *file)
{
    int status;

    WRITELOCK(this, FALSE, "w db_dictionary::init");
    db_shutdown();

    pickle_dict_desc f(file, PICKLE_READ);
    filename = strdup(file);
    if (filename == NULL) {
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::init: could not allocate space");
        FATAL3("db_dictionary::init: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }
    int len = strlen(filename);
    tmpfilename = new char[len+5];
    if (tmpfilename == NULL) {
        delete filename;
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::init: could not allocate space");
        FATAL3("db_dictionary::init: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }
    logfilename = new char[len+5];
    if (logfilename == NULL) {
        delete filename;
        delete tmpfilename;
        WRITEUNLOCK(this, FALSE,
                "db_dictionary::init: cannot allocate space");
        FATAL3("db_dictionary::init: cannot allocate space",
            DB_MEMORY_LIMIT, FALSE);
    }

    sprintf(tmpfilename, "%s.tmp", filename);
    sprintf(logfilename, "%s.log", filename);
    unlink(tmpfilename);  /* get rid of partial checkpoints */
    dictionary = NULL;

    /* load dictionary */
    if ((status = f.transfer(&dictionary)) < 0) {
        initialized = FALSE;
    } else if (status == 1) {  /* no dictionary exists, create one */
        dictionary = new db_dict_desc;
        if (dictionary == NULL) {
        WRITEUNLOCK(this, FALSE,
            "db_dictionary::init: could not allocate space");
        FATAL3("db_dictionary::init: could not allocate space",
            DB_MEMORY_LIMIT, FALSE);
        }
        dictionary->tables.tables_len = 0;
        dictionary->tables.tables_val = NULL;
        dictionary->count = 0;
        dictionary->impl_vers = DB_CURRENT_VERSION;
        initialized = TRUE;
    } else  /* dictionary loaded successfully */
        initialized = TRUE;

    if (initialized == TRUE) {
        int num_changes = 0;
        changed = FALSE;
        reset_log();
        if ((num_changes = incorporate_log(logfilename)) < 0)
        syslog(LOG_ERR,
            "incorporation of dictionary logfile '%s' failed",
            logfilename);
        changed = (num_changes > 0);
    }

    WRITEUNLOCK(this, initialized, "wu db_dictionary::init");
    return (initialized);
}

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

    WRITELOCK(dict, FALSE, "w apply_log_entry");
    if (db_update_version.earlier_than(j->get_version())) {
        ++ *count;
#ifdef DEBUG
        j->print();
#endif /* DEBUG */
        switch (j->get_action()) {
        case DB_ADD_TABLE:
            dict->add_table_aux(j->get_table_name(),
                j->get_table_object(), INMEMORY_ONLY);
            // ignore status
            break;

        case DB_REMOVE_TABLE:
            dict->delete_table_aux(j->get_table_name(),
                            INMEMORY_ONLY);
            // ignore status
            break;

        default:
            WARNING("db::apply_log_entry: unknown action_type");
            WRITEUNLOCK(dict, FALSE, "wu apply_log_entry");
            return (FALSE);
        }
        db_update_version.assign(j->get_version());
    }

    WRITEUNLOCK(dict, TRUE, "wu apply_log_entry");
    return (TRUE);
}

int
db_dictionary::incorporate_log(char *file_name)
{
    db_dictlog f(file_name, PICKLE_READ);
    int ret;

    WRITELOCK(this, -1, "w db_dictionary::incorporate_log");
    setNoWriteThrough();
    ret = f.execute_on_log(&(apply_log_entry), (char *) this);
    clearNoWriteThrough();
    WRITEUNLOCK(this, -1, "wu db_dictionary::incorporate_log");
    return (ret);
}


/* Frees memory of filename and tables.  Has no effect on disk storage. */
db_status
db_dictionary::db_shutdown()
{
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::db_shutdown");
    if (!initialized) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::db_shutdown");
        return (DB_SUCCESS); /* DB_NOTFOUND? */
    }

    if (filename) {
        delete filename;
        filename = NULL;
    }
    if (tmpfilename) {
        delete tmpfilename;
        tmpfilename = NULL;
    }
    if (logfilename) {
        delete logfilename;
        logfilename = NULL;
    }
    if (dictionary) {
        delete_dictionary(dictionary);
        dictionary = NULL;
    }
    initialized = FALSE;
    changed = FALSE;
    reset_log();

    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::db_shutdown");
    return (DB_SUCCESS);
}

/*
 * Dump contents of this dictionary (minus the database representations)
 * to its file. Returns 0 if operation succeeds, -1 otherwise.
 */
int
db_dictionary::dump()
{
    int status;

    READLOCK(this, -1, "r db_dictionary::dump");
    if (!initialized) {
        READUNLOCK(this, -1, "ru db_dictionary::dump");
        return (-1);
    }

    unlink(tmpfilename);  /* get rid of partial dumps */
    pickle_dict_desc f(tmpfilename, PICKLE_WRITE);

    status = f.transfer(&dictionary);   /* dump table descs */
    if (status != 0) {
        WARNING("db_dictionary::dump: could not write out dictionary");
    } else if (rename(tmpfilename, filename) < 0) {
        WARNING_M("db_dictionary::dump: could not rename temp file: ");
        status = -1;
    }

    READUNLOCK(this, -1, "ru db_dictionary::dump");
    return (status);
}

/*
 * Write out in-memory copy of dictionary to file.
 * 1.  Update major version.
 * 2.  Dump contents to temporary file.
 * 3.  Rename temporary file to real dictionary file.
 * 4.  Remove log file.
 * A checkpoint is done only if it has changed since the previous checkpoint.
 * Returns DB_SUCCESS if checkpoint was successful; error code otherwise
 */
db_status
db_dictionary::checkpoint()
{
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::checkpoint");

    if (changed == FALSE) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::checkpoint");
        return (DB_SUCCESS);
    }

    vers *oldv = new vers(db_update_version);   // copy
    vers * newv = db_update_version.nextmajor();    // get next version
    db_update_version.assign(newv);         // update version
    delete newv;

    if (dump() != 0) {
        WARNING_M(
            "db_dictionary::checkpoint: could not dump dictionary: ");
        db_update_version.assign(oldv);  // rollback
        delete oldv;
        WRITEUNLOCK(this, DB_INTERNAL_ERROR,
            "wu db_dictionary::checkpoint");
        return (DB_INTERNAL_ERROR);
    }
    unlink(logfilename);    /* should do atomic rename and log delete */
    reset_log();        /* should check for what? */
    delete oldv;
    changed = FALSE;
    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::checkpoint");
    return (DB_SUCCESS);
}

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

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

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

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

    if ((logfile->open()) == NULL) {
        WARNING_M("db_dictionary::open_log: could not open log file: ");
        delete logfile;
        logfile = NULL;
        WRITEUNLOCK(this, -1, "wu db_dictionary::open_log");
        return (-1);
    }

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

/*
 * closes any open log files for all tables in dictionary or 'tab'.
 * "tab" is an optional argument.
 */
static int close_standby_list();

db_status
db_dictionary::db_standby(char *tab)
{
    db_table_desc *tbl;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::db_standby");
    if (!initialized) {
        WRITEUNLOCK(this, DB_BADDICTIONARY,
                "wu db_dictionary::db_standby");
        return (DB_BADDICTIONARY);
    }

    if (tab == NULL) {
        close_log();  // close dictionary log
        close_standby_list();
        WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::db_standby");
        return (DB_SUCCESS);
    }

    if ((tbl = find_table_desc(tab)) == NULL) {
        WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::db_standby");
        return (DB_BADTABLE);
    }

    if (tbl->database != NULL)
        tbl->database->close_log();
    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::db_standby");
    return (DB_SUCCESS);
}

/*
 * Returns db_table_desc of table name 'tab'.  'prev', if supplied,
 * is set to the entry located ahead of 'tab's entry in the dictionary.
 */
db_table_desc*
db_dictionary::find_table_desc(char *tab)
{
    db_table_desc   *ret;

    READLOCK(this, NULL, "r db_dictionary::find_table_desc");
    if (initialized)
        ret = search_dictionary(dictionary, tab);
    else
        ret = NULL;

    READUNLOCK(this, ret, "r db_dictionary::find_table_desc");
    return (ret);
}

db_table_desc *
db_dictionary::find_table_desc(char *tab, bool_t searchDeferred) {
    db_table_desc   *ret = NULL;

    READLOCK(this, NULL, "r db_dictionary::find_table_desc_d");

    /* If desired, look in the deferred dictionary first */
    if (initialized && searchDeferred && deferred.dictionary != NULL)
        ret = search_dictionary(deferred.dictionary, tab);

    /* No result yet => search the "normal" dictionary */
    if (ret == NULL)
        ret = find_table_desc(tab);

    READUNLOCK(this, ret, "r db_dictionary::find_table_desc_d");
    return (ret);
}

db *
db_dictionary::find_table(char *tab, db_table_desc **where) {
    /* Most operations should use the deferred dictionary if it exists */
    return (find_table(tab, where, TRUE, TRUE, TRUE));
}

db *
db_dictionary::find_table(char *tab, db_table_desc **where,
                bool_t searchDeferred) {
    return (find_table(tab, where, searchDeferred, TRUE, TRUE));
}

db *
db_dictionary::find_table(char *tab, db_table_desc **where,
                bool_t searchDeferred, bool_t doLDAP,
                bool_t doLoad) {
    db          *res;
    int         lstat;
    db_status       dstat;
    const char      *myself = "db_dictionary::find_table";

    res = find_table_noLDAP(tab, where, searchDeferred, doLoad);
    /* If found, or shouldn't try LDAP, we're done */
    if (res != 0 || !doLDAP)
        return (res);

    /* See if we can retrieve the object from LDAP */
    dstat = dbCreateFromLDAP(tab, &lstat);
    if (dstat != DB_SUCCESS) {
        if (dstat == DB_NOTFOUND) {
            if (lstat != LDAP_SUCCESS) {
                logmsg(MSG_NOTIMECHECK, LOG_INFO,
                    "%s: LDAP error for \"%s\": %s",
                    myself, NIL(tab),
                    ldap_err2string(lstat));
            }
        } else {
            logmsg(MSG_NOTIMECHECK, LOG_INFO,
                "%s: DB error %d for \"%s\"",
                myself, dstat, NIL(tab));
        }
        return (0);
    }

    /* Try the dictionary again */
    res = find_table_noLDAP(tab, where, searchDeferred, doLoad);

    return (res);
}

/*
 * Return database structure of table named by 'tab'.
 * If 'where' is set, set it to the table_desc of 'tab.'
 * If the database is loaded in from stable store if it has not been loaded.
 * If it cannot be loaded, it is initialized using the scheme stored in
 * the table_desc.  NULL is returned if the initialization fails.
 */
db *
db_dictionary::find_table_noLDAP(char *tab, db_table_desc **where,
                bool_t searchDeferred, bool_t doLoad)
{
    if (!initialized)
        return (NULL);

    db_table_desc* tbl;
    db *dbase = NULL;
    int     lret;

    READLOCK(this, NULL, "r db_dictionary::find_table");
    tbl = find_table_desc(tab, searchDeferred);
    if (tbl == NULL) {
        READUNLOCK(this, NULL, "ru db_dictionary::find_table");
        return (NULL);      // not found
    }

    if (tbl->database != NULL || !doLoad) {
        if (tbl->database && where) *where = tbl;
        READUNLOCK(this, NULL, "ru db_dictionary::find_table");
        return (tbl->database);  // return handle
    }

    READUNLOCK(this, NULL, "ru db_dictionary::find_table");
    WRITELOCK(this, NULL, "w db_dictionary::find_table");
    /* Re-check; some other thread might have loaded the db */
    if (tbl->database != NULL) {
        if (where) *where = tbl;
        WRITEUNLOCK(this, NULL, "wu db_dictionary::find_table");
        return (tbl->database);  // return handle
    }

    // need to load in/init database
    dbase = new db(tab);

    if (dbase == NULL) {
        WRITEUNLOCK(this, NULL,
            "db_dictionary::find_table: could not allocate space");
        FATAL3("db_dictionary::find_table: could not allocate space",
            DB_MEMORY_LIMIT, NULL);
    }

    /*
     * Lock the newly created 'dbase', so we can release the general
     * db_dictionary lock.
     */
    WRITELOCKNR(dbase, lret, "w dbase db_dictionary::find_table");
    if (lret != 0) {
        WRITEUNLOCK(this, NULL,
            "db_dictionary::find_table: could not lock dbase");
        FATAL3("db_dictionary::find_table: could not lock dbase",
            DB_LOCK_ERROR, NULL);
    }
    /* Assign tbl->database, and then release the 'this' lock */
    tbl->database = dbase;
    WRITEUNLOCK(this, NULL, "wu db_dictionary::find_table");

    if (dbase->load()) {            // try to load in database
        if (where) *where = tbl;
        WRITEUNLOCK(dbase, dbase, "wu dbase db_dictionary::find_table");
        return (dbase);
    }

    delete dbase;
    tbl->database = NULL;
    WARNING("db_dictionary::find_table: could not load database");
    return (NULL);
}

/* Log action to be taken on the  dictionary and update db_update_version. */

db_status
db_dictionary::log_action(int action, char *tab, table_obj *tobj)
{
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::log_action");

    vers *newv = db_update_version.nextminor();
    db_dictlog_entry le(action, newv, tab, tobj);

    if (open_log() < 0) {
        delete newv;
        WRITEUNLOCK(this, DB_STORAGE_LIMIT,
                "wu db_dictionary::log_action");
        return (DB_STORAGE_LIMIT);
    }

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

    db_update_version.assign(newv);
    delete newv;
    changed = TRUE;

    WRITEUNLOCK(this, DB_LOCK_ERROR, "wu db_dictionary::log_action");
    return (DB_SUCCESS);
}

// For a complete 'delete' operation, we want the following behaviour:
// 1. If there is an entry in the log, the physical table exists and is
//    stable.
// 2. If there is no entry in the log, the physical table may or may not
//    exist.

db_status
db_dictionary::delete_table_aux(char *tab, int mode)
{
    db_status   ret;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::delete_table_aux");
    if (!initialized) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::delete_table_aux");
        return (DB_BADDICTIONARY);
    }

    db_table_desc *tbl;
    if ((tbl = find_table_desc(tab)) == NULL) { // table not found
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::delete_table_aux");
        return (DB_NOTFOUND);
    }

    if (mode != INMEMORY_ONLY) {
        int need_free = 0;

        // Update log.
        db_status status = log_action(DB_REMOVE_TABLE, tab);
        if (status != DB_SUCCESS) {
            WRITEUNLOCK(this, status,
                "wu db_dictionary::delete_table_aux");
            return (status);
        }

        // Remove physical structures
        db *dbase = tbl->database;
        if (dbase == NULL) {    // need to get desc to access files
            dbase = new db(tab);
            need_free = 1;
        }
        if (dbase == NULL) {
            WARNING(
        "db_dictionary::delete_table: could not create db structure");
            WRITEUNLOCK(this, DB_MEMORY_LIMIT,
                    "wu db_dictionary::delete_table_aux");
            return (DB_MEMORY_LIMIT);
        }
        dbase->remove_files();  // remove physical files
        if (need_free)
            delete dbase;
    }

    // Remove in-memory structures
    ret = remove_from_dictionary(dictionary, tab, TRUE);
    WRITEUNLOCK(this, ret, "wu db_dictionary::delete_table_aux");
    return (ret);
}

/*
 * Delete table with given name 'tab' from dictionary.
 * Returns error code if table does not exist or if dictionary has not been
 * initialized.   Dictionary is updated to stable store if deletion is
 * successful.  Fatal error occurs if dictionary cannot be saved.
 * Returns DB_SUCCESS if dictionary has been updated successfully.
 * Note that the files associated with the table are also removed.
 */
db_status
db_dictionary::delete_table(char *tab)
{
    return (delete_table_aux(tab, !INMEMORY_ONLY));
}

// For a complete 'add' operation, we want the following behaviour:
// 1. If there is an entry in the log, then the physical table exists and
//    has been initialized properly.
// 2. If there is no entry in the log, the physical table may or may not
//    exist.  In this case, we don't really care because we cannot get at
//    it.  The next time we add a table with the same name to the dictionary,
//    it will be initialized properly.
// This mode is used when the table is first created.
//
// For an INMEMORY_ONLY operation, only the internal structure is created and
// updated.  This mode is used when the database gets loaded and the internal
// dictionary gets updated from the log entries.

db_status
db_dictionary::add_table_aux(char *tab, table_obj* tobj, int mode)
{
    db_status   ret;

    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::add_table_aux");
    if (!initialized) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::add_table_aux");
        return (DB_BADDICTIONARY);
    }

    if (find_table_desc(tab) != NULL) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::add_table_aux");
        return (DB_NOTUNIQUE);      // table already exists
    }

    // create data structures for table
    db_table_desc *new_table = 0;
    db_status status = create_table_desc(tab, tobj, &new_table);

    if (status != DB_SUCCESS) {
        WRITEUNLOCK(this, DB_LOCK_ERROR,
                "wu db_dictionary::add_table_aux");
        return (status);
    }

    if (mode != INMEMORY_ONLY) {
        // create physical structures for table
        new_table->database = new db(tab);
        if (new_table->database == NULL) {
            delete_table_desc(new_table);
            WRITEUNLOCK(this, DB_MEMORY_LIMIT,
        "db_dictionary::add_table: could not allocate space for db");
            FATAL3(
            "db_dictionary::add_table: could not allocate space for db",
            DB_MEMORY_LIMIT, DB_MEMORY_LIMIT);
        }
        if (new_table->database->init(new_table->scheme) == 0) {
            WARNING(
    "db_dictionary::add_table: could not initialize database from scheme");
            new_table->database->remove_files();
            delete_table_desc(new_table);
            WRITEUNLOCK(this, DB_STORAGE_LIMIT,
                "wu db_dictionary::add_table_aux");
            return (DB_STORAGE_LIMIT);
        }

        // update 'external' copy of dictionary
        status = log_action(DB_ADD_TABLE, tab, tobj);

        if (status != DB_SUCCESS) {
            new_table->database->remove_files();
            delete_table_desc(new_table);
            WRITEUNLOCK(this, status,
                    "wu db_dictionary::add_table_aux");
            return (status);
        }
    }

    // finally, update in-memory copy of dictionary
    ret = add_to_dictionary(dictionary, new_table);
    WRITEUNLOCK(this, ret, "wu db_dictionary::add_table_aux");
    return (ret);
}

/*
 * Add table with given name 'tab' and description 'zdesc' to dictionary.
 * Returns errror code if table already exists, or if no memory can be found
 * to store the descriptor, or if dictionary has not been intialized.
 * Dictionary is updated to stable store if addition is successful.
 * Fatal error occurs if dictionary cannot be saved.
 * Returns DB_SUCCESS if dictionary has been updated successfully.
*/
db_status
db_dictionary::add_table(char *tab, table_obj* tobj)
{
    return (add_table_aux(tab, tobj, !INMEMORY_ONLY));
}

/*
 * Translate given NIS attribute list to a db_query structure.
 * Return FALSE if dictionary has not been initialized, or
 * table does not have a scheme (which should be a fatal error?).
 */
db_query*
db_dictionary::translate_to_query(db_table_desc* tbl, int numattrs,
                nis_attr* attrlist)
{
    READLOCK(this, NULL, "r db_dictionary::translate_to_query");
    if (!initialized ||
        tbl->scheme == NULL || numattrs == 0 || attrlist == NULL) {
        READUNLOCK(this, NULL, "ru db_dictionary::translate_to_query");
        return (NULL);
    }

    db_query *q = new db_query(tbl->scheme, numattrs, attrlist);
    if (q == NULL) {
        READUNLOCK(this, NULL,
            "db_dictionary::translate: could not allocate space");
        FATAL3("db_dictionary::translate: could not allocate space",
            DB_MEMORY_LIMIT, NULL);
    }

    if (q->size() == 0) {
        delete q;
        READUNLOCK(this, NULL, "ru db_dictionary::translate_to_query");
        return (NULL);
    }
    READUNLOCK(this, NULL, "ru db_dictionary::translate_to_query");
    return (q);
}

static db_table_names gt_answer;
static int gt_posn;

static db_status
get_table_name(db_table_desc* tbl)
{
    if (tbl)
        return (DB_BADTABLE);

    if (gt_posn < gt_answer.db_table_names_len)
        gt_answer.db_table_names_val[gt_posn++] =
            strdup(tbl->table_name);
    else
        return (DB_BADTABLE);

    return (DB_SUCCESS);
}


/*
 * Return the names of tables in this dictionary.
 * XXX This routine is used only for testing only;
 *  if to be used for real, need to free memory sensibly, or
 *  caller of get_table_names should have freed them.
 */
db_table_names*
db_dictionary::get_table_names()
{
    READLOCK(this, NULL, "r db_dictionary::get_table_names");
    gt_answer.db_table_names_len = dictionary->count;
    gt_answer.db_table_names_val = new db_table_namep[dictionary->count];
    gt_posn = 0;
    if ((gt_answer.db_table_names_val) == NULL) {
        READUNLOCK(this, NULL,
    "db_dictionary::get_table_names: could not allocate space for names");
        FATAL3(
    "db_dictionary::get_table_names: could not allocate space for names",
        DB_MEMORY_LIMIT, NULL);
    }

    enumerate_dictionary(dictionary, &get_table_name);
    READUNLOCK(this, NULL, "ru db_dictionary::get_table_names");
    return (&gt_answer);
}

static db_status
db_checkpoint_aux(db_table_desc *current)
{
    db *dbase;
    int status;

    if (current == NULL)
        return (DB_BADTABLE);

    if (current->database == NULL) {  /* need to load it in */
        dbase = new db(current->table_name);
        if (dbase == NULL) {
            FATAL3(
            "db_dictionary::db_checkpoint: could not allocate space",
            DB_MEMORY_LIMIT, DB_MEMORY_LIMIT);
        }
        if (dbase->load() == 0) {
            syslog(LOG_ERR,
            "db_dictionary::db_checkpoint: could not load table %s",
                            current->table_name);
            delete dbase;
            return (DB_BADTABLE);
        }
        status = dbase->checkpoint();
        delete dbase;  // unload
    } else
        status = current->database->checkpoint();

    if (status == 0)
        return (DB_STORAGE_LIMIT);
    return (DB_SUCCESS);
}

/* Like db_checkpoint_aux except only stops on LIMIT errors */
static db_status
db_checkpoint_aux_cont(db_table_desc *current)
{
    db_status status = db_checkpoint_aux(current);

    if (status == DB_STORAGE_LIMIT || status == DB_MEMORY_LIMIT)
        return (status);
    else
        return (DB_SUCCESS);
}

db_status
db_dictionary::db_checkpoint(char *tab)
{
    db_table_desc *tbl;
    db_status   ret;
    bool_t      init;

    READLOCK(this, DB_LOCK_ERROR, "r db_dictionary::db_checkpoint");
    init = initialized;
    READUNLOCK(this, DB_LOCK_ERROR, "ru db_dictionary::db_checkpoint");
    if (!init)
        return (DB_BADDICTIONARY);

    checkpoint();   // checkpoint dictionary first

    READLOCK(this, DB_LOCK_ERROR, "r db_dictionary::db_checkpoint");

    if (tab == NULL) {
        ret = enumerate_dictionary(dictionary, &db_checkpoint_aux_cont);
        READUNLOCK(this, ret, "ru db_dictionary::db_checkpoint");
        return (ret);
    }

    if ((tbl = find_table_desc(tab)) == NULL) {
        READUNLOCK(this, DB_LOCK_ERROR,
                "ru db_dictionary::db_checkpoint");
        return (DB_BADTABLE);
    }

    ret = db_checkpoint_aux(tbl);
    READUNLOCK(this, ret, "ru db_dictionary::db_checkpoint");
    return (ret);
}

/* *********************** db_standby **************************** */
/* Deal with list of tables that need to be 'closed' */

#define OPENED_DBS_CHUNK    12
static db   **db_standby_list;
static uint_t   db_standby_size = 0;
static uint_t   db_standby_count = 0;
DECLMUTEXLOCK(db_standby_list);

/*
 * Returns 1 if all databases on the list could be closed, 0
 * otherwise.
 */
static int
close_standby_list()
{
    db      *database;
    int     i, ret;
    const char  *myself = "close_standby_list";

    MUTEXLOCK(db_standby_list, "close_standby_list");

    if (db_standby_count == 0) {
        MUTEXUNLOCK(db_standby_list, "close_standby_list");
        return (1);
    }

    for (i = 0, ret = 0; i < db_standby_size; i++) {
        if ((database = db_standby_list[i])) {
            /*
             * In order to avoid a potential dead-lock, we
             * check to see if close_log() would be able to
             * lock the db; if not, just skip the db.
             */
            int lockok;

            TRYWRITELOCK(database, lockok,
                "try w db_dictionary::close_standby_list");

            if (lockok == 0) {
                database->close_log(1);
                db_standby_list[i] = (db*)NULL;
                --db_standby_count;
                WRITEUNLOCK(database, db_standby_count == 0,
                    "db_dictionary::close_standby_list");
                if (db_standby_count == 0) {
                    ret = 1;
                    break;
                }
            } else if (lockok != EBUSY) {
                logmsg(MSG_NOTIMECHECK, LOG_INFO,
                    "%s: try-lock error %d",
                    myself, lockok);
            } /* else it's EBUSY; skip to the next one */
        }
    }

    MUTEXUNLOCK(db_standby_list, "close_standby_list");

    return (ret);
}

/*
 * Add given database to list of databases that have been opened for updates.
 * If size of list exceeds maximum, close opened databases first.
 */

int
add_to_standby_list(db* database)
{
    int     i;
    const char  *myself = "add_to_standby_list";

    MUTEXLOCK(db_standby_list, "add_to_standby_list");

    if (database == 0) {
        MUTEXUNLOCK(db_standby_list, "add_to_standby_list");
        return (1);
    }

    /* Try to keep the list below OPENED_DBS_CHUNK */
    if (db_standby_count >= OPENED_DBS_CHUNK) {
        MUTEXUNLOCK(db_standby_list, "add_to_standby_list");
        close_standby_list();
        MUTEXLOCK(db_standby_list, "add_to_standby_list");
    }

    if (db_standby_count >= db_standby_size) {
        db  **ndsl = (db **)realloc(db_standby_list,
                    (db_standby_size+OPENED_DBS_CHUNK) *
                        sizeof (ndsl[0]));

        if (ndsl == 0) {
            MUTEXUNLOCK(db_standby_list, "add_to_standby_list");
            logmsg(MSG_NOMEM, LOG_ERR,
                "%s: realloc(%d) => NULL",
                myself, (db_standby_size+OPENED_DBS_CHUNK) *
                sizeof (ndsl[0]));
            return (0);
        }

        db_standby_list = ndsl;

        for (i = db_standby_size; i < db_standby_size+OPENED_DBS_CHUNK;
                i++)
            db_standby_list[i] = 0;

        db_standby_size += OPENED_DBS_CHUNK;
    }

    for (i = 0; i < db_standby_size; i++) {
        if (db_standby_list[i] == (db*)NULL) {
            db_standby_list[i] = database;
            ++db_standby_count;
            MUTEXUNLOCK(db_standby_list, "add_to_standby_list");
            return (1);
        }
    }

    MUTEXUNLOCK(db_standby_list, "add_to_standby_list");

    return (0);
}

int
remove_from_standby_list(db* database)
{
    int i;

    MUTEXLOCK(db_standby_list, "remove_from_standby_list");

    if (database == 0) {
        MUTEXUNLOCK(db_standby_list, "remove_from_standby_list");
        return (1);
    }

    for (i = 0; i < db_standby_size; i++) {
        if ((database == db_standby_list[i])) {
            db_standby_list[i] = (db*)NULL;
            --db_standby_count;
            MUTEXUNLOCK(db_standby_list,
                    "remove_from_standby_list");
            return (1);
        }
    }

    MUTEXUNLOCK(db_standby_list, "remove_from_standby_list");

    return (0);
}

/* Release space for copied dictionary */
static void
db_release_dictionary(db_dict_desc_p d) {

    int i;

    if (d != NULL) {
        for (i = 0; i < d->tables.tables_len; i++) {
            db_table_desc_p n, t = d->tables.tables_val[i];
            while (t != NULL) {
                n = t->next;
                delete_table_desc(t);
                t = n;
            }
        }
        delete d;
    }

    return;
}

/*
 * Make a copy of the dictionary
 */
db_dict_desc_p
db_dictionary::db_copy_dictionary(void) {

    db_dict_desc_p  tmp;
    int     i, ok = 1, count = 0;

    WRITELOCK(this, NULL, "db_dictionary::db_copy_dictionary w");

    if (dictionary == NULL) {
        WRITEUNLOCK(this, NULL,
            "db_dictionary::db_copy_dictionary wu");
        return (NULL);
    }

    tmp = new db_dict_desc;
    if (tmp == NULL) {
        WRITEUNLOCK(this, NULL,
            "db_dictionary::db_copy_dictionary wu: no memory");
        return (NULL);
    }

    tmp->tables.tables_val = (db_table_desc_p *)calloc(
                        tmp->tables.tables_len,
                    sizeof (tmp->tables.tables_val[0]));
    if (tmp->tables.tables_val == NULL) {
        delete tmp;
        WRITEUNLOCK(this, NULL,
            "db_dictionary::db_copy_dictionary wu: no memory");
        return (NULL);
    }

    tmp->impl_vers = dictionary->impl_vers;
    tmp->tables.tables_len = 0;
    tmp->count = 0;

    /* For each table ... */
    for (i = 0; ok && i < dictionary->tables.tables_len; i++) {
        db_table_desc_p tbl = NULL,
                t = dictionary->tables.tables_val[i];
        /* ... and each bucket in the chain ... */
        while (ok && t != NULL) {
            db_table_desc_p     n, savenext = t->next;
            t->next = NULL;
            if (db_clone_bucket(t, &n)) {
                if (tbl != NULL) {
                    tbl->next = n;
                } else {
                    tmp->tables.tables_val[i] = n;
                }
                tbl = n;
                tmp->count++;
            } else {
                ok = 0;
            }
            t->next = savenext;
        }
        tmp->tables.tables_len++;
    }

    if (ok) {
#ifdef  NISDB_LDAP_DEBUG
        if ((tmp->tables.tables_len !=
                dictionary->tables.tables_len) ||
            (tmp->count != dictionary->count))
            abort();
#endif  /* NISDB_LDAP_DEBUG */
    } else {
        db_release_dictionary(tmp);
        tmp = NULL;
    }

    return (tmp);
}

/*
 * Set deferred commit mode. To do this, we make a copy of the table
 * (structures and data), and put that on the deferred dictionary list.
 * This list is used for lookups during a resync, so clients continue
 * to see the pre-resync data. Meanwhile, any changes (including table
 * deletes) are done to the (temporarily hidden to clients) table in
 * the normal dictionary.
 */
db_status
db_dictionary::defer(char *table) {
    db_status   ret = DB_SUCCESS;
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::defer");
    db_table_desc   *tbl = find_table_desc(table);
    int     res;
    const char  *myself = "db_dictionary::defer";

    if (tbl != NULL) {
        db_table_desc   *clone, *savenext = tbl->next;
        /*
         * Only want to clone one db_table_desc, so temporarily
         * unlink the tail.
         */
        tbl->next = NULL;
        res = db_clone_bucket(tbl, &clone);
        /* Restore link to tail */
        tbl->next = savenext;
        if (res == 1) {
            db_status   stat;
            if (deferred.dictionary == NULL) {
                deferred.dictionary = new db_dict_desc;
                if (deferred.dictionary == NULL) {
                    WRITEUNLOCK(this, DB_MEMORY_LIMIT,
                        "wu db_dictionary::defer");
                    return (DB_MEMORY_LIMIT);
                }
                deferred.dictionary->tables.tables_len = 0;
                deferred.dictionary->tables.tables_val = NULL;
                deferred.dictionary->count = 0;
                deferred.dictionary->impl_vers =
                            DB_CURRENT_VERSION;
            }
            ret = DB_SUCCESS;
            /* Initialize and load the database for the clone */
            if (clone->database == 0) {
                clone->database = new db(table);
                if (clone->database != 0) {
                    if (clone->database->load()) {
                        logmsg(MSG_NOTIMECHECK,
#ifdef  NISDB_LDAP_DEBUG
                            LOG_WARNING,
#else
                            LOG_INFO,
#endif  /* NISDB_LDAP_DEBUG */
                    "%s: Clone DB for \"%s\" loaded",
                            myself, NIL(table));
                    } else {
                        logmsg(MSG_NOTIMECHECK, LOG_ERR,
                    "%s: Error loading clone DB for \"%s\"",
                            myself, NIL(table));
                        delete clone->database;
                        clone->database = 0;
                        ret = DB_INTERNAL_ERROR;
                    }
                } else {
                    logmsg(MSG_NOTIMECHECK, LOG_ERR,
                    "%s: Unable to clone DB for \"%s\"",
                        myself, NIL(table));
                    ret = DB_MEMORY_LIMIT;
                }
            }
            if (clone->database != 0) {
                clone->database->markDeferred();
                stat = add_to_dictionary(deferred.dictionary,
                            clone);
                ret = stat;
                if (stat != DB_SUCCESS) {
                    delete clone->database;
                    clone->database = 0;
                    delete clone;
                    if (stat == DB_NOTUNIQUE) {
                        /* Already deferred */
                        ret = DB_SUCCESS;
                    }
                }
            } else {
                delete clone;
                /* Return value already set above */
            }
        } else {
            ret = DB_INTERNAL_ERROR;
        }
    } else {
        ret = DB_NOTFOUND;
    }
    WRITEUNLOCK(this, ret, "wu db_dictionary::defer");
    return (ret);
}

/*
 * Unset deferred commit mode and roll back changes; doesn't recover the
 * disk data, but that's OK, since we only want to be able to continue
 * serving the table until we can try a full dump again.
 *
 * The rollback is done by removing (and deleting) the updated table from
 * the dictionary, and then moving the saved table from the deferred
 * dictionary list to the actual one.
 */
db_status
db_dictionary::rollback(char *table) {
    db_status   ret = DB_SUCCESS;
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::rollback");
    db_table_desc   *old = search_dictionary(deferred.dictionary, table);
    db_table_desc   *upd = search_dictionary(dictionary, table);

    if (old == NULL) {
        WRITEUNLOCK(this, DB_NOTFOUND, "wu db_dictionary::rollback");
        return (DB_NOTFOUND);
    }

    /*
     * Remove old incarnation from deferred dictionary. We already hold
     * a pointer ('old') to it, so don't delete.
     */
    ret = remove_from_dictionary(deferred.dictionary, table, FALSE);
    if (ret != DB_SUCCESS) {
#ifdef  NISDB_LDAP_DEBUG
        abort();
#endif  /* NISDB_LDAP_DEBUG */
        WRITEUNLOCK(this, ret, "wu db_dictionary::rollback");
        return (ret);
    }

    if (old->database != 0)
        old->database->unmarkDeferred();

    /*
     * Remove updated incarnation from dictionary. If 'upd' is NULL,
     * the table has been removed while we were in deferred mode, and
     * that's OK; we just need to retain the old incarnation.
     */
    if (upd != NULL) {
        ret = remove_from_dictionary(dictionary, table, FALSE);
        if (ret != DB_SUCCESS) {
#ifdef  NISDB_LDAP_DEBUG
            abort();
#endif  /* NISDB_LDAP_DEBUG */
            /*
             * Cut our losses; delete old incarnation, and leave
             * updated one in place.
             */
            delete_table_desc(old);
            WRITEUNLOCK(this, ret, "wu db_dictionary::rollback");
            return (ret);
        }
        /* Throw away updates */
        delete_table_desc(upd);
    }

    /* (Re-)insert old incarnation in the dictionary */
    ret = add_to_dictionary(dictionary, old);
    if (ret != DB_SUCCESS) {
#ifdef  NISDB_LDAP_DEBUG
        abort();
#endif  /* NISDB_LDAP_DEBUG */
        /* At least avoid memory leak */
        delete_table_desc(old);
        syslog(LOG_ERR,
    "db_dictionary::rollback: rollback error %d for \"%s\"", ret, table);
    }

    WRITEUNLOCK(this, ret, "wu db_dictionary::rollback");
    return (ret);
}

/*
 * Commit changes. Done by simply removing and deleting the pre-resync
 * data from the deferred dictionary.
 */
db_status
db_dictionary::commit(char *table) {
    db_status   ret = DB_SUCCESS;
    WRITELOCK(this, DB_LOCK_ERROR, "w db_dictionary::commit");
    db_table_desc   *old = search_dictionary(deferred.dictionary, table);

    if (old == NULL) {
        /* Fine (we hope); nothing to do */
        WRITEUNLOCK(this, ret, "wu db_dictionary::commit");
        return (DB_SUCCESS);
    }

    ret = remove_from_dictionary(deferred.dictionary, table, FALSE);
    if (ret == DB_SUCCESS)
        delete_table_desc(old);
#ifdef  NISDB_LDAP_DEBUG
    else
        abort();
#endif  /* NISDB_LDAP_DEBUG */

    WRITEUNLOCK(this, ret, "wu db_dictionary::commit");
    return (ret);
}

/*
 * The noWriteThrough flag is used to prevent modifies/updates to LDAP
 * while we're incorporating log data into the in-memory tables.
 */
void
db_dictionary::setNoWriteThrough(void) {
    ASSERTWHELD(this->dict);
    noWriteThrough.flag++;
}

void
db_dictionary::clearNoWriteThrough(void) {
    ASSERTWHELD(this->dict);
    if (noWriteThrough.flag > 0)
        noWriteThrough.flag--;
#ifdef  NISDB_LDAP_DEBUG
    else
        abort();
#endif  /* NISDB_LDAP_DEBUG */
}