---

vsdb.c

#include "config.h"

#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <netinet/in.h>

#include <string.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <errno.h>
#include <alloca.h>
#include <dirent.h>
#include <time.h>

#include "panic.h"
#include "vsdb.h"
#include "util.h"

#ifndef HAVE_FSEEKO
#define fseeko fseek
#define ftello ftell
#endif

static const char magic0[] = { 'J', 'W', 'M', 1};

#define KEYDATA_OFFSET 12
#define MAGIC_OFFSET 0
#define SIZE_OFFSET 8
#define GENERATION_OFFSET 4

#define BAD_OVERFLOW ((uint32_t)(-1))

#define TEMP_NAME_MAX (512)

typedef struct vsdb_hashentry  vsdb_hashentry_t;
typedef struct vsdb_make vsdb_make_t;
struct revision;

#define MODE_READONLY(x) (!MODE_WRITEABLE(x))
#define MODE_WRITEABLE(x) ((x) & O_WRONLY || (x) & O_RDWR)
#define MODE_READABLE(x) ((x) & O_RDONLY || (x) & O_RDWR)
#define MODE_READONLY(x) (!MODE_WRITEABLE(x))

/* 
 * file layout:
 * |magic/flags 4|generation 4 | num 4|record/key data|hash entries 20*num|
 */

static uint32_t vsdb_hash(void *data, size_t sz);
static vsdb_make_t *vsdb_make_start(int fd);
static void vsdb_make_add(vsdb_make_t *vm, void *key, size_t keysz, void *data, size_t datasz);
static void vsdb_make_done(vsdb_make_t *vm, int foo);
static int reload_revision(vsdb_t *vsdb);
static int vsdb_check_latest(vsdb_t *vsdb);
static void revision_unref(struct revision *);
static int vsdb_init_names(vsdb_t *v, char *name);


static void (*errfn)(char *) = NULL;


struct revision {
        uint32_t generation;
        vsdb_hashentry_t *h;
        uintmax_t num;
        FILE *file;
        int ref;
        vsdb_t *vsdb;
        ino_t inode;
        time_t mtime;
};

struct vsdb {
        struct revision *cr;
        char *name;
        char *dbname;
        int mode;
        int ref;
};


struct vsdb_transaction {
        struct revision *r;
        struct event *event;
        int mode;
};

struct event {
        void *key;
        void *data;
        uintmax_t keysize;
        uintmax_t datasize;
        struct event *next;
};


struct vsdb_hashentry {
        uint32_t hash;
        uint32_t offset;
        uint32_t keysize;
        uint32_t datasize;
        uint32_t overflow;
};


struct vsdb_make {
        FILE *file;
        vsdb_hashentry_t *h;
        size_t nh;
        size_t bh;
};


static void
warn_errfn(char *s)
{
        fputs(s, stderr);
        fputc('\n', stderr);
}

static void
die_errfn(char *s)
{
        fputs(s, stderr);
        fputc('\n', stderr);
        exit(1);
}

void
vsdb_set_verbose_mode(bool val)
{
        verbose_mode = val;
        if(val)
                errfn = warn_errfn;
        else 
                errfn = NULL;
}

void
vsdb_set_die_on_error(void)
{
        errfn = die_errfn;
}

void
vsdb_set_errfn(void (*fn)(char *))
{
        errfn = fn;
}


static void
errorf(char *fmt, ...)
{
        va_list ap;
        char buf[256];
        if(!errfn)
                return;
        va_start(ap, fmt);
        vsnprintf(buf, 256, fmt, ap);
        va_end(ap);
        buf[255] = '\0';
        errfn(buf);
}

static void
errornof(char *fmt, ...)
{
        int err;
        va_list ap;
        char buf[256];
        char buf2[256];
        if(!errfn)
                return;
        err = errno;
        va_start(ap, fmt);
        vsnprintf(buf, 256, fmt, ap);
        va_end(ap);
        buf[255] = '\0';
        snprintf(buf2, 256, "%s: %s", buf, strerror(err));
        buf2[255] = '\0';
        errfn(buf2);
}

void 
vsdb_get_info(vsdb_t *vsdb, struct vsdb_info *info)
{
        int i;
        assert(vsdb);
        assert(info);
        assert(vsdb->cr);
        info->mtime = vsdb->cr->mtime;
        info->num_entries = vsdb->cr->num;
        info->generation = vsdb->cr->generation;
        info->name = vsdb->name;
        info->key_size = 0;
        info->data_size = 0;
        for(i = 0; i < info->num_entries; i++) {
                info->key_size += ntohl(vsdb->cr->h[i].keysize);
                info->data_size += ntohl(vsdb->cr->h[i].datasize);
        }
}

uintmax_t 
vsdb_transaction_num_entries(vsdb_transaction_t *vt)
{
        assert(vt);
        return vt->r->num;
}

/* link which does an extra check useful for certain filesystems (like NFS) */

static int
nfs_link(char *fin, char *fout)
{
        struct stat st;
        int ret, ret2;
        ret = link(fin, fout);
        if(ret != 0) { //extra check for broken NFS
                ret2 = stat(fin, &st); 
                if(!ret2 && st.st_nlink == 2)
                        ret = 0;
        }
        return ret;
}

static void
db_file(char *home, char **name, uint32_t dgen, uint32_t fgen)
{
        int ret;
        bfree((void *)name);
        ret = asprintf(name, "%s/dbd.%u/db.%u", home, dgen, fgen);
        if(ret == -1)
                panicno("asprintf");
}

static void 
dbd_dir(char *home, char **name, uint32_t dgen)
{
        int ret;
        bfree((void *)name);
        ret = asprintf(name, "%s/dbd.%u", home, dgen);
        if(ret == -1)
                panicno("asprintf");
}

static void
unlink_old(char *home, uint32_t gen, uint32_t num) 
{
        char *n = NULL;
        assert(num >= 0);
        while(num) {
                db_file(home, &n, gen, gen - num);
                unlink(n);
                num--;
        }
        free(n);
}

static int
rename_dir(char *home, uint32_t from, uint32_t to)
{
        int ret;
        char *da = NULL, *db = NULL;
        assert (to - 1 == from);
        dbd_dir(home, &da, from);
        dbd_dir(home, &db, to);
        ret = rename(da, db);
        free(da);
        free(db);
        return ret;
}

static int
update_db_cache(vsdb_t *vsdb) 
{
        int ret;
        DIR *dir = NULL;
        struct dirent *dirent = NULL;
        //if(vsdb_check_latest(vsdb) == 0) 
        //        return;
        print_verbose("[update_db_cache] started for %s", vsdb->name);
        dir = opendir(vsdb->name);
        if(!dir) {
                errornof("[update_db_cache] cannot opendir %s", vsdb->name);
                return -1;
        }

        for(dirent = readdir(dir);dirent;dirent = readdir(dir)) {
                if(!strncmp(dirent->d_name,"dbd.", 4)) {
                        char *temp_file = NULL;
                        char *rn = NULL;
                        char *d = NULL;
                        unsigned ui;
                        ret = sscanf(dirent->d_name, "dbd.%u", &ui);
                        print_verbose("update_db_cache: found dir %s %u", dirent->d_name, ui);
                        if(ret != 1) {
                                print_verbose("update_db_cache: unable to parse dirname %s", dirent->d_name);
                                continue;
                        }
                        create_temporary_filename(vsdb->name, &temp_file);

                        db_file(vsdb->name, &rn, ui, ui + 1);
                        ret = nfs_link(rn, temp_file);
                        if(ret == 0) {
                                rename(temp_file, vsdb->dbname);
                                rename_dir(vsdb->name, ui, ui + 1);
                        } else {

                                db_file(vsdb->name, &rn, ui, ui );
                                ret = nfs_link(rn, temp_file);
                                if(ret == 0) {
                                        rename(temp_file, vsdb->dbname);
                                } else
                                        unlink(temp_file);
                        }

                        free(temp_file);
                        free(rn);
                        free(d);
                        closedir(dir);
                        return 0;
                        //link( dbd.ui + 1 , tempfile) || link(dbd.ui tempfile) && mv tempfile db.cache
                }
        }
        closedir(dir);
        errorf("[update_db_cache] No dbd.<num> directory found in %s", vsdb->name);
        return -1;
}

vsdb_transaction_t *
vsdb_transaction_begin(vsdb_t *vsdb, int mode)
{

        vsdb_transaction_t *vt = (vsdb_transaction_t *)malloc(sizeof(vsdb_transaction_t));
        assert(vt);
        assert(vsdb);
        assert(mode == O_RDONLY || mode == O_RDWR || mode == O_WRONLY);

        if(MODE_WRITEABLE(mode) && ! MODE_WRITEABLE(vsdb->mode)) {
                errno = EPERM;
                errornof("transaction opened writeable for non-writable database");
                return NULL;
        }
        if(MODE_READABLE(mode) && ! MODE_READABLE(vsdb->mode)) {
                errno = EPERM;
                errornof("transaction opened readable for non-readable database");
                return NULL;
        }
        
        vsdb_check_latest(vsdb);
        
        vt->mode = mode;
        vt->r = vsdb->cr;
        vt->event = NULL;
        vt->r->ref++;
        return vt;
}

void
vsdb_transaction_rollback(vsdb_transaction_t *vt)
{
        revision_unref(vt->r);
        struct event *e = vt->event;
        while(e) {
                struct event *et = e->next;
                free(e);
                e = et;
        }
        free(vt);
}

int
vsdb_transaction_commit(vsdb_transaction_t *vt)
{
        char *s = NULL;
        int i;
        int fd;
        int ret;
        int err;
        vsdb_t *vsdb = vt->r->vsdb;
        vsdb_make_t *vm = NULL;
        struct event *e = NULL;
        char *temp_file = NULL;
        int attempts = 0;

        if(MODE_READONLY(vt->mode) || vt->event == NULL) {
                vsdb_transaction_rollback(vt);
                return 0;
        }

        fd = create_temporary_file(vt->r->vsdb->name, &temp_file);
        if(fd == -1) {
                errornof("[vsdb_transaction_commit] cannot create tempfile");
                return -1;
        }

        assert(fd >= 0);
        assert(temp_file);

        vm = vsdb_make_start(fd);
        assert(vm);
        for(e = vt->event; e; e = e->next) {
                if(e->data) 
                        vsdb_make_add(vm, e->key, e->keysize, e->data, e->datasize);
        }
        if(!(vt->r->vsdb->mode & O_TRUNC))
                for(i = 0; i < vt->r->num; i++) {
                        uint32_t keysz, datasz;
                        void *kptr;
                        void *dptr;
                        keysz = ntohl(vt->r->h[i].keysize);
                        datasz = ntohl(vt->r->h[i].datasize);
                        kptr = malloc(keysz);
                        dptr = malloc(datasz);
                        assert(kptr);
                        assert(dptr);
                        ret = fseek(vt->r->file, ntohl(vt->r->h[i].offset), SEEK_SET);
                        assert(ret != -1);

                        ret = fread(kptr, keysz, 1, vt->r->file);
                        assert(ret != -1);
                        ret = fread(dptr, datasz, 1, vt->r->file);
                        assert(ret != -1);

                        for(e = vt->event; e; e = e->next) {
                                if(e->keysize == keysz && !memcmp(e->key, kptr, keysz)) break;
                        }

                        if(!e)
                                vsdb_make_add(vm, kptr, keysz, dptr, datasz);

                        free(kptr);
                        free(dptr);
                }

        vsdb_make_done(vm, vt->r->generation + 1);


        db_file(vt->r->vsdb->name, &s, vt->r->generation, vt->r->generation + 1);
        assert(s);

        ret = nfs_link(temp_file, s);
        if(ret == 0) { 
                //char *odir = NULL, *ndir = NULL, *ofile = NULL;

                rename(temp_file, vt->r->vsdb->dbname); // update db.cache
                rename_dir(vt->r->vsdb->name, vt->r->generation, vt->r->generation + 1);
                unlink_old(vt->r->vsdb->name, vt->r->generation + 1, 1);

                

                free(s);
                free(temp_file);
                vsdb = vt->r->vsdb;
                vt->r->vsdb->mode &= ~O_TRUNC;
                vsdb_transaction_rollback(vt);
                sync();
                ret = vsdb_check_latest(vsdb);
                assert(ret != -1);
                assert(ret == 0); //always should happen since we just commited succesfully.

                return 0;
        }
        err = errno;
        remove(temp_file);
        free(temp_file);
        free(s);

        attempts = 0;
        do { //look for new database
                struct timespec ts;
                ret = vsdb_check_latest(vsdb);
                if(ret <= 0) {
                        vsdb_transaction_rollback(vt);
                        if(ret == -1) {
                                errorf("[vsdb_transaction_commit] vsdb_check_latest had error");
                                return -1;
                        }
                        assert(ret == 0);
                        print_verbose("new database found");
                        return 1;
                }
                ts.tv_sec = 0;
                ts.tv_nsec = 2000;
                nanosleep(&ts, NULL);
                attempts++;
        } while(attempts < 7);


        // something odd might be happening. investigate
        // update failed, let's find out why.

        if(err == EEXIST) {
                //link exists. possible partial update.
                ret = rename_dir(vt->r->vsdb->name, vt->r->generation, vt->r->generation + 1);
                unlink_old(vt->r->vsdb->name, vt->r->generation + 1, 1);
        }

        update_db_cache(vsdb);

        vsdb_transaction_rollback(vt);
        return 1;
}



static void
revision_unref(struct revision *r)
{
        if(!r)
                return;
        assert(r->ref > 0);
        if(r->ref == 1) {
                fclose(r->file);
                free(r->h);
                free(r);
        } else
                --r->ref;
}

int vsdb_check_latest(vsdb_t *vsdb) { int ret; struct stat st; assert(vsdb);
        ret = stat(vsdb->dbname, &st); if(ret == -1) { errornof("stat(%s)",
                                                                vsdb->dbname);
        return -1; } if(st.st_mtime != vsdb->cr->mtime || st.st_ino !=
                        vsdb->cr->inode) { return reload_revision(vsdb); }
        return 1; }

static int
reload_revision(vsdb_t *vsdb)
{
        int err;
        int ret;
        off_t offset;
        struct revision *r = malloc(sizeof(struct revision));
        uint32_t u32;
        struct stat st;

        assert(vsdb);
        assert(r);

        revision_unref(vsdb->cr);
        vsdb->cr = r;

        r->ref = 1;
        r->vsdb = vsdb;
        r->file = fopen(vsdb->dbname, "r");
        if(!r->file) {
                errornof("fopen(%s, \"r\")", vsdb->dbname);
                return -1;
        }
        ret = fstat(fileno(r->file), &st);
        assert(ret != -1);
        r->inode = st.st_ino;
        r->mtime = st.st_mtime;
        
        /* verify magic */
        err = fseeko(r->file, MAGIC_OFFSET, SEEK_SET);
        assert(err != -1);
        err = fread(&u32, 4, 1, r->file);
        assert(err != -1);
        assert(!memcmp(&u32, magic0, 4));

        /* read size */
        err = fseeko(r->file, SIZE_OFFSET, SEEK_SET);
        assert(err != -1);
        err = fread(&u32, 4, 1, r->file);
        assert(err != -1);
        r->num = ntohl(u32);

        /* read generation */
        err = fseeko(r->file, GENERATION_OFFSET, SEEK_SET);
        assert(err != -1);
        err = fread(&u32, 4, 1, r->file);
        assert(err != -1);
        r->generation = ntohl(u32);

        /* find hashtable */
        err = fseeko(r->file,  0, SEEK_END);
        assert(err != -1);
        offset = ftello(r->file);
        assert(offset != -1);
        //printf("filesize: %u\n", (unsigned)offset);
        offset -= sizeof(vsdb_hashentry_t)*r->num;
        //printf("reading offset: %u\n", (unsigned)offset);


        err = fseeko(r->file, offset, SEEK_SET);
        assert(err != -1);


        
        r->h = malloc(sizeof(vsdb_hashentry_t)*r->num);
        assert(r->h);
        
        err = fread(r->h, sizeof(vsdb_hashentry_t), r->num, r->file);
        if(err == -1) {
                panicno("fread");
        }

        /*
        for(i = 0; i < r->num; i++) {
                printf("Number: %i\n", (int)i);
                printf("keysize: %u\n", ntohl(r->h[i].keysize));
                printf("datasize: %u\n", ntohl(r->h[i].datasize));
                printf("hash: %X\n", ntohl(r->h[i].hash));
                printf("overflow: %u\n", ntohl(r->h[i].overflow));
                printf("offset: %u\n\n", ntohl(r->h[i].offset));
        }
        */

        return 0;
        
}


static void 
create_database(char *s)
{
        int fd, ret;
        vsdb_make_t *vm = NULL;
        char *temp_file = NULL;
        char *temp_file2 = NULL;
        
        print_verbose("Creating database: %s", s);

        fd = create_temporary_file(s, &temp_file);
        if(fd < 0) {
                panicno("creating database");
                return;
        }
        
        vm = vsdb_make_start(fd);
        if(!vm)
                return;
        vsdb_make_done(vm, 0);

        asprintf(&temp_file2, "%s/dbd.0", s);
        mkdir(temp_file2, 0777);
        free(temp_file2);
        asprintf(&temp_file2, "%s/dbd.0/db.0", s);
        ret = link(temp_file, temp_file2);
        assert(ret != -1);
        free(temp_file2);
        asprintf(&temp_file2, "%s/db.cache", s);
        ret = rename(temp_file, temp_file2);
        assert(ret != -1);
        free(temp_file2);
        free(temp_file);
}

static void
vsdb_new_database(char *s)
{
        int ret;
        ret = mkdir(s, 0777);
        if(!ret) {
                create_database(s);
        }
        
}

int
vsdb_repair_database(char *s)
{
        // ensure db.cache is pointing to proper database
        // ensure only single dbd.# exists
        // ensure directory is properly named
        // erase any old db.# files.
        vsdb_t vsdb;
        vsdb_init_names(&vsdb, s);

        return update_db_cache(&vsdb); //TODO hacky
}

static int
vsdb_init_names(vsdb_t *v, char *name)
{
        int ret;
        if(name[0] == '/') {
                v->name = strdup(name);
                if(!v->name) {
                        errornof("[vsdb_init_names] strdup");
                        return -1;
                }
        } else {
                char *cwd = get_current_dir_name();
                if(!cwd) {
                        errornof("[vsdb_init_names] get_current_dir_name");
                        return -1;
                }
                ret = asprintf(&v->name, "%s/%s", cwd, name);
                free(cwd);
                if(ret == -1) {
                        errornof("[vsdb_init_names] asprintf");
                        return -1;
                }
        }

        if(v->name[strlen(v->name) - 1] == '/')
                v->name[strlen(v->name) - 1] = '\0';

        ret = asprintf(&v->dbname, "%s/db.cache", v->name);
        if(ret == -1) {
                free(v->name);
                free(v);
                errornof("[vsdb_init_names] asprintf");
                return -1;
        }
        assert(v->dbname);

        return 0;
}

vsdb_t *
vsdb_open(char *name, int mode)
{
        vsdb_t *v;
        int ret;

        assert(name);
        if(!name[0]) 
                return NULL;
        print_verbose("Attempting open: %s", name);

        v = (vsdb_t *)malloc(sizeof(vsdb_t));
        assert(v != NULL);
        v->mode = mode;
        v->ref = 1;
        v->cr = NULL;
        v->name = NULL;
        v->dbname = NULL;

        ret = vsdb_init_names(v, name);
        if(ret == -1) {
                free(v);
                return NULL;
        }



        /* create database if it doesn't exist */
        if(mode & O_CREAT) {
                print_verbose("creating database: %s", name);
                vsdb_new_database(v->name);
        }



        if(reload_revision(v) != 0) {
                free(v->name);
                free(v->dbname);
                free(v);
                errorf("[vsdb_open] reload_revision");
                return NULL;
        }

        return v;
}

void
vsdb_close(vsdb_t *v)
{
        if(v->mode & O_TRUNC) {
                vsdb_transaction_t *vt;
                do {
                        vt = vsdb_transaction_begin(v, O_WRONLY);
                } while(vsdb_transaction_commit(vt) > 0);
        }
        revision_unref(v->cr);
        free(v->name);
        free(v->dbname);
        free(v);
}

void
vsdb_clear(vsdb_transaction_t *vt, void *key, size_t keysz)
{
        struct event *e = (struct event *)malloc(sizeof(struct event));
        assert(key || keysz == 0);
        e->key = key;
        e->keysize = keysz;
        e->data = NULL;
        e->datasize = 0;

        e->next = vt->event;
        vt->event = e;
}

void 
vsdb_set(vsdb_transaction_t *vt, void *key, size_t keysz, void *data, size_t datasz)
{
        struct event *e = (struct event *)malloc(sizeof(struct event));
        assert(key || keysz == 0);
        assert(data || datasz == 0);
        e->key = key;
        e->keysize = keysz;
        if(data == NULL && datasz == 0) 
                e->data = e;
        else
                e->data = data;
        e->datasize = datasz;

        e->next = vt->event;
        vt->event = e;

}

int
vsdb_lookup(vsdb_transaction_t *vt, void *key, size_t keysz, uintmax_t *which)
{
        int i;
        int ret;
        uint32_t hash = vsdb_hash(key, keysz);
        uint32_t u32;
        *which = 0;
        if(vt->r->num == 0)
                return 1;
        if(vt->r->vsdb->mode & O_TRUNC)
                return 1;  // simulate empty database
        i = hash % vt->r->num;
        while(1) {
                if(hash == ntohl(vt->r->h[i].hash) && ntohl(vt->r->h[i].keysize) == keysz) {
                        off_t offset = ntohl(vt->r->h[i].offset);
                        void *keyt = alloca(keysz);
                        assert(keyt);
                        ret = fseeko(vt->r->file, offset, SEEK_SET);
                        if(ret == -1)
                                return -1;
                        ret = fread(keyt, keysz, 1, vt->r->file);
                        if(ret == -1) 
                                return -1;
                        if(!memcmp(keyt, key, keysz)) {
                                *which = i;
                                return 0;
                        }

                }
                u32 = ntohl(vt->r->h[i].overflow); 
                if(u32 != BAD_OVERFLOW) 
                        i = u32;
                else
                        return 1;
        }
        
}

void
vsdb_aread_entry(vsdb_transaction_t *vt, uintmax_t entry, void **key, size_t *keysz, void **data, size_t *datasz)
{
        int ret;
        size_t datasize = ntohl(vt->r->h[entry].datasize);
        size_t keysize = ntohl(vt->r->h[entry].keysize);

        ret = fseeko(vt->r->file, ntohl(vt->r->h[entry].offset), SEEK_SET);
        assert(ret != -1);
        

        if(keysz)
                *keysz = keysize;
        if(key) {
                *key = malloc(keysize);
                ret = fread(*key, keysize, 1, vt->r->file);
                assert(ret != -1);
        } else {
                ret = fseeko(vt->r->file, keysize, SEEK_CUR);
                assert(ret != -1);
        }
        if(datasz)
                *datasz = datasize;
        if(data) {
                *data = malloc(datasize);
                ret = fread(*data, datasize, 1, vt->r->file);
                assert(ret != -1);
        }
}

/*
static int
vsdb_lookup_rr(vsdb_transaction_t *vt, void *key, size_t keysz, void **data, size_t *datasz)
{

        int ret;
        uintmax_t which;
        ret = vsdb_lookup(vt, key, keysz, &which);
        if(ret != 0)
                return ret;
        vsdb_aread_entry(vt, which, NULL, NULL, data, datasz);
        return 0;
}
*/

char *
vsdb_lookup_ss(vsdb_transaction_t *vt, char *key)
{
        void *data;
        size_t datasz;
        uintmax_t which;
        int ret;
        ret = vsdb_lookup(vt, key, strlen(key), &which);
        if(ret == 0) {
                char *ndata;
                vsdb_aread_entry(vt, which, NULL, NULL, &data, &datasz);
                ndata = malloc(datasz + 1);
                assert(ndata);
                memcpy(ndata, data, datasz);
                free(data);
                ndata[datasz] = '\0';
                return ndata;
        }
        return NULL;
}

static uint32_t 
vsdb_hash(void *data, size_t sz)
{
        int i;
        char *c = data;
        uint32_t hash = 5381;
        for(i = 0; i < sz; i++)
                hash = ((hash << 5) + hash) ^ *c++;
        return hash;
}


static vsdb_make_t *
vsdb_make_start(int fd)
{
        int ret;

        vsdb_make_t *vm = malloc(sizeof(vsdb_make_t));
        if(!vm)
                return vm;

        vm->file = fdopen(fd, "w");
        if(!vm->file)
        {
                perror("fdopen");
                fclose(vm->file);
                return NULL;
        }
        ret = fwrite(magic0, 4, 1, vm->file);
        assert(ret != -1);
        
        ret = fseeko(vm->file, KEYDATA_OFFSET, SEEK_SET);
        if(ret == -1) {
                perror("fseeko");
                fclose(vm->file);
                free(vm);
                return NULL;
        }
        vm->h = NULL;
        vm->nh = 0;
        vm->bh = 0;
        return vm;
}


static void
vsdb_make_add(vsdb_make_t *vm, void *key, size_t keysz, 
              void *data, size_t datasz)
{
        ssize_t werr;
        vsdb_hashentry_t vh;
        off_t off = ftello(vm->file);
        if(off == -1) 
                panicno("ftello");
        
        vh.hash = vsdb_hash(key, keysz);
        vh.offset = off;
        vh.keysize = keysz;
        vh.datasize = datasz;
        vh.overflow = BAD_OVERFLOW;
        assert(vm->bh >= vm->nh);
        if(vm->bh == vm->nh) {
                if (!vm->bh) 
                        vm->bh = 4;
                vm->bh *= 2;
                vm->h = realloc(vm->h, vm->bh*sizeof(vsdb_hashentry_t));
                assert(vm->h);
        }
        vm->h[vm->nh] = vh;
        vm->nh++;

        werr = fwrite(key, keysz, 1, vm->file); 
        if (werr == -1 )
                panicno("fwrite");
        werr = fwrite(data, datasz, 1, vm->file); 
        if (werr == -1 )
                panicno("fwrite");

}

static void
vsdb_make_done(vsdb_make_t *vm, int generation)
{
        int err;
        uint32_t buf;
        off_t offset;
        int i,j;
        vsdb_hashentry_t *he = malloc(vm->nh * sizeof(vsdb_hashentry_t));
        assert(sizeof(vsdb_hashentry_t) == 20);
        assert(he);
        memset(he, 0, vm->nh * sizeof(vsdb_hashentry_t));
        for(i = 0; i < vm->nh; i++) {
                uint32_t loc = (vm->h[i].hash % vm->nh);
                if(he[loc].offset == 0) {
                        he[loc] = vm->h[i];
                        vm->h[i].offset = 0; //indicates this was placed in this stage
                } 
        }

        i = 0; j = 0;
        while(1) {
                while(i < vm->nh && vm->h[i].offset == 0) i++;
                if(i >= vm->nh) break;
                while(j < vm->nh && he[j].offset) j++;
                if(j >= vm->nh) assert(!"this shouldn't happen");

                assert(vm->h[i].offset);
                assert(!he[j].offset);

                he[j] = vm->h[i];
                he[j].overflow = he[vm->h[i].hash % vm->nh].overflow;
                he[vm->h[i].hash % vm->nh].overflow = j;
                i++; j++;
        };

        offset = ftello(vm->file);
        //printf("make offset: %u\n", (unsigned)offset);

        for(i = 0; i < vm->nh*sizeof(vsdb_hashentry_t)/4; i++)
        {

                buf = htonl(((uint32_t *)he)[i]);
                err = fwrite(&buf, 4, 1, vm->file);
                assert(err != -1);
        }
        /* write size */
        err = fseek(vm->file, SIZE_OFFSET, SEEK_SET);
        assert(err != -1);
        buf = htonl(vm->nh);
        err = fwrite(&buf, 4, 1, vm->file);
        assert(err != -1);

        /* write generation */
        err = fseek(vm->file, GENERATION_OFFSET, SEEK_SET);
        assert(err != -1);
        buf = htonl(generation);
        err = fwrite(&buf, 4, 1, vm->file);
        assert(err != -1);

        fflush(vm->file);
        /* clean up */
        err = fclose(vm->file);
        assert(err != -1);
        free(vm->h);
        free(he);
        free(vm);
}

---