datastore_db_trans.c

一个C语言写的快速贝叶斯垃圾邮件过滤工具

    flags |= DB_CREATE | dbenv_defflags;

    ret = env->dbe->open(env->dbe, bfp->dirname, flags, DS_MODE);
    if (ret != 0) {
	env->dbe->close(env->dbe, 0);
	print_error(__FILE__, __LINE__, "DB_ENV->open, err: %d, %s", ret, db_strerror(ret));
	switch (ret) {
	    case DB_RUNRECOVERY:
		diag_dbeopen(flags, bfp);
		break;
	    case EINVAL:
		fprintf(stderr, "\n"
			"If you have just got a message that only private environments are supported,\n"
			"your Berkeley DB %d.%d was not configured properly.\n"
			"Bogofilter requires shared environments to support Berkeley DB transactions.\n",
			DB_VERSION_MAJOR, DB_VERSION_MINOR);
		fprintf(stderr,
			"Reconfigure and recompile Berkeley DB with the right mutex interface,\n"
			"see the docs/ref/build_unix/conf.html file that comes with your db source code.\n"
			"This can happen when the DB library was compiled with POSIX threads\n"
			"but your system does not support NPTL.\n");
		break;
	}

	exit(EX_ERROR);
    }

    if (DEBUG_DATABASE(1))
	fprintf(dbgout, "DB_ENV->open(home=%s)\n", bfp->dirname);

    return env;
}

static void dbx_cleanup(dbe_t *env)
{
    dbx_cleanup_lite(env);
}

/* close the environment, but do not release locks */
static void dbx_cleanup_lite(dbe_t *env)
{
    if (env) {
	if (env->dbe) {
	    int ret;

	    /* checkpoint if more than 64 kB of logs have been written
	     * or 120 min have passed since the previous checkpoint */
	    /*                                kB  min flags */
	    ret = BF_TXN_CHECKPOINT(env->dbe, 64, 120, 0);
	    ret = dbx_sync(env->dbe, ret);
	    if (ret)
		print_error(__FILE__, __LINE__, "DBE->dbx_checkpoint err: %d, %s", ret, db_strerror(ret));

	    if (db_log_autoremove)
		dbe_env_purgelogs(env->dbe);

	    ret = env->dbe->close(env->dbe, 0);
	    if (DEBUG_DATABASE(1) || ret)
		fprintf(dbgout, "DB_ENV->close(%p): %s\n", (void *)env->dbe,
			db_strerror(ret));
	    clear_lock();
	    if (lockfd >= 0)
		close(lockfd); /* release locks */
	}

	xfree(env->directory);
	xfree(env);
    }
}

static int dbx_sync(DB_ENV *dbe, int ret)
{
#if DB_AT_LEAST(3,0) && DB_AT_MOST(4,0)
    /* flush dirty pages in buffer pool */
    while (ret == DB_INCOMPLETE) {
	rand_sleep(10000,1000000);
	ret = BF_MEMP_SYNC(dbe, NULL);
    }
#else
    (void)dbe;
    ret = 0;
#endif

    return ret;
}

ex_t dbx_recover(bfpath *bfp, bool catastrophic, bool force)
{
    dbe_t *env = xcalloc(1, sizeof(dbe_t));

    /* set exclusive/write lock for recovery */
    while ((force || needs_recovery())
	    && (db_try_glock(bfp, F_WRLCK, F_SETLKW) <= 0))
	rand_sleep(10000,1000000);

    /* ok, when we have the lock, a concurrent process may have
     * proceeded with recovery */
    if (!(force || needs_recovery()))
	return EX_OK;

    if (DEBUG_DATABASE(0))
        fprintf(dbgout, "running %s data base recovery\n",
	    catastrophic ? "catastrophic" : "regular");
    env = dbe_xinit(env, bfp,
		    catastrophic ? DB_RECOVER_FATAL : DB_RECOVER);
    if (env == NULL) {
	exit(EX_ERROR);
    }

    clear_lockfile();
    dbx_cleanup_lite(env);

    return EX_OK;
}

static ex_t dbx_common_close(DB_ENV *dbe, bfpath *bfp)
{
    int e;

    if (db_log_autoremove)
	dbe_env_purgelogs(dbe);

    if (DEBUG_DATABASE(0))
	fprintf(dbgout, "closing environment\n");

    e = dbe->close(dbe, 0);
    if (e != 0) {
	print_error(__FILE__, __LINE__, "Error closing environment \"%s\": %s",
		    bfp->dirname, db_strerror(e));
	exit(EX_ERROR);
    }

    clear_lock();
    db_try_glock(bfp, F_UNLCK, F_SETLKW); /* release lock */
    return EX_OK;
}

static ex_t dbe_env_purgelogs(DB_ENV *dbe)
{
    char **i, **list;
    ex_t e;

    /* figure redundant log files and nuke them */
    e = BF_LOG_ARCHIVE(dbe, &list, DB_ARCH_ABS);
    if (e != 0) {
	print_error(__FILE__, __LINE__,
		"DB_ENV->log_archive failed: %s",
		db_strerror(e));
	exit(EX_ERROR);
    }

    if (list != NULL) {
	if (DEBUG_DATABASE(0))
	    fprintf(dbgout, "removing inactive logfiles\n");

	for (i = list; *i != NULL; i++) {
	    if (DEBUG_DATABASE(1))
		fprintf(dbgout, " removing logfile %s\n", *i);
	    if (unlink(*i)) {
		if (errno != ENOENT)
		    print_error(__FILE__, __LINE__,
			    "cannot unlink \"%s\": %s", *i, strerror(errno));
		/* proceed anyways */
	    }
	}
	xfree(list);
    }
    return EX_OK;
}

/** checkpoint the open environment \a dbe once and unconditionally */
static ex_t dbe_env_checkpoint(DB_ENV *dbe) {
    int e;

    if (DEBUG_DATABASE(0))
	fprintf(dbgout, "checkpoint database\n");

    /* checkpoint the transactional system */
    e = BF_TXN_CHECKPOINT(dbe, 0, 0, 0);
    e = dbx_sync(dbe, e);
    if (e != 0) {
	print_error(__FILE__, __LINE__, "DB_ENV->txn_checkpoint failed: %s",
		db_strerror(e));
	exit(EX_ERROR);
    }

    return EX_OK;
}

static ex_t dbe_simpleop(bfpath *bfp, ex_t func(DB_ENV *env))
{
    ex_t e;
    DB_ENV *dbe = dbe_recover_open(bfp, 0);

    if (dbe == NULL)
	exit(EX_ERROR);

    e = func(dbe);
    dbx_common_close(dbe, bfp);
    return e;
}

ex_t dbx_checkpoint(bfpath *bfp)
{
    return dbe_simpleop(bfp, dbe_env_checkpoint);
}

static ex_t i_purgelogs(DB_ENV *dbe)
{
    int e = dbe_env_checkpoint(dbe);

    if (e != 0)
	return e;
    else
	return dbe_env_purgelogs(dbe);
}

ex_t dbx_purgelogs(bfpath *bfp)
{
    return dbe_simpleop(bfp, i_purgelogs);
}

ex_t dbx_remove(bfpath *bfp)
{
    DB_ENV *dbe = dbe_recover_open(bfp, DB_PRIVATE);

    if (dbe == NULL)
	exit(EX_ERROR);

    return dbx_common_close(dbe, bfp);
}

void dbx_log_flush(DB_ENV *dbe)
{
    int ret;

    ret = BF_LOG_FLUSH(dbe, NULL);

    if (DEBUG_DATABASE(1))
	fprintf(dbgout, "DB_ENV->log_flush(%p): %s\n", (void *)dbe,
		db_strerror(ret));
}

const char **dsm_help_bogofilter(void)
{
    static const char *help_text[] = {
	NULL
    };
    return &help_text[0]; 
}

const char **dsm_help_bogoutil(void)
{
    static const char *help_text[] = {
	"environment maintenance:\n",
	"      --db-transaction=BOOL   - enable or disable transactions\n",
	"                                (only effective at creation time)\n",
	"      --db-verify=file        - verify data file.\n",
	"      --db-checkpoint=dir     - flush buffer cache and checkpoint database\n",
	"      --db-list-logfiles=dir [FLAGS]\n"
	"                              - list logfiles in environment\n",
	"      --db-prune=dir          - remove inactive log files in dir.\n",
	"      --db-recover=dir        - run recovery on database in dir.\n",
	"      --db-recover-harder=dir - run catastrophic recovery on database.\n",
	"      --db-remove-environment - remove environment.\n",

#ifdef	HAVE_DECL_DB_CREATE
	"      --db-lk-max-locks       - set max lock count.\n",
	"      --db-lk-max-objects     - set max object count.\n",
	"      --db-log-autoremove     - set autoremoving of logs.\n",
#ifdef	FUTURE_DB_OPTIONS
	"      --db-txn-durable        - set durable mode.\n",
#endif
#endif
	"\n",
	NULL
    };
    return &help_text[0]; 
}

bool dsm_options_bogofilter(int option, const char *name, const char *val)
{
    switch (option) {
	case O_DB_TRANSACTION:		
	    eTransaction = get_txn(name, val);								return true;

#ifdef	HAVE_DECL_DB_CREATE
	case O_DB_LOG_AUTOREMOVE:	db_log_autoremove	= get_bool(name, val);			return true;
#ifdef	FUTURE_DB_OPTIONS
	case O_DB_TXN_DURABLE:		db_txn_durable		= get_bool(name, val);			return true;
#endif
#endif
	default:				return false;
    }
}

bool dsm_options_bogoutil(int option, cmd_t *flag, int *count, const char **ds_file, const char *name, const char *val)
{
    switch (option) {
	case O_DB_TRANSACTION:
	    eTransaction = get_txn(name, val);
	    return true;

	case O_DB_RECOVER:
	    *flag = M_RECOVER;
	    *count += 1;
	    *ds_file = val;
	    return true;

	case O_DB_RECOVER_HARDER:
	    *flag = M_CRECOVER;
	    *count += 1;
	    *ds_file = val;
	    return true;

	case O_DB_CHECKPOINT:
	    *flag = M_CHECKPOINT;
	    *count += 1;
	    *ds_file = val;
	    return true;

	case O_DB_LIST_LOGFILES:
	    *flag = M_LIST_LOGFILES;
	    *count += 1;
	    *ds_file = val;
	    return true;

	case O_DB_PRUNE:
	    *flag = M_PURGELOGS;
	    *count += 1;
	    *ds_file = val;
	    return true;

	case O_DB_REMOVE_ENVIRONMENT:
	    *flag = M_REMOVEENV;
	    *ds_file = val;
	    *count += 1;
	    return true;

#ifdef	FUTURE_DB_OPTIONS
	case O_DB_TXN_DURABLE:
	    db_txn_durable    = get_bool(name, val);
	    return true;
#endif

	default:
	    return false;
    }
}

/** probe if the directory contains an environment, and if so,
 * if it has transactions
 */
e_txn probe_txn(bfpath *bfp)
{
    DB_ENV *dbe;
    int r;
#if DB_AT_LEAST(4,2)
    u_int32_t flags;
#endif

    r = db_env_create(&dbe, 0);
    if (r) {
	print_error(__FILE__, __LINE__, "cannot create environment handle: %s",
		db_strerror(r));
	return T_ERROR;
    }

    /* we might call dbe->set_flags here to set DB_NOPANIC, but this is
     * only supported from 4.1 onwards and probably not worth the
     * effort, we'll just check for DB_RUNRECOVERY */

#if DB_AT_LEAST(3,2)
    r = dbe->open(dbe, bfp->dirname, DB_JOINENV, DS_MODE);
#else
    r = ENOENT;
#endif
    if (r == DB_RUNRECOVERY) {
	dbe->close(dbe, 0);
	return T_ENABLED;
    }

    if (r == ENOENT) {
	struct stat st;
	int w;
	char *t = bfp->filepath;
	struct dirent *de;
	e_txn rc = T_DONT_KNOW;
	DIR *d;

	/* no environment found by JOINENV, but clean up handle */
	dbe->close(dbe, 0);

	/* retry, looking for log\.[0-9]{10} files - needed for instance
	 * after bogoutil --db-remove DIR or when DB_JOINENV is
	 * unsupported */
	d = opendir(bfp->dirname);
	if (d == NULL) {
	    print_error(__FILE__, __LINE__, "cannot open directory %s: %s",
		    t, strerror(r));
	    rc = T_ERROR;
	} else {
	    while ((errno = 0, de = readdir(d))) {
		if (strlen(de->d_name) == 14
		    && strncmp(de->d_name, "log.", 4) == 0
		    && strspn(de->d_name + 4, "0123456789") == 10)
		{
		    rc = T_ENABLED;
		    break;
		}
	    }
	    if (errno)
		rc = T_ERROR;
	    closedir(d);

	    if (rc != T_ERROR && rc != T_ENABLED) {
		w = stat(t, &st);
		if (w == 0) {
		    rc = T_DISABLED;
		} else if (errno != ENOENT) {
		    rc = T_ERROR;
		    print_error(__FILE__, __LINE__, "cannot stat %s: %s",
			    t, db_strerror(r));
		}
	    }
	}
	return rc;
    } /* if (r == ENOENT) for environment join */

    if (r != 0) {
	print_error(__FILE__, __LINE__, "cannot join environment: %s",
		db_strerror(r));
	return T_ERROR;
    }

    /* environment found, validate if it has transactions */
#if DB_AT_LEAST(4,2)
    r = dbe->get_open_flags(dbe, &flags);
    if (r) {
	print_error(__FILE__, __LINE__, "cannot query flags: %s",
		db_strerror(r));
	return T_ERROR;
    }

    dbe->close(dbe, 0);
    if ((flags & DB_INIT_TXN) == 0) {
	print_error(__FILE__, __LINE__,
		"environment found but does not support transactions.");
	return T_ERROR;
    }
#else
    dbe->close(dbe, 0);
#endif
    return T_ENABLED;
}

static ex_t dbx_list_logfiles(bfpath *bfp, int argc, char **argv)
{
    ex_t e;
    char **list, **i;
    int j;
    u_int32_t flags = 0;
    DB_ENV *dbe = dbe_recover_open(bfp, 0);

    if (dbe == NULL)
	exit(EX_ERROR);

    for (j = 0; j < argc; j++) {
	if (strcasecmp(argv[j], "all") == 0)
	    flags |= DB_ARCH_LOG;
	if (strcasecmp(argv[j], "absolute") == 0)
	    flags |= DB_ARCH_ABS;
    }

    e = BF_LOG_ARCHIVE(dbe, &list, flags);
    if (e != 0) {
	print_error(__FILE__, __LINE__,
		"DB_ENV->log_archive failed: %s",
		db_strerror(e));
	exit(EX_ERROR);
    }

    if (list) {
	for (i = list; *i; i++) {
	    if (flags & DB_ARCH_ABS)
		puts(*i);
	    else
		printf("%s%s%s\n", bfp->dirname, DIRSEP_S, *i);
	}
    }

    xfree(list);
    fflush(stdout);
    e = ferror(stdout) ? EX_ERROR : EX_OK;

    if (dbx_common_close(dbe, bfp)) e = EX_ERROR;
    return e;
}