<commit_before>/* -*- mode: C; c-basic-offset: 4 -*- */
#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."

#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

const char *toku_patent_string = "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it.";
const char *toku_copyright_string = "Copyright (c) 2007 Tokutek Inc.  All rights reserved.";

#include <assert.h>
#include <errno.h>
#include <limits.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <ctype.h>
#include <unistd.h>
#include <libgen.h>

#include "ydb-internal.h"

#include "brt-internal.h"
#include "cachetable.h"
#include "log.h"
#include "memory.h"

struct __toku_db_txn_internal {
    //TXNID txnid64; /* A sixty-four bit txn id. */
    TOKUTXN tokutxn;
    DB_TXN *parent;
};

static char *construct_full_name(const char *dir, const char *fname);
static int do_associated_inserts (DB_TXN *txn, DBT *key, DBT *data, DB *secondary);
    
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR == 1
typedef void (*toku_env_errcall_t)(const char *, char *);
#elif DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3
typedef void (*toku_env_errcall_t)(const DB_ENV *, const char *, const char *);
#else
#error
#endif

struct __toku_db_env_internal {
    int is_panicked;
    int ref_count;
    u_int32_t open_flags;
    int open_mode;
    toku_env_errcall_t errcall;
    void *errfile;
    const char *errpfx;
    char *dir;                  /* A malloc'd copy of the directory. */
    char *tmp_dir;
    char *lg_dir;
    char **data_dirs;
    u_int32_t n_data_dirs;
    //void (*noticecall)(DB_ENV *, db_notices);
    unsigned long cachetable_size;
    CACHETABLE cachetable;
    TOKULOGGER logger;
};

// If errcall is set, call it with the format string and optionally the stderrstring (if include_stderrstring).  The prefix is passed as a separate argument.
// If errfile is set, print to the errfile: prefix, fmt string, maybe include the stderr string.
// Both errcall and errfile may be called.
// If errfile is not set and errcall is not set, the use stderr as the errfile.
void toku_do_error_all_cases(const DB_ENV * env, int error, int include_stderrstring, int use_stderr_if_nothing_else, const char *fmt, va_list ap) {
    if (env->i->errcall) {
	// errcall gets prefix sent separately
	// the error message is the printf message, maybe followed by ": " and the dbstrerror (if include_stderrstring is set)
	char buf [4000];
	int count=0;
	if (fmt) {
	    count=vsnprintf(buf, sizeof(buf), fmt, ap);
	}
	if (include_stderrstring) {
	    count+=snprintf(&buf[count], sizeof(buf)-count, ": %s", db_strerror(error));
	}
	env->i->errcall(env, env->i->errpfx, buf);
    }
    {
	FILE *efile=env->i->errfile;
	if (efile==0 && env->i->errcall==0 && use_stderr_if_nothing_else) {
	    efile = stderr;
	}
	if (efile) {
	    if (env->i->errpfx) fprintf(efile, "%s: ", env->i->errpfx);
	    vfprintf(efile, fmt, ap);
	    if (include_stderrstring) {
		fprintf(efile, ": %s", db_strerror(error));
	    }
	}
    }
}


static int env_is_panicked(DB_ENV *dbenv) {
    if (dbenv==0) return 0;
    return dbenv->i->is_panicked || toku_logger_panicked(dbenv->i->logger);
}
#define HANDLE_PANICKED_ENV(env) ({ if (env_is_panicked(env)) return EINVAL; })
#define HANDLE_PANICKED_DB(db) HANDLE_PANICKED_ENV(db->dbenv)

// Handle all the error cases (but don't do the default thing.)
static int do_error (DB_ENV *dbenv, int error, const char *string, ...) {
    if (toku_logger_panicked(dbenv->i->logger)) dbenv->i->is_panicked=1;
    va_list ap;
    va_start(ap, string);
    toku_do_error_all_cases(dbenv, error, 1, 0, string, ap);
    va_end(ap);
    return error;
}


static void toku_db_env_err(const DB_ENV * env, int error, const char *fmt, ...) {
    va_list ap;
    va_start(ap, fmt);
    toku_do_error_all_cases(env, error, 1, 1, fmt, ap);
    va_end(ap);
}

#define barf() ({ fprintf(stderr, "YDB: BARF %s:%d in %s\n", __FILE__, __LINE__, __func__); })
#define barff(fmt,...) ({ fprintf(stderr, "YDB: BARF %s:%d in %s, ", __FILE__, __LINE__, __func__); fprintf(stderr, fmt, __VA_ARGS__); })
#define note() ({ fprintf(svtderr, "YDB: Note %s:%d in %s\n", __FILE__, __LINE__, __func__); })
#define notef(fmt,...) ({ fprintf(stderr, "YDB: Note %s:%d in %s, ", __FILE__, __LINE__, __func__); fprintf(stderr, fmt, __VA_ARGS__); })

#if 0
static void print_flags(u_int32_t flags) {
    u_int32_t gotit = 0;
    int doneone = 0;
#define doit(flag) if (flag & flags) { if (doneone) fprintf(stderr, " | "); fprintf(stderr, "%s", #flag);  doneone=1; gotit|=flag; }
    doit(DB_INIT_LOCK);
    doit(DB_INIT_LOG);
    doit(DB_INIT_MPOOL);
    doit(DB_INIT_TXN);
    doit(DB_CREATE);
    doit(DB_THREAD);
    doit(DB_RECOVER);
    doit(DB_PRIVATE);
    if (gotit != flags)
        fprintf(stderr, "  flags 0x%x not accounted for", flags & ~gotit);
    fprintf(stderr, "\n");
}
#endif

/* TODO make these thread safe */

/* a count of the open env handles */
static int toku_ydb_refs = 0;

static void ydb_add_ref() {
    toku_ydb_refs += 1;
}

static void ydb_unref() {
    assert(toku_ydb_refs > 0);
    toku_ydb_refs -= 1;
    if (toku_ydb_refs == 0) {
        /* call global destructors */
        toku_malloc_cleanup();
    }
}

static void db_env_add_ref(DB_ENV *env) {
    env->i->ref_count += 1;
}

static void db_env_unref(DB_ENV *env) {
    env->i->ref_count -= 1;
    if (env->i->ref_count == 0)
        env->close(env, 0);
}

static inline int db_env_opened(DB_ENV *env) {
    return env->i->cachetable != 0;
}

static inline int db_opened(DB *db) {
    return db->i->full_fname != 0;
}

static int db_env_parse_config_line(DB_ENV* dbenv, char *command, char *value) {
    int r;
    
    if (!strcmp(command, "set_data_dir")) {
        r = dbenv->set_data_dir(dbenv, value);
    }
    else if (!strcmp(command, "set_tmp_dir")) {
        r = dbenv->set_tmp_dir(dbenv, value);
    }
    else if (!strcmp(command, "set_lg_dir")) {
        r = dbenv->set_lg_dir(dbenv, value);
    }
    else r = -1;
        
    return r;
}

static int db_env_read_config(DB_ENV *env) {
    HANDLE_PANICKED_ENV(env);
    const char* config_name = "DB_CONFIG";
    char* full_name = NULL;
    char* linebuffer = NULL;
    int buffersize;
    FILE* fp = NULL;
    int r = 0;
    int r2 = 0;
    char* command;
    char* value;
    
    full_name = construct_full_name(env->i->dir, config_name);
    if (full_name == 0) {
        r = ENOMEM;
        goto cleanup;
    }
    if ((fp = fopen(full_name, "r")) == NULL) {
        //Config file is optional.
        if (errno == ENOENT) {
            r = EXIT_SUCCESS;
            goto cleanup;
        }
        r = errno;
        goto cleanup;
    }
    //Read each line, applying configuration parameters.
    //After ignoring leading white space, skip any blank lines
    //or comments (starts with #)
    //Command contains no white space.  Value may contain whitespace.
    int linenumber;
    int ch = '\0';
    BOOL eof = FALSE;
    char* temp;
    char* end;
    int index;
    
    buffersize = 1<<10; //1KB
    linebuffer = toku_malloc(buffersize);
    if (!linebuffer) {
        r = ENOMEM;
        goto cleanup;
    }
    for (linenumber = 1; !eof; linenumber++) {
        /* Read a single line. */
        for (index = 0; TRUE; index++) {
            if ((ch = getc(fp)) == EOF) {
                eof = TRUE;
                if (ferror(fp)) {
                    /* Throw away current line and print warning. */
                    r = errno;
                    goto readerror;
                }
                break;
            }
            if (ch == '\n') break;
            if (index + 1 >= buffersize) {
                //Double the buffer.
                buffersize *= 2;
                linebuffer = toku_realloc(linebuffer, buffersize);
                if (!linebuffer) {
                    r = ENOMEM;
                    goto cleanup;
                }
            }
            linebuffer[index] = ch;
        }
        linebuffer[index] = '\0';
        end = &linebuffer[index];

        /* Separate the line into command/value */
        command = linebuffer;
        //Strip leading spaces.
        while (isspace(*command) && command < end) command++;
        //Find end of command.
        temp = command;
        while (!isspace(*temp) && temp < end) temp++;
        *temp++ = '\0'; //Null terminate command.
        value = temp;
        //Strip leading spaces.
        while (isspace(*value) && value < end) value++;
        if (value < end) {
            //Strip trailing spaces.
            temp = end;
            while (isspace(*(temp-1))) temp--;
            //Null terminate value.
            *temp = '\0';
        }
        //Parse the line.
        if (strlen(command) == 0 || command[0] == '#') continue; //Ignore Comments.
        r = db_env_parse_config_line(env, command, value < end ? value : "");
        if (r != 0) goto parseerror;
    }
    if (0) {
readerror:
        do_error(env, r, "Error reading from DB_CONFIG:%d.\n", linenumber);
    }
    if (0) {
parseerror:
        do_error(env, r, "Error parsing DB_CONFIG:%d.\n", linenumber);
    }
cleanup:
    if (full_name) toku_free(full_name);
    if (linebuffer) toku_free(linebuffer);
    if (fp) r2 = fclose(fp);
    return r ? r : r2;
}

static int toku_db_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
    HANDLE_PANICKED_ENV(env);
    int r;

    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "The environment is already open\n");
    }

    if ((flags & DB_USE_ENVIRON) && (flags & DB_USE_ENVIRON_ROOT)) {
	return do_error(env, EINVAL, "DB_USE_ENVIRON and DB_USE_ENVIRON_ROOT are incompatible flags\n");
    }

    if (home) {
        if ((flags & DB_USE_ENVIRON) || (flags & DB_USE_ENVIRON_ROOT)) {
	    return do_error(env, EINVAL, "DB_USE_ENVIRON and DB_USE_ENVIRON_ROOT are incompatible with specifying a home\n");
	}
    }
    else if ((flags & DB_USE_ENVIRON) ||
             ((flags & DB_USE_ENVIRON_ROOT) && geteuid() == 0)) home = getenv("DB_HOME");

    if (!home) home = ".";

	// Verify that the home exists.
	{
	struct stat buf;
	r = stat(home, &buf);
	if (r!=0) {
	    return do_error(env, errno, "Error from stat(\"%s\",...)\n", home);
	}
    }



    if (!(flags & DB_PRIVATE)) {
        // This means that we don't have to do anything with shared memory.  
        // And that's good enough for mysql. 
	return do_error(env, EINVAL, "TokuDB requires DB_PRIVATE when opening an env\n");
    }

    if (env->i->dir)
        toku_free(env->i->dir);
    env->i->dir = toku_strdup(home);
    if (env->i->dir == 0) {
	return do_error(env, ENOMEM, "Out of memory\n");
    }
    if (0) {
        died1:
        toku_free(env->i->dir);
        env->i->dir = NULL;
        return r;
    }
    if ((r = db_env_read_config(env)) != 0) {
	goto died1;
    }
    
    env->i->open_flags = flags;
    env->i->open_mode = mode;

    if (flags & (DB_INIT_TXN | DB_INIT_LOG)) {
        char* full_dir = NULL;
        if (env->i->lg_dir) full_dir = construct_full_name(env->i->dir, env->i->lg_dir);
	assert(env->i->logger);
        r = toku_logger_open(full_dir ? full_dir : env->i->dir, env->i->logger);
        if (full_dir) toku_free(full_dir);
	if (r!=0) {
	    do_error(env, r, "Could not open logger\n");
	died2:
	    toku_logger_close(&env->i->logger);
	    goto died1;
	}
    }

    r = toku_brt_create_cachetable(&env->i->cachetable, env->i->cachetable_size, ZERO_LSN, env->i->logger);
    if (r!=0) goto died2;
    return 0;
}

static int toku_db_env_close(DB_ENV * env, u_int32_t flags) {
    // Even if the env is panicedk, try to close as much as we can.
    int is_panicked = env_is_panicked(env);
    int r0=0,r1=0;
    if (env->i->cachetable)
        r0=toku_cachetable_close(&env->i->cachetable);
    if (env->i->logger)
        r1=toku_logger_close(&env->i->logger);
    if (env->i->data_dirs) {
        u_int32_t i;
        assert(env->i->n_data_dirs > 0);
        for (i = 0; i < env->i->n_data_dirs; i++) {
            toku_free(env->i->data_dirs[i]);
        }
        toku_free(env->i->data_dirs);
    }
    if (env->i->lg_dir)
        toku_free(env->i->lg_dir);
    if (env->i->tmp_dir)
        toku_free(env->i->tmp_dir);
    toku_free(env->i->dir);
    toku_free(env->i);
    toku_free(env);
    ydb_unref();
    if (flags!=0) return EINVAL;
    if (r0) return r0;
    if (r1) return r1;
    if (is_panicked) return EINVAL;
    return 0;
}

static int toku_db_env_log_archive(DB_ENV * env, char **list[], u_int32_t flags) {
    env=env; flags=flags; // Suppress compiler warnings.
    *list = NULL;
    return 0;
}

static int toku_db_env_log_flush(DB_ENV * env, const DB_LSN * lsn) {
    HANDLE_PANICKED_ENV(env);
    env=env; lsn=lsn;
    barf();
    return 1;
}

static int toku_db_env_set_cachesize(DB_ENV * env, u_int32_t gbytes, u_int32_t bytes, int ncache __attribute__((__unused__))) {
    HANDLE_PANICKED_ENV(env);
    u_int64_t cs64 = ((u_int64_t) gbytes << 30) + bytes;
    unsigned long cs = cs64;
    if (cs64 > cs)
        return EINVAL;
    env->i->cachetable_size = cs;
    return 0;
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3

static int toku_db_env_get_cachesize(DB_ENV * env, u_int32_t *gbytes, u_int32_t *bytes, int *ncache) {
    HANDLE_PANICKED_ENV(env);
    *gbytes = env->i->cachetable_size >> 30;
    *bytes = env->i->cachetable_size & ((1<<30)-1);
    *ncache = 1;
    return 0;
}

#endif

static int toku_db_env_set_data_dir(DB_ENV * env, const char *dir) {
    HANDLE_PANICKED_ENV(env);
    u_int32_t i;
    int r;
    char** temp;
    char* new_dir;
    
    if (db_env_opened(env) || !dir) {
	return do_error(env, EINVAL, "You cannot set the data dir after opening the env\n");
    }
    
    if (env->i->data_dirs) {
        assert(env->i->n_data_dirs > 0);
        for (i = 0; i < env->i->n_data_dirs; i++) {
            if (!strcmp(dir, env->i->data_dirs[i])) {
                //It is already in the list.  We're done.
                return 0;
            }
        }
    }
    else assert(env->i->n_data_dirs == 0);
    new_dir = toku_strdup(dir);
    if (0) {
        died1:
        toku_free(new_dir);
        return r;
    }
    if (new_dir==NULL) {
	assert(errno == ENOMEM);
	return do_error(env, errno, "Out of memory\n");
    }
    temp = (char**) toku_realloc(env->i->data_dirs, (1 + env->i->n_data_dirs) * sizeof(char*));
    if (temp==NULL) {assert(errno == ENOMEM); r = ENOMEM; goto died1;}
    else env->i->data_dirs = temp;
    env->i->data_dirs[env->i->n_data_dirs] = new_dir;
    env->i->n_data_dirs++;
    return 0;
}

static void toku_db_env_set_errcall(DB_ENV * env, toku_env_errcall_t errcall) {
    env->i->errcall = errcall;
}

static void toku_db_env_set_errfile(DB_ENV*env, FILE*errfile) {
    env->i->errfile = errfile;
}

static void toku_db_env_set_errpfx(DB_ENV * env, const char *errpfx) {
    env->i->errpfx = errpfx;
}

static int toku_db_env_set_flags(DB_ENV * env, u_int32_t flags, int onoff) {
    HANDLE_PANICKED_ENV(env);
    if (flags != 0 && onoff) {
	return do_error(env, EINVAL, "TokuDB does not (yet) support any nonzero ENV flags\n");
    }
    return 0;
}

static int toku_db_env_set_lg_bsize(DB_ENV * env, u_int32_t bsize) {
    HANDLE_PANICKED_ENV(env);
    bsize=bsize;
    return do_error(env, EINVAL, "TokuDB does not (yet) support ENV->set_lg_bsize\n");
}

static int toku_db_env_set_lg_dir(DB_ENV * env, const char *dir) {
    HANDLE_PANICKED_ENV(env);
    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "Cannot set log dir after opening the env\n");
    }

    if (env->i->lg_dir) toku_free(env->i->lg_dir);
    if (dir) {
        env->i->lg_dir = toku_strdup(dir);
        if (!env->i->lg_dir) {
	    return do_error(env, ENOMEM, "Out of memory\n");
	}
    }
    else env->i->lg_dir = NULL;
    return 0;
}

static int toku_db_env_set_lg_max(DB_ENV * env, u_int32_t lg_max) {
    HANDLE_PANICKED_ENV(env);
    lg_max=lg_max;
    return do_error(env, EINVAL, "TokuDB does not (yet) support set_lg_max\n");
}

static int toku_db_env_set_lk_detect(DB_ENV * env, u_int32_t detect) {
    HANDLE_PANICKED_ENV(env);
    detect=detect;
    return do_error(env, EINVAL, "TokuDB does not (yet) support set_lk_detect\n");
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR <= 4
static int toku_db_env_set_lk_max(DB_ENV * env, u_int32_t lk_max) {
    HANDLE_PANICKED_ENV(env);
    lk_max=lk_max;
    return 0;
}
#endif

//void __toku_db_env_set_noticecall (DB_ENV *env, void (*noticecall)(DB_ENV *, db_notices)) {
//    env->i->noticecall = noticecall;
//}

static int toku_db_env_set_tmp_dir(DB_ENV * env, const char *tmp_dir) {
    HANDLE_PANICKED_ENV(env);
    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "Cannot set the tmp dir after opening an env\n");
    }
    if (!tmp_dir) {
	return do_error(env, EINVAL, "Tmp dir bust be non-null\n");
    }
    if (env->i->tmp_dir)
        toku_free(env->i->tmp_dir);
    env->i->tmp_dir = toku_strdup(tmp_dir);
    return env->i->tmp_dir ? 0 : ENOMEM;
}

static int toku_db_env_set_verbose(DB_ENV * env, u_int32_t which, int onoff) {
    HANDLE_PANICKED_ENV(env);
    which=which; onoff=onoff;
    return 1;
}

static int toku_db_env_txn_checkpoint(DB_ENV * env, u_int32_t kbyte, u_int32_t min, u_int32_t flags) {
    env=env; kbyte=kbyte; min=min; flags=flags;
    return 0;
}

static int toku_db_env_txn_stat(DB_ENV * env, DB_TXN_STAT ** statp, u_int32_t flags) {
    HANDLE_PANICKED_ENV(env);
    statp=statp;flags=flags;
    return 1;
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR == 1
void toku_default_errcall(const char *errpfx, char *msg) {
#else
void toku_default_errcall(const DB_ENV *env, const char *errpfx, const char *msg) {
    env = env;
#endif
    fprintf(stderr, "YDB: %s: %s", errpfx, msg);
}

static int toku_txn_begin(DB_ENV * env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags);

int db_env_create(DB_ENV ** envp, u_int32_t flags) {
    if (flags!=0) return EINVAL;
    DB_ENV *MALLOC(result);
    if (result == 0)
        return ENOMEM;
    memset(result, 0, sizeof *result);
    result->err = toku_db_env_err;
    result->open = toku_db_env_open;
    result->close = toku_db_env_close;
    result->txn_checkpoint = toku_db_env_txn_checkpoint;
    result->log_flush = toku_db_env_log_flush;
    result->set_errcall = toku_db_env_set_errcall;
    result->set_errfile = toku_db_env_set_errfile;
    result->set_errpfx = toku_db_env_set_errpfx;
    //result->set_noticecall = toku_db_env_set_noticecall;
    result->set_flags = toku_db_env_set_flags;
    result->set_data_dir = toku_db_env_set_data_dir;
    result->set_tmp_dir = toku_db_env_set_tmp_dir;
    result->set_verbose = toku_db_env_set_verbose;
    result->set_lg_bsize = toku_db_env_set_lg_bsize;
    result->set_lg_dir = toku_db_env_set_lg_dir;
    result->set_lg_max = toku_db_env_set_lg_max;
    result->set_cachesize = toku_db_env_set_cachesize;
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3
    result->get_cachesize = toku_db_env_get_cachesize;
#endif
    result->set_lk_detect = toku_db_env_set_lk_detect;
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR <= 4
    result->set_lk_max = toku_db_env_set_lk_max;
#endif
    result->log_archive = toku_db_env_log_archive;
    result->txn_stat = toku_db_env_txn_stat;
    result->txn_begin = toku_txn_begin;

    MALLOC(result->i);
    if (result->i == 0) {
        toku_free(result);
        return ENOMEM;
    }
    memset(result->i, 0, sizeof *result->i);
    result->i->is_panicked=0;
    result->i->ref_count = 1;
    result->i->errcall = 0;
    result->i->errpfx = 0;
    result->i->errfile = 0;

    {
	int r = toku_logger_create(&result->i->logger);
	if (r!=0) {
	    toku_free(result->i);
	    toku_free(result);
	    return r;
	}
	assert(result->i->logger);
    }

    ydb_add_ref();
    *envp = result;
    return 0;
}

static int toku_db_txn_commit(DB_TXN * txn, u_int32_t flags) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    //notef("flags=%d\n", flags);
    int r;
    int nosync = (flags & DB_TXN_NOSYNC)!=0;
    flags &= ~DB_TXN_NOSYNC;
    if (!txn) return EINVAL;
    if (flags!=0) goto return_invalid;
    r = toku_logger_commit(txn->i->tokutxn, nosync);
    if (0) {
    return_invalid:
	r = EINVAL;
	toku_free(txn->i->tokutxn);
    }
    // Cleanup */
    if (txn->i)
        toku_free(txn->i);
    toku_free(txn);
    return r; // The txn is no good after the commit.
}

static u_int32_t toku_db_txn_id(DB_TXN * txn) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    barf();
    abort();
}

static TXNID next_txn = 0;

static int toku_txn_abort(DB_TXN * txn) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    return toku_logger_abort(txn->i->tokutxn);
}

static int toku_txn_begin(DB_ENV * env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags) {
    HANDLE_PANICKED_ENV(env);
    if (!toku_logger_is_open(env->i->logger)) return do_error(env, EINVAL, "Environment does not have logging enabled\n");
    flags=flags;
    DB_TXN *MALLOC(result);
    if (result == 0)
        return ENOMEM;
    memset(result, 0, sizeof *result);
    //notef("parent=%p flags=0x%x\n", stxn, flags);
    result->mgrp = env;
    result->abort = toku_txn_abort;
    result->commit = toku_db_txn_commit;
    result->id = toku_db_txn_id;
    MALLOC(result->i);
    assert(result->i);
    result->i->parent = stxn;
    int r = toku_logger_txn_begin(stxn ? stxn->i->tokutxn : 0, &result->i->tokutxn, next_txn++, env->i->logger);
    if (r != 0)
        return r;
    *txn = result;
    return 0;
}

#if 0
int txn_commit(DB_TXN * txn, u_int32_t flags) {
    fprintf(stderr, "%s:%d\n", __FILE__, __LINE__);
    return toku_logger_log_commit(txn->i->tokutxn);
}
#endif

int log_compare(const DB_LSN * a, const DB_LSN * b) {
    fprintf(stderr, "%s:%d log_compare(%p,%p)\n", __FILE__, __LINE__, a, b);
    abort();
}

static int maybe_do_associate_create (DB_TXN*txn, DB*primary, DB*secondary) {
    DBC *dbc;
    int r = secondary->cursor(secondary, txn, &dbc, 0);
    if (r!=0) return r;
    DBT key,data;
    r = dbc->c_get(dbc, &key, &data, DB_FIRST);
    {
	int r2=dbc->c_close(dbc);
	if (r!=DB_NOTFOUND) {
	    return r2;
	}
    }
    /* Now we know the secondary is empty. */
    r = primary->cursor(primary, txn, &dbc, 0);
    if (r!=0) return r;
    for (r = dbc->c_get(dbc, &key, &data, DB_FIRST);
	 r==0;
	 r = dbc->c_get(dbc, &key, &data, DB_NEXT)) {
	r = do_associated_inserts(txn, &key, &data, secondary);
	if (r!=0) {
	    dbc->c_close(dbc);
	    return r;
	}
    }
    return 0;
}

static int toku_db_associate (DB *primary, DB_TXN *txn, DB *secondary,
			      int (*callback)(DB *secondary, const DBT *key, const DBT *data, DBT *result),
			      u_int32_t flags) {
    HANDLE_PANICKED_DB(primary);
    HANDLE_PANICKED_DB(secondary);
    unsigned int brtflags;
    
    if (secondary->i->primary) return EINVAL; // The secondary already has a primary
    if (primary->i->primary)   return EINVAL; // The primary already has a primary

    toku_brt_get_flags(primary->i->brt, &brtflags);
    if (brtflags & TOKU_DB_DUPSORT) return EINVAL;  //The primary may not have duplicate keys.
    if (brtflags & TOKU_DB_DUP)     return EINVAL;  //The primary may not have duplicate keys.

    if (!list_empty(&secondary->i->associated)) return EINVAL; // The secondary is in some list (or it is a primary)
    assert(secondary->i->associate_callback==0);      // Something's wrong if this isn't null we made it this far.
    secondary->i->associate_callback = callback;
#ifdef DB_IMMUTABLE_KEY
    secondary->i->associate_is_immutable = (DB_IMMUTABLE_KEY&flags)!=0;
    flags &= ~DB_IMMUTABLE_KEY;
#else
    secondary->i->associate_is_immutable = 0;
#endif
    if (flags!=0 && flags!=DB_CREATE) return EINVAL; // after removing DB_IMMUTABLE_KEY the flags better be 0 or DB_CREATE
    list_push(&primary->i->associated, &secondary->i->associated);
    secondary->i->primary = primary;
    if (flags==DB_CREATE) {
	// To do this:  If the secondary is empty, then open a cursor on the primary.  Step through it all, doing the callbacks.
	// Then insert each callback result into the secondary.
	return maybe_do_associate_create(txn, primary, secondary);
    }
    return 0;
}

static int toku_db_close(DB * db, u_int32_t flags) {
    if (db->i->primary==0) {
	// It is a primary.  Unlink all the secondaries. */
	while (!list_empty(&db->i->associated)) {
	    assert(list_struct(list_head(&db->i->associated),
			       struct __toku_db_internal,
			       associated)->primary==db);
	    list_remove(list_head(&db->i->associated));
	}
    } else {
	// It is a secondary.  Remove it from the list, (which it must be in .*/
	if (!list_empty(&db->i->associated)) {
	    list_remove(&db->i->associated);
	}
    }
    flags=flags;
    int r = toku_close_brt(db->i->brt);
    if (r != 0)
        return r;
    // printf("%s:%d %d=__toku_db_close(%p)\n", __FILE__, __LINE__, r, db);
    int is_panicked = env_is_panicked(db->dbenv); // Even if panicked, let's close as much as we can.
    db_env_unref(db->dbenv);
    toku_free(db->i->database_name);
    toku_free(db->i->full_fname);
    toku_free(db->i);
    toku_free(db);
    ydb_unref();
    if (r==0 && is_panicked) return EINVAL;
    return r;
}

struct __toku_dbc_internal {
    BRT_CURSOR c;
    DB_TXN *txn;
};

static int verify_secondary_key(DB *secondary, DBT *pkey, DBT *data, DBT *skey) {
    int r = 0;
    DBT idx;

    assert(secondary->i->primary != 0);
    memset(&idx, 0, sizeof(idx));
    secondary->i->associate_callback(secondary, pkey, data, &idx);
    if (r==DB_DONOTINDEX) return DB_SECONDARY_BAD;
#ifdef DB_DBT_MULTIPLE
    if (idx.flags & DB_DBT_MULTIPLE) {
        return EINVAL; // We aren't ready for this
    }
#endif
	if (skey->size != idx.size || memcmp(skey->data, idx.data, idx.size) != 0) r = DB_SECONDARY_BAD;
    if (idx.flags & DB_DBT_APPMALLOC) {
    	free(idx.data);
    }
    return r;
}

static inline int keyeq(DBC *c, DBT *a, DBT *b) {
    DB *db = c->dbp;
    return db->i->brt->compare_fun(db, a, b) == 0;
}

static int toku_c_get_next_dup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = c->c_get(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0) goto finish;
    r = c->c_get(c, &nkey, &nval, DB_NEXT);
    if (r != 0) {
        int rr = c->c_get(c, &nkey, &nval, DB_LAST); assert(rr == 0); /* sticky */
        goto finish;
    }
    if (!keyeq(c, &currentkey, &nkey)) {
        int rr = c->c_get(c, &nkey, &nval, DB_PREV); assert(rr == 0); /* sticky */
        r = DB_NOTFOUND;
        goto finish;
    }
    r = c->c_get(c, key, data, DB_CURRENT);
finish:
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

#ifdef DB_PREV_DUP

static int toku_c_get_prev_dup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = c->c_get(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0) goto finish;
    r = c->c_get(c, &nkey, &nval, DB_PREV);
    if (r != 0) {
        int rr = c->c_get(c, &nkey, &nval, DB_FIRST); assert(rr == 0); /* sticky */
        goto finish;
    }
    if (!keyeq(c, &currentkey, &nkey)) {
        int rr = c->c_get(c, &nkey, &nval, DB_NEXT); assert(rr == 0); /* sticky */
        r = DB_NOTFOUND;
        goto finish;
    }
    r = c->c_get(c, key, data, DB_CURRENT);
finish:
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

#endif

static int toku_c_get_next_nodup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = c->c_get(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0)
        r = c->c_get(c, key, data, DB_FIRST);
    else {
        while (r == 0) {
            r = c->c_get(c, &nkey, &nval, DB_NEXT);
            if (r != 0) break;
            if (!keyeq(c, &currentkey, &nkey)) break;
        }
        if (r == 0) r = c->c_get(c, key, data, DB_CURRENT);
    }
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

static int toku_c_get_prev_nodup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = c->c_get(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0)
        r = c->c_get(c, key, data, DB_LAST);
    else {
        while (r == 0) {
            r = c->c_get(c, &nkey, &nval, DB_PREV);
            if (r != 0) break;
            if (!keyeq(c, &currentkey, &nkey)) break;
        }
        if (r == 0) r = c->c_get(c, key, data, DB_CURRENT);
    }
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

static int toku_c_get_noassociate(DBC * c, DBT * key, DBT * data, u_int32_t flag) {
    HANDLE_PANICKED_DB(c->dbp);
    int r;
    u_int32_t op = flag & DB_OPFLAGS_MASK;
    if (op == DB_NEXT_DUP)
        r = toku_c_get_next_dup(c, key, data);
    else if (op == DB_NEXT_NODUP)
        r = toku_c_get_next_nodup(c, key, data);
    else if (op == DB_PREV_NODUP) <commit_msg>test next/prev_dup/nodup.  closes #259<commit_after>/* -*- mode: C; c-basic-offset: 4 -*- */
#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."

#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."

const char *toku_patent_string = "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it.";
const char *toku_copyright_string = "Copyright (c) 2007 Tokutek Inc.  All rights reserved.";

#include <assert.h>
#include <errno.h>
#include <limits.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <ctype.h>
#include <unistd.h>
#include <libgen.h>

#include "ydb-internal.h"

#include "brt-internal.h"
#include "cachetable.h"
#include "log.h"
#include "memory.h"

struct __toku_db_txn_internal {
    //TXNID txnid64; /* A sixty-four bit txn id. */
    TOKUTXN tokutxn;
    DB_TXN *parent;
};

static char *construct_full_name(const char *dir, const char *fname);
static int do_associated_inserts (DB_TXN *txn, DBT *key, DBT *data, DB *secondary);
    
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR == 1
typedef void (*toku_env_errcall_t)(const char *, char *);
#elif DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3
typedef void (*toku_env_errcall_t)(const DB_ENV *, const char *, const char *);
#else
#error
#endif

struct __toku_db_env_internal {
    int is_panicked;
    int ref_count;
    u_int32_t open_flags;
    int open_mode;
    toku_env_errcall_t errcall;
    void *errfile;
    const char *errpfx;
    char *dir;                  /* A malloc'd copy of the directory. */
    char *tmp_dir;
    char *lg_dir;
    char **data_dirs;
    u_int32_t n_data_dirs;
    //void (*noticecall)(DB_ENV *, db_notices);
    unsigned long cachetable_size;
    CACHETABLE cachetable;
    TOKULOGGER logger;
};

// If errcall is set, call it with the format string and optionally the stderrstring (if include_stderrstring).  The prefix is passed as a separate argument.
// If errfile is set, print to the errfile: prefix, fmt string, maybe include the stderr string.
// Both errcall and errfile may be called.
// If errfile is not set and errcall is not set, the use stderr as the errfile.
void toku_do_error_all_cases(const DB_ENV * env, int error, int include_stderrstring, int use_stderr_if_nothing_else, const char *fmt, va_list ap) {
    if (env->i->errcall) {
	// errcall gets prefix sent separately
	// the error message is the printf message, maybe followed by ": " and the dbstrerror (if include_stderrstring is set)
	char buf [4000];
	int count=0;
	if (fmt) {
	    count=vsnprintf(buf, sizeof(buf), fmt, ap);
	}
	if (include_stderrstring) {
	    count+=snprintf(&buf[count], sizeof(buf)-count, ": %s", db_strerror(error));
	}
	env->i->errcall(env, env->i->errpfx, buf);
    }
    {
	FILE *efile=env->i->errfile;
	if (efile==0 && env->i->errcall==0 && use_stderr_if_nothing_else) {
	    efile = stderr;
	}
	if (efile) {
	    if (env->i->errpfx) fprintf(efile, "%s: ", env->i->errpfx);
	    vfprintf(efile, fmt, ap);
	    if (include_stderrstring) {
		fprintf(efile, ": %s", db_strerror(error));
	    }
	}
    }
}


static int env_is_panicked(DB_ENV *dbenv) {
    if (dbenv==0) return 0;
    return dbenv->i->is_panicked || toku_logger_panicked(dbenv->i->logger);
}
#define HANDLE_PANICKED_ENV(env) ({ if (env_is_panicked(env)) return EINVAL; })
#define HANDLE_PANICKED_DB(db) HANDLE_PANICKED_ENV(db->dbenv)

// Handle all the error cases (but don't do the default thing.)
static int do_error (DB_ENV *dbenv, int error, const char *string, ...) {
    if (toku_logger_panicked(dbenv->i->logger)) dbenv->i->is_panicked=1;
    va_list ap;
    va_start(ap, string);
    toku_do_error_all_cases(dbenv, error, 1, 0, string, ap);
    va_end(ap);
    return error;
}


static void toku_db_env_err(const DB_ENV * env, int error, const char *fmt, ...) {
    va_list ap;
    va_start(ap, fmt);
    toku_do_error_all_cases(env, error, 1, 1, fmt, ap);
    va_end(ap);
}

#define barf() ({ fprintf(stderr, "YDB: BARF %s:%d in %s\n", __FILE__, __LINE__, __func__); })
#define barff(fmt,...) ({ fprintf(stderr, "YDB: BARF %s:%d in %s, ", __FILE__, __LINE__, __func__); fprintf(stderr, fmt, __VA_ARGS__); })
#define note() ({ fprintf(svtderr, "YDB: Note %s:%d in %s\n", __FILE__, __LINE__, __func__); })
#define notef(fmt,...) ({ fprintf(stderr, "YDB: Note %s:%d in %s, ", __FILE__, __LINE__, __func__); fprintf(stderr, fmt, __VA_ARGS__); })

#if 0
static void print_flags(u_int32_t flags) {
    u_int32_t gotit = 0;
    int doneone = 0;
#define doit(flag) if (flag & flags) { if (doneone) fprintf(stderr, " | "); fprintf(stderr, "%s", #flag);  doneone=1; gotit|=flag; }
    doit(DB_INIT_LOCK);
    doit(DB_INIT_LOG);
    doit(DB_INIT_MPOOL);
    doit(DB_INIT_TXN);
    doit(DB_CREATE);
    doit(DB_THREAD);
    doit(DB_RECOVER);
    doit(DB_PRIVATE);
    if (gotit != flags)
        fprintf(stderr, "  flags 0x%x not accounted for", flags & ~gotit);
    fprintf(stderr, "\n");
}
#endif

/* TODO make these thread safe */

/* a count of the open env handles */
static int toku_ydb_refs = 0;

static void ydb_add_ref() {
    toku_ydb_refs += 1;
}

static void ydb_unref() {
    assert(toku_ydb_refs > 0);
    toku_ydb_refs -= 1;
    if (toku_ydb_refs == 0) {
        /* call global destructors */
        toku_malloc_cleanup();
    }
}

static void db_env_add_ref(DB_ENV *env) {
    env->i->ref_count += 1;
}

static void db_env_unref(DB_ENV *env) {
    env->i->ref_count -= 1;
    if (env->i->ref_count == 0)
        env->close(env, 0);
}

static inline int db_env_opened(DB_ENV *env) {
    return env->i->cachetable != 0;
}

static inline int db_opened(DB *db) {
    return db->i->full_fname != 0;
}

static int db_env_parse_config_line(DB_ENV* dbenv, char *command, char *value) {
    int r;
    
    if (!strcmp(command, "set_data_dir")) {
        r = dbenv->set_data_dir(dbenv, value);
    }
    else if (!strcmp(command, "set_tmp_dir")) {
        r = dbenv->set_tmp_dir(dbenv, value);
    }
    else if (!strcmp(command, "set_lg_dir")) {
        r = dbenv->set_lg_dir(dbenv, value);
    }
    else r = -1;
        
    return r;
}

static int db_env_read_config(DB_ENV *env) {
    HANDLE_PANICKED_ENV(env);
    const char* config_name = "DB_CONFIG";
    char* full_name = NULL;
    char* linebuffer = NULL;
    int buffersize;
    FILE* fp = NULL;
    int r = 0;
    int r2 = 0;
    char* command;
    char* value;
    
    full_name = construct_full_name(env->i->dir, config_name);
    if (full_name == 0) {
        r = ENOMEM;
        goto cleanup;
    }
    if ((fp = fopen(full_name, "r")) == NULL) {
        //Config file is optional.
        if (errno == ENOENT) {
            r = EXIT_SUCCESS;
            goto cleanup;
        }
        r = errno;
        goto cleanup;
    }
    //Read each line, applying configuration parameters.
    //After ignoring leading white space, skip any blank lines
    //or comments (starts with #)
    //Command contains no white space.  Value may contain whitespace.
    int linenumber;
    int ch = '\0';
    BOOL eof = FALSE;
    char* temp;
    char* end;
    int index;
    
    buffersize = 1<<10; //1KB
    linebuffer = toku_malloc(buffersize);
    if (!linebuffer) {
        r = ENOMEM;
        goto cleanup;
    }
    for (linenumber = 1; !eof; linenumber++) {
        /* Read a single line. */
        for (index = 0; TRUE; index++) {
            if ((ch = getc(fp)) == EOF) {
                eof = TRUE;
                if (ferror(fp)) {
                    /* Throw away current line and print warning. */
                    r = errno;
                    goto readerror;
                }
                break;
            }
            if (ch == '\n') break;
            if (index + 1 >= buffersize) {
                //Double the buffer.
                buffersize *= 2;
                linebuffer = toku_realloc(linebuffer, buffersize);
                if (!linebuffer) {
                    r = ENOMEM;
                    goto cleanup;
                }
            }
            linebuffer[index] = ch;
        }
        linebuffer[index] = '\0';
        end = &linebuffer[index];

        /* Separate the line into command/value */
        command = linebuffer;
        //Strip leading spaces.
        while (isspace(*command) && command < end) command++;
        //Find end of command.
        temp = command;
        while (!isspace(*temp) && temp < end) temp++;
        *temp++ = '\0'; //Null terminate command.
        value = temp;
        //Strip leading spaces.
        while (isspace(*value) && value < end) value++;
        if (value < end) {
            //Strip trailing spaces.
            temp = end;
            while (isspace(*(temp-1))) temp--;
            //Null terminate value.
            *temp = '\0';
        }
        //Parse the line.
        if (strlen(command) == 0 || command[0] == '#') continue; //Ignore Comments.
        r = db_env_parse_config_line(env, command, value < end ? value : "");
        if (r != 0) goto parseerror;
    }
    if (0) {
readerror:
        do_error(env, r, "Error reading from DB_CONFIG:%d.\n", linenumber);
    }
    if (0) {
parseerror:
        do_error(env, r, "Error parsing DB_CONFIG:%d.\n", linenumber);
    }
cleanup:
    if (full_name) toku_free(full_name);
    if (linebuffer) toku_free(linebuffer);
    if (fp) r2 = fclose(fp);
    return r ? r : r2;
}

static int toku_db_env_open(DB_ENV * env, const char *home, u_int32_t flags, int mode) {
    HANDLE_PANICKED_ENV(env);
    int r;

    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "The environment is already open\n");
    }

    if ((flags & DB_USE_ENVIRON) && (flags & DB_USE_ENVIRON_ROOT)) {
	return do_error(env, EINVAL, "DB_USE_ENVIRON and DB_USE_ENVIRON_ROOT are incompatible flags\n");
    }

    if (home) {
        if ((flags & DB_USE_ENVIRON) || (flags & DB_USE_ENVIRON_ROOT)) {
	    return do_error(env, EINVAL, "DB_USE_ENVIRON and DB_USE_ENVIRON_ROOT are incompatible with specifying a home\n");
	}
    }
    else if ((flags & DB_USE_ENVIRON) ||
             ((flags & DB_USE_ENVIRON_ROOT) && geteuid() == 0)) home = getenv("DB_HOME");

    if (!home) home = ".";

	// Verify that the home exists.
	{
	struct stat buf;
	r = stat(home, &buf);
	if (r!=0) {
	    return do_error(env, errno, "Error from stat(\"%s\",...)\n", home);
	}
    }



    if (!(flags & DB_PRIVATE)) {
        // This means that we don't have to do anything with shared memory.  
        // And that's good enough for mysql. 
	return do_error(env, EINVAL, "TokuDB requires DB_PRIVATE when opening an env\n");
    }

    if (env->i->dir)
        toku_free(env->i->dir);
    env->i->dir = toku_strdup(home);
    if (env->i->dir == 0) {
	return do_error(env, ENOMEM, "Out of memory\n");
    }
    if (0) {
        died1:
        toku_free(env->i->dir);
        env->i->dir = NULL;
        return r;
    }
    if ((r = db_env_read_config(env)) != 0) {
	goto died1;
    }
    
    env->i->open_flags = flags;
    env->i->open_mode = mode;

    if (flags & (DB_INIT_TXN | DB_INIT_LOG)) {
        char* full_dir = NULL;
        if (env->i->lg_dir) full_dir = construct_full_name(env->i->dir, env->i->lg_dir);
	assert(env->i->logger);
        r = toku_logger_open(full_dir ? full_dir : env->i->dir, env->i->logger);
        if (full_dir) toku_free(full_dir);
	if (r!=0) {
	    do_error(env, r, "Could not open logger\n");
	died2:
	    toku_logger_close(&env->i->logger);
	    goto died1;
	}
    }

    r = toku_brt_create_cachetable(&env->i->cachetable, env->i->cachetable_size, ZERO_LSN, env->i->logger);
    if (r!=0) goto died2;
    return 0;
}

static int toku_db_env_close(DB_ENV * env, u_int32_t flags) {
    // Even if the env is panicedk, try to close as much as we can.
    int is_panicked = env_is_panicked(env);
    int r0=0,r1=0;
    if (env->i->cachetable)
        r0=toku_cachetable_close(&env->i->cachetable);
    if (env->i->logger)
        r1=toku_logger_close(&env->i->logger);
    if (env->i->data_dirs) {
        u_int32_t i;
        assert(env->i->n_data_dirs > 0);
        for (i = 0; i < env->i->n_data_dirs; i++) {
            toku_free(env->i->data_dirs[i]);
        }
        toku_free(env->i->data_dirs);
    }
    if (env->i->lg_dir)
        toku_free(env->i->lg_dir);
    if (env->i->tmp_dir)
        toku_free(env->i->tmp_dir);
    toku_free(env->i->dir);
    toku_free(env->i);
    toku_free(env);
    ydb_unref();
    if (flags!=0) return EINVAL;
    if (r0) return r0;
    if (r1) return r1;
    if (is_panicked) return EINVAL;
    return 0;
}

static int toku_db_env_log_archive(DB_ENV * env, char **list[], u_int32_t flags) {
    env=env; flags=flags; // Suppress compiler warnings.
    *list = NULL;
    return 0;
}

static int toku_db_env_log_flush(DB_ENV * env, const DB_LSN * lsn) {
    HANDLE_PANICKED_ENV(env);
    env=env; lsn=lsn;
    barf();
    return 1;
}

static int toku_db_env_set_cachesize(DB_ENV * env, u_int32_t gbytes, u_int32_t bytes, int ncache __attribute__((__unused__))) {
    HANDLE_PANICKED_ENV(env);
    u_int64_t cs64 = ((u_int64_t) gbytes << 30) + bytes;
    unsigned long cs = cs64;
    if (cs64 > cs)
        return EINVAL;
    env->i->cachetable_size = cs;
    return 0;
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3

static int toku_db_env_get_cachesize(DB_ENV * env, u_int32_t *gbytes, u_int32_t *bytes, int *ncache) {
    HANDLE_PANICKED_ENV(env);
    *gbytes = env->i->cachetable_size >> 30;
    *bytes = env->i->cachetable_size & ((1<<30)-1);
    *ncache = 1;
    return 0;
}

#endif

static int toku_db_env_set_data_dir(DB_ENV * env, const char *dir) {
    HANDLE_PANICKED_ENV(env);
    u_int32_t i;
    int r;
    char** temp;
    char* new_dir;
    
    if (db_env_opened(env) || !dir) {
	return do_error(env, EINVAL, "You cannot set the data dir after opening the env\n");
    }
    
    if (env->i->data_dirs) {
        assert(env->i->n_data_dirs > 0);
        for (i = 0; i < env->i->n_data_dirs; i++) {
            if (!strcmp(dir, env->i->data_dirs[i])) {
                //It is already in the list.  We're done.
                return 0;
            }
        }
    }
    else assert(env->i->n_data_dirs == 0);
    new_dir = toku_strdup(dir);
    if (0) {
        died1:
        toku_free(new_dir);
        return r;
    }
    if (new_dir==NULL) {
	assert(errno == ENOMEM);
	return do_error(env, errno, "Out of memory\n");
    }
    temp = (char**) toku_realloc(env->i->data_dirs, (1 + env->i->n_data_dirs) * sizeof(char*));
    if (temp==NULL) {assert(errno == ENOMEM); r = ENOMEM; goto died1;}
    else env->i->data_dirs = temp;
    env->i->data_dirs[env->i->n_data_dirs] = new_dir;
    env->i->n_data_dirs++;
    return 0;
}

static void toku_db_env_set_errcall(DB_ENV * env, toku_env_errcall_t errcall) {
    env->i->errcall = errcall;
}

static void toku_db_env_set_errfile(DB_ENV*env, FILE*errfile) {
    env->i->errfile = errfile;
}

static void toku_db_env_set_errpfx(DB_ENV * env, const char *errpfx) {
    env->i->errpfx = errpfx;
}

static int toku_db_env_set_flags(DB_ENV * env, u_int32_t flags, int onoff) {
    HANDLE_PANICKED_ENV(env);
    if (flags != 0 && onoff) {
	return do_error(env, EINVAL, "TokuDB does not (yet) support any nonzero ENV flags\n");
    }
    return 0;
}

static int toku_db_env_set_lg_bsize(DB_ENV * env, u_int32_t bsize) {
    HANDLE_PANICKED_ENV(env);
    bsize=bsize;
    return do_error(env, EINVAL, "TokuDB does not (yet) support ENV->set_lg_bsize\n");
}

static int toku_db_env_set_lg_dir(DB_ENV * env, const char *dir) {
    HANDLE_PANICKED_ENV(env);
    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "Cannot set log dir after opening the env\n");
    }

    if (env->i->lg_dir) toku_free(env->i->lg_dir);
    if (dir) {
        env->i->lg_dir = toku_strdup(dir);
        if (!env->i->lg_dir) {
	    return do_error(env, ENOMEM, "Out of memory\n");
	}
    }
    else env->i->lg_dir = NULL;
    return 0;
}

static int toku_db_env_set_lg_max(DB_ENV * env, u_int32_t lg_max) {
    HANDLE_PANICKED_ENV(env);
    lg_max=lg_max;
    return do_error(env, EINVAL, "TokuDB does not (yet) support set_lg_max\n");
}

static int toku_db_env_set_lk_detect(DB_ENV * env, u_int32_t detect) {
    HANDLE_PANICKED_ENV(env);
    detect=detect;
    return do_error(env, EINVAL, "TokuDB does not (yet) support set_lk_detect\n");
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR <= 4
static int toku_db_env_set_lk_max(DB_ENV * env, u_int32_t lk_max) {
    HANDLE_PANICKED_ENV(env);
    lk_max=lk_max;
    return 0;
}
#endif

//void __toku_db_env_set_noticecall (DB_ENV *env, void (*noticecall)(DB_ENV *, db_notices)) {
//    env->i->noticecall = noticecall;
//}

static int toku_db_env_set_tmp_dir(DB_ENV * env, const char *tmp_dir) {
    HANDLE_PANICKED_ENV(env);
    if (db_env_opened(env)) {
	return do_error(env, EINVAL, "Cannot set the tmp dir after opening an env\n");
    }
    if (!tmp_dir) {
	return do_error(env, EINVAL, "Tmp dir bust be non-null\n");
    }
    if (env->i->tmp_dir)
        toku_free(env->i->tmp_dir);
    env->i->tmp_dir = toku_strdup(tmp_dir);
    return env->i->tmp_dir ? 0 : ENOMEM;
}

static int toku_db_env_set_verbose(DB_ENV * env, u_int32_t which, int onoff) {
    HANDLE_PANICKED_ENV(env);
    which=which; onoff=onoff;
    return 1;
}

static int toku_db_env_txn_checkpoint(DB_ENV * env, u_int32_t kbyte, u_int32_t min, u_int32_t flags) {
    env=env; kbyte=kbyte; min=min; flags=flags;
    return 0;
}

static int toku_db_env_txn_stat(DB_ENV * env, DB_TXN_STAT ** statp, u_int32_t flags) {
    HANDLE_PANICKED_ENV(env);
    statp=statp;flags=flags;
    return 1;
}

#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR == 1
void toku_default_errcall(const char *errpfx, char *msg) {
#else
void toku_default_errcall(const DB_ENV *env, const char *errpfx, const char *msg) {
    env = env;
#endif
    fprintf(stderr, "YDB: %s: %s", errpfx, msg);
}

static int toku_txn_begin(DB_ENV * env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags);

int db_env_create(DB_ENV ** envp, u_int32_t flags) {
    if (flags!=0) return EINVAL;
    DB_ENV *MALLOC(result);
    if (result == 0)
        return ENOMEM;
    memset(result, 0, sizeof *result);
    result->err = toku_db_env_err;
    result->open = toku_db_env_open;
    result->close = toku_db_env_close;
    result->txn_checkpoint = toku_db_env_txn_checkpoint;
    result->log_flush = toku_db_env_log_flush;
    result->set_errcall = toku_db_env_set_errcall;
    result->set_errfile = toku_db_env_set_errfile;
    result->set_errpfx = toku_db_env_set_errpfx;
    //result->set_noticecall = toku_db_env_set_noticecall;
    result->set_flags = toku_db_env_set_flags;
    result->set_data_dir = toku_db_env_set_data_dir;
    result->set_tmp_dir = toku_db_env_set_tmp_dir;
    result->set_verbose = toku_db_env_set_verbose;
    result->set_lg_bsize = toku_db_env_set_lg_bsize;
    result->set_lg_dir = toku_db_env_set_lg_dir;
    result->set_lg_max = toku_db_env_set_lg_max;
    result->set_cachesize = toku_db_env_set_cachesize;
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR >= 3
    result->get_cachesize = toku_db_env_get_cachesize;
#endif
    result->set_lk_detect = toku_db_env_set_lk_detect;
#if DB_VERSION_MAJOR == 4 && DB_VERSION_MINOR <= 4
    result->set_lk_max = toku_db_env_set_lk_max;
#endif
    result->log_archive = toku_db_env_log_archive;
    result->txn_stat = toku_db_env_txn_stat;
    result->txn_begin = toku_txn_begin;

    MALLOC(result->i);
    if (result->i == 0) {
        toku_free(result);
        return ENOMEM;
    }
    memset(result->i, 0, sizeof *result->i);
    result->i->is_panicked=0;
    result->i->ref_count = 1;
    result->i->errcall = 0;
    result->i->errpfx = 0;
    result->i->errfile = 0;

    {
	int r = toku_logger_create(&result->i->logger);
	if (r!=0) {
	    toku_free(result->i);
	    toku_free(result);
	    return r;
	}
	assert(result->i->logger);
    }

    ydb_add_ref();
    *envp = result;
    return 0;
}

static int toku_db_txn_commit(DB_TXN * txn, u_int32_t flags) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    //notef("flags=%d\n", flags);
    int r;
    int nosync = (flags & DB_TXN_NOSYNC)!=0;
    flags &= ~DB_TXN_NOSYNC;
    if (!txn) return EINVAL;
    if (flags!=0) goto return_invalid;
    r = toku_logger_commit(txn->i->tokutxn, nosync);
    if (0) {
    return_invalid:
	r = EINVAL;
	toku_free(txn->i->tokutxn);
    }
    // Cleanup */
    if (txn->i)
        toku_free(txn->i);
    toku_free(txn);
    return r; // The txn is no good after the commit.
}

static u_int32_t toku_db_txn_id(DB_TXN * txn) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    barf();
    abort();
}

static TXNID next_txn = 0;

static int toku_txn_abort(DB_TXN * txn) {
    HANDLE_PANICKED_ENV(txn->mgrp);
    return toku_logger_abort(txn->i->tokutxn);
}

static int toku_txn_begin(DB_ENV * env, DB_TXN * stxn, DB_TXN ** txn, u_int32_t flags) {
    HANDLE_PANICKED_ENV(env);
    if (!toku_logger_is_open(env->i->logger)) return do_error(env, EINVAL, "Environment does not have logging enabled\n");
    flags=flags;
    DB_TXN *MALLOC(result);
    if (result == 0)
        return ENOMEM;
    memset(result, 0, sizeof *result);
    //notef("parent=%p flags=0x%x\n", stxn, flags);
    result->mgrp = env;
    result->abort = toku_txn_abort;
    result->commit = toku_db_txn_commit;
    result->id = toku_db_txn_id;
    MALLOC(result->i);
    assert(result->i);
    result->i->parent = stxn;
    int r = toku_logger_txn_begin(stxn ? stxn->i->tokutxn : 0, &result->i->tokutxn, next_txn++, env->i->logger);
    if (r != 0)
        return r;
    *txn = result;
    return 0;
}

#if 0
int txn_commit(DB_TXN * txn, u_int32_t flags) {
    fprintf(stderr, "%s:%d\n", __FILE__, __LINE__);
    return toku_logger_log_commit(txn->i->tokutxn);
}
#endif

int log_compare(const DB_LSN * a, const DB_LSN * b) {
    fprintf(stderr, "%s:%d log_compare(%p,%p)\n", __FILE__, __LINE__, a, b);
    abort();
}

static int maybe_do_associate_create (DB_TXN*txn, DB*primary, DB*secondary) {
    DBC *dbc;
    int r = secondary->cursor(secondary, txn, &dbc, 0);
    if (r!=0) return r;
    DBT key,data;
    r = dbc->c_get(dbc, &key, &data, DB_FIRST);
    {
	int r2=dbc->c_close(dbc);
	if (r!=DB_NOTFOUND) {
	    return r2;
	}
    }
    /* Now we know the secondary is empty. */
    r = primary->cursor(primary, txn, &dbc, 0);
    if (r!=0) return r;
    for (r = dbc->c_get(dbc, &key, &data, DB_FIRST);
	 r==0;
	 r = dbc->c_get(dbc, &key, &data, DB_NEXT)) {
	r = do_associated_inserts(txn, &key, &data, secondary);
	if (r!=0) {
	    dbc->c_close(dbc);
	    return r;
	}
    }
    return 0;
}

static int toku_db_associate (DB *primary, DB_TXN *txn, DB *secondary,
			      int (*callback)(DB *secondary, const DBT *key, const DBT *data, DBT *result),
			      u_int32_t flags) {
    HANDLE_PANICKED_DB(primary);
    HANDLE_PANICKED_DB(secondary);
    unsigned int brtflags;
    
    if (secondary->i->primary) return EINVAL; // The secondary already has a primary
    if (primary->i->primary)   return EINVAL; // The primary already has a primary

    toku_brt_get_flags(primary->i->brt, &brtflags);
    if (brtflags & TOKU_DB_DUPSORT) return EINVAL;  //The primary may not have duplicate keys.
    if (brtflags & TOKU_DB_DUP)     return EINVAL;  //The primary may not have duplicate keys.

    if (!list_empty(&secondary->i->associated)) return EINVAL; // The secondary is in some list (or it is a primary)
    assert(secondary->i->associate_callback==0);      // Something's wrong if this isn't null we made it this far.
    secondary->i->associate_callback = callback;
#ifdef DB_IMMUTABLE_KEY
    secondary->i->associate_is_immutable = (DB_IMMUTABLE_KEY&flags)!=0;
    flags &= ~DB_IMMUTABLE_KEY;
#else
    secondary->i->associate_is_immutable = 0;
#endif
    if (flags!=0 && flags!=DB_CREATE) return EINVAL; // after removing DB_IMMUTABLE_KEY the flags better be 0 or DB_CREATE
    list_push(&primary->i->associated, &secondary->i->associated);
    secondary->i->primary = primary;
    if (flags==DB_CREATE) {
	// To do this:  If the secondary is empty, then open a cursor on the primary.  Step through it all, doing the callbacks.
	// Then insert each callback result into the secondary.
	return maybe_do_associate_create(txn, primary, secondary);
    }
    return 0;
}

static int toku_db_close(DB * db, u_int32_t flags) {
    if (db->i->primary==0) {
	// It is a primary.  Unlink all the secondaries. */
	while (!list_empty(&db->i->associated)) {
	    assert(list_struct(list_head(&db->i->associated),
			       struct __toku_db_internal,
			       associated)->primary==db);
	    list_remove(list_head(&db->i->associated));
	}
    } else {
	// It is a secondary.  Remove it from the list, (which it must be in .*/
	if (!list_empty(&db->i->associated)) {
	    list_remove(&db->i->associated);
	}
    }
    flags=flags;
    int r = toku_close_brt(db->i->brt);
    if (r != 0)
        return r;
    // printf("%s:%d %d=__toku_db_close(%p)\n", __FILE__, __LINE__, r, db);
    int is_panicked = env_is_panicked(db->dbenv); // Even if panicked, let's close as much as we can.
    db_env_unref(db->dbenv);
    toku_free(db->i->database_name);
    toku_free(db->i->full_fname);
    toku_free(db->i);
    toku_free(db);
    ydb_unref();
    if (r==0 && is_panicked) return EINVAL;
    return r;
}

struct __toku_dbc_internal {
    BRT_CURSOR c;
    DB_TXN *txn;
};

static int verify_secondary_key(DB *secondary, DBT *pkey, DBT *data, DBT *skey) {
    int r = 0;
    DBT idx;

    assert(secondary->i->primary != 0);
    memset(&idx, 0, sizeof(idx));
    secondary->i->associate_callback(secondary, pkey, data, &idx);
    if (r==DB_DONOTINDEX) return DB_SECONDARY_BAD;
#ifdef DB_DBT_MULTIPLE
    if (idx.flags & DB_DBT_MULTIPLE) {
        return EINVAL; // We aren't ready for this
    }
#endif
	if (skey->size != idx.size || memcmp(skey->data, idx.data, idx.size) != 0) r = DB_SECONDARY_BAD;
    if (idx.flags & DB_DBT_APPMALLOC) {
    	free(idx.data);
    }
    return r;
}

static int toku_c_get_noassociate(DBC * c, DBT * key, DBT * data, u_int32_t flag);

static inline int keyeq(DBC *c, DBT *a, DBT *b) {
    DB *db = c->dbp;
    return db->i->brt->compare_fun(db, a, b) == 0;
}

static int toku_c_get_next_dup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = toku_c_get_noassociate(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0) goto finish;
    r = toku_c_get_noassociate(c, &nkey, &nval, DB_NEXT);
    if (r != 0) {
        int rr = toku_c_get_noassociate(c, &nkey, &nval, DB_LAST); assert(rr == 0); /* sticky */
        goto finish;
    }
    if (!keyeq(c, &currentkey, &nkey)) {
        int rr = toku_c_get_noassociate(c, &nkey, &nval, DB_PREV); assert(rr == 0); /* sticky */
        r = DB_NOTFOUND;
        goto finish;
    }
    r = toku_c_get_noassociate(c, key, data, DB_CURRENT);
finish:
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

#ifdef DB_PREV_DUP

static int toku_c_get_prev_dup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = toku_c_get_noassociate(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0) goto finish;
    r = toku_c_get_noassociate(c, &nkey, &nval, DB_PREV);
    if (r != 0) {
        int rr = toku_c_get_noassociate(c, &nkey, &nval, DB_FIRST); assert(rr == 0); /* sticky */
        goto finish;
    }
    if (!keyeq(c, &currentkey, &nkey)) {
        int rr = toku_c_get_noassociate(c, &nkey, &nval, DB_NEXT); assert(rr == 0); /* sticky */
        r = DB_NOTFOUND;
        goto finish;
    }
    r = toku_c_get_noassociate(c, key, data, DB_CURRENT);
finish:
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

#endif

static int toku_c_get_next_nodup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = toku_c_get_noassociate(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0)
        r = toku_c_get_noassociate(c, key, data, DB_FIRST);
    else {
        while (r == 0) {
            r = toku_c_get_noassociate(c, &nkey, &nval, DB_NEXT);
            if (r != 0) break;
            if (!keyeq(c, &currentkey, &nkey)) break;
        }
        if (r == 0) r = toku_c_get_noassociate(c, key, data, DB_CURRENT);
    }
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

static int toku_c_get_prev_nodup(DBC *c, DBT *key, DBT *data) {
    int r;
    DBT currentkey; memset(&currentkey, 0, sizeof currentkey); currentkey.flags = DB_DBT_REALLOC;
    DBT currentval; memset(&currentval, 0, sizeof currentval); currentval.flags = DB_DBT_REALLOC;
    DBT nkey; memset(&nkey, 0, sizeof nkey); nkey.flags = DB_DBT_REALLOC;
    DBT nval; memset(&nval, 0, sizeof nval); nval.flags = DB_DBT_REALLOC;
    r = toku_c_get_noassociate(c, &currentkey, &currentval, DB_CURRENT+256); 
    if (r != 0)
        r = toku_c_get_noassociate(c, key, data, DB_LAST);
    else {
        while (r == 0) {
            r = toku_c_get_noassociate(c, &nkey, &nval, DB_PREV);
            if (r != 0) break;
            if (!keyeq(c, &currentkey, &nkey)) break;
        }
        if (r == 0) r = toku_c_get_noassociate(c, key, data, DB_CURRENT);
    }
    if (nkey.data) toku_free(nkey.data);
    if (nval.data) toku_free(nval.data);
    if (currentkey.data) toku_free(currentkey.data);
    if (currentval.data) toku_free(currentval.data);
    return r;
}

static int toku_c_get_noassociate(DBC * c, DBT * key, DBT * data, u_int32_t flag) {
    HANDLE_PANICKED_DB(c->dbp);
    int r;
    u_int32_t op = flag & DB_OPFLAGS_MASK;
    if (op == DB_NEXT_DUP)
        r = toku_c_get_next_dup(c, key, data);
    else if (op == DB_NEXT_NODUP)
        r = toku_c_get_next_nodup(c, key, data);
    else if (op == DB_PREV_NODUP) 
{'fraction_non_alphanumeric': 0.13977191163569663, 'fraction_numerical': 0.010256001276018821, 'mean_word_length': 3.043859649122807, 'pattern_counts': {'":': 6, '<': 59, '<?xml version=': 0, '>': 651, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3538028', 'n_tokens_mistral': 24939, 'n_tokens_neox': 24149, 'n_words': 5695}