db_join.c

这是国外的resip协议栈

/*
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 1998-2004
 *	Sleepycat Software.  All rights reserved.
 *
 * $Id: db_join.c,v 11.75 2004/09/22 03:30:23 bostic Exp $
 */

#include "db_config.h"

#ifndef NO_SYSTEM_INCLUDES
#include <sys/types.h>

#include <stdlib.h>
#include <string.h>
#endif

#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/db_join.h"
#include "dbinc/btree.h"

static int __db_join_close_pp __P((DBC *));
static int __db_join_cmp __P((const void *, const void *));
static int __db_join_del __P((DBC *, u_int32_t));
static int __db_join_get __P((DBC *, DBT *, DBT *, u_int32_t));
static int __db_join_get_pp __P((DBC *, DBT *, DBT *, u_int32_t));
static int __db_join_getnext __P((DBC *, DBT *, DBT *, u_int32_t, u_int32_t));
static int __db_join_primget __P((DB *,
    DB_TXN *, u_int32_t, DBT *, DBT *, u_int32_t));
static int __db_join_put __P((DBC *, DBT *, DBT *, u_int32_t));

/*
 * Check to see if the Nth secondary cursor of join cursor jc is pointing
 * to a sorted duplicate set.
 */
#define	SORTED_SET(jc, n)   ((jc)->j_curslist[(n)]->dbp->dup_compare != NULL)

/*
 * This is the duplicate-assisted join functionality.  Right now we're
 * going to write it such that we return one item at a time, although
 * I think we may need to optimize it to return them all at once.
 * It should be easier to get it working this way, and I believe that
 * changing it should be fairly straightforward.
 *
 * We optimize the join by sorting cursors from smallest to largest
 * cardinality.  In most cases, this is indeed optimal.  However, if
 * a cursor with large cardinality has very few data in common with the
 * first cursor, it is possible that the join will be made faster by
 * putting it earlier in the cursor list.  Since we have no way to detect
 * cases like this, we simply provide a flag, DB_JOIN_NOSORT, which retains
 * the sort order specified by the caller, who may know more about the
 * structure of the data.
 *
 * The first cursor moves sequentially through the duplicate set while
 * the others search explicitly for the duplicate in question.
 *
 */

/*
 * __db_join --
 *	This is the interface to the duplicate-assisted join functionality.
 * In the same way that cursors mark a position in a database, a cursor
 * can mark a position in a join.  While most cursors are created by the
 * cursor method of a DB, join cursors are created through an explicit
 * call to DB->join.
 *
 * The curslist is an array of existing, initialized cursors and primary
 * is the DB of the primary file.  The data item that joins all the
 * cursors in the curslist is used as the key into the primary and that
 * key and data are returned.  When no more items are left in the join
 * set, the  c_next operation off the join cursor will return DB_NOTFOUND.
 *
 * PUBLIC: int __db_join __P((DB *, DBC **, DBC **, u_int32_t));
 */
int
__db_join(primary, curslist, dbcp, flags)
	DB *primary;
	DBC **curslist, **dbcp;
	u_int32_t flags;
{
	DB_ENV *dbenv;
	DBC *dbc;
	JOIN_CURSOR *jc;
	size_t ncurs, nslots;
	u_int32_t i;
	int ret;

	dbenv = primary->dbenv;
	dbc = NULL;
	jc = NULL;

	if ((ret = __os_calloc(dbenv, 1, sizeof(DBC), &dbc)) != 0)
		goto err;

	if ((ret = __os_calloc(dbenv, 1, sizeof(JOIN_CURSOR), &jc)) != 0)
		goto err;

	if ((ret = __os_malloc(dbenv, 256, &jc->j_key.data)) != 0)
		goto err;
	jc->j_key.ulen = 256;
	F_SET(&jc->j_key, DB_DBT_USERMEM);

	F_SET(&jc->j_rdata, DB_DBT_REALLOC);

	for (jc->j_curslist = curslist;
	    *jc->j_curslist != NULL; jc->j_curslist++)
		;

	/*
	 * The number of cursor slots we allocate is one greater than
	 * the number of cursors involved in the join, because the
	 * list is NULL-terminated.
	 */
	ncurs = (size_t)(jc->j_curslist - curslist);
	nslots = ncurs + 1;

	/*
	 * !!! -- A note on the various lists hanging off jc.
	 *
	 * j_curslist is the initial NULL-terminated list of cursors passed
	 * into __db_join.  The original cursors are not modified; pristine
	 * copies are required because, in databases with unsorted dups, we
	 * must reset all of the secondary cursors after the first each
	 * time the first one is incremented, or else we will lose data
	 * which happen to be sorted differently in two different cursors.
	 *
	 * j_workcurs is where we put those copies that we're planning to
	 * work with.  They're lazily c_dup'ed from j_curslist as we need
	 * them, and closed when the join cursor is closed or when we need
	 * to reset them to their original values (in which case we just
	 * c_dup afresh).
	 *
	 * j_fdupcurs is an array of cursors which point to the first
	 * duplicate in the duplicate set that contains the data value
	 * we're currently interested in.  We need this to make
	 * __db_join_get correctly return duplicate duplicates;  i.e., if a
	 * given data value occurs twice in the set belonging to cursor #2,
	 * and thrice in the set belonging to cursor #3, and once in all
	 * the other cursors, successive calls to __db_join_get need to
	 * return that data item six times.  To make this happen, each time
	 * cursor N is allowed to advance to a new datum, all cursors M
	 * such that M > N have to be reset to the first duplicate with
	 * that datum, so __db_join_get will return all the dup-dups again.
	 * We could just reset them to the original cursor from j_curslist,
	 * but that would be a bit slower in the unsorted case and a LOT
	 * slower in the sorted one.
	 *
	 * j_exhausted is a list of boolean values which represent
	 * whether or not their corresponding cursors are "exhausted",
	 * i.e. whether the datum under the corresponding cursor has
	 * been found not to exist in any unreturned combinations of
	 * later secondary cursors, in which case they are ready to be
	 * incremented.
	 */

	/* We don't want to free regions whose callocs have failed. */
	jc->j_curslist = NULL;
	jc->j_workcurs = NULL;
	jc->j_fdupcurs = NULL;
	jc->j_exhausted = NULL;

	if ((ret = __os_calloc(dbenv, nslots, sizeof(DBC *),
	    &jc->j_curslist)) != 0)
		goto err;
	if ((ret = __os_calloc(dbenv, nslots, sizeof(DBC *),
	    &jc->j_workcurs)) != 0)
		goto err;
	if ((ret = __os_calloc(dbenv, nslots, sizeof(DBC *),
	    &jc->j_fdupcurs)) != 0)
		goto err;
	if ((ret = __os_calloc(dbenv, nslots, sizeof(u_int8_t),
	    &jc->j_exhausted)) != 0)
		goto err;
	for (i = 0; curslist[i] != NULL; i++) {
		jc->j_curslist[i] = curslist[i];
		jc->j_workcurs[i] = NULL;
		jc->j_fdupcurs[i] = NULL;
		jc->j_exhausted[i] = 0;
	}
	jc->j_ncurs = (u_int32_t)ncurs;

	/*
	 * If DB_JOIN_NOSORT is not set, optimize secondary cursors by
	 * sorting in order of increasing cardinality.
	 */
	if (!LF_ISSET(DB_JOIN_NOSORT))
		qsort(jc->j_curslist, ncurs, sizeof(DBC *), __db_join_cmp);

	/*
	 * We never need to reset the 0th cursor, so there's no
	 * solid reason to use workcurs[0] rather than curslist[0] in
	 * join_get.  Nonetheless, it feels cleaner to do it for symmetry,
	 * and this is the most logical place to copy it.
	 *
	 * !!!
	 * There's no need to close the new cursor if we goto err only
	 * because this is the last thing that can fail.  Modifier of this
	 * function beware!
	 */
	if ((ret =
	    __db_c_dup(jc->j_curslist[0], jc->j_workcurs, DB_POSITION)) != 0)
		goto err;

	dbc->c_close = __db_join_close_pp;
	dbc->c_del = __db_join_del;
	dbc->c_get = __db_join_get_pp;
	dbc->c_put = __db_join_put;
	dbc->internal = (DBC_INTERNAL *)jc;
	dbc->dbp = primary;
	jc->j_primary = primary;

	/* Stash the first cursor's transaction here for easy access. */
	dbc->txn = curslist[0]->txn;

	*dbcp = dbc;

	MUTEX_THREAD_LOCK(dbenv, primary->mutexp);
	TAILQ_INSERT_TAIL(&primary->join_queue, dbc, links);
	MUTEX_THREAD_UNLOCK(dbenv, primary->mutexp);

	return (0);

err:	if (jc != NULL) {
		if (jc->j_curslist != NULL)
			__os_free(dbenv, jc->j_curslist);
		if (jc->j_workcurs != NULL) {
			if (jc->j_workcurs[0] != NULL)
				(void)__db_c_close(jc->j_workcurs[0]);
			__os_free(dbenv, jc->j_workcurs);
		}
		if (jc->j_fdupcurs != NULL)
			__os_free(dbenv, jc->j_fdupcurs);
		if (jc->j_exhausted != NULL)
			__os_free(dbenv, jc->j_exhausted);
		__os_free(dbenv, jc);
	}
	if (dbc != NULL)
		__os_free(dbenv, dbc);
	return (ret);
}

/*
 * __db_join_close_pp --
 *	DBC->c_close pre/post processing for join cursors.
 */
static int
__db_join_close_pp(dbc)
	DBC *dbc;
{
	DB_ENV *dbenv;
	DB *dbp;
	int handle_check, ret;

	dbp = dbc->dbp;
	dbenv = dbp->dbenv;

	PANIC_CHECK(dbenv);

	handle_check = IS_REPLICATED(dbenv, dbp);
	if (handle_check &&
	    (ret = __db_rep_enter(dbp, 0, 0, dbc->txn != NULL)) != 0)
		return (ret);

	ret = __db_join_close(dbc);

	if (handle_check)
		__env_db_rep_exit(dbenv);

	return (ret);
}

static int
__db_join_put(dbc, key, data, flags)
	DBC *dbc;
	DBT *key;
	DBT *data;
	u_int32_t flags;
{
	PANIC_CHECK(dbc->dbp->dbenv);

	COMPQUIET(key, NULL);
	COMPQUIET(data, NULL);
	COMPQUIET(flags, 0);
	return (EINVAL);
}

static int
__db_join_del(dbc, flags)
	DBC *dbc;
	u_int32_t flags;
{
	PANIC_CHECK(dbc->dbp->dbenv);

	COMPQUIET(flags, 0);
	return (EINVAL);
}

/*
 * __db_join_get_pp --
 *	DBjoin->get pre/post processing.
 */
static int
__db_join_get_pp(dbc, key, data, flags)
	DBC *dbc;
	DBT *key, *data;
	u_int32_t flags;
{
	DB *dbp;
	DB_ENV *dbenv;
	u_int32_t handle_check, save_flags;
	int ret;

	dbp = dbc->dbp;
	dbenv = dbp->dbenv;

	/* Save the original flags value. */
	save_flags = flags;

	PANIC_CHECK(dbenv);

	if (LF_ISSET(DB_DIRTY_READ | DB_DEGREE_2 | DB_RMW)) {
		if (!LOCKING_ON(dbp->dbenv))
			return (__db_fnl(dbp->dbenv, "DBcursor->c_get"));

		LF_CLR(DB_DIRTY_READ | DB_DEGREE_2 | DB_RMW);
	}

	switch (flags) {
	case 0:
	case DB_JOIN_ITEM:
		break;
	default:
		return (__db_ferr(dbp->dbenv, "DBcursor->c_get", 0));
	}

	/*
	 * A partial get of the key of a join cursor don't make much sense;
	 * the entire key is necessary to query the primary database
	 * and find the datum, and so regardless of the size of the key
	 * it would not be a performance improvement.  Since it would require
	 * special handling, we simply disallow it.
	 *
	 * A partial get of the data, however, potentially makes sense (if
	 * all possible data are a predictable large structure, for instance)
	 * and causes us no headaches, so we permit it.
	 */
	if (F_ISSET(key, DB_DBT_PARTIAL)) {
		__db_err(dbp->dbenv,
		    "DB_DBT_PARTIAL may not be set on key during join_get");
		return (EINVAL);
	}

	handle_check = IS_REPLICATED(dbp->dbenv, dbp);
	if (handle_check &&
	    (ret = __db_rep_enter(dbp, 1, 0, dbc->txn != NULL)) != 0)
		return (ret);

	/* Restore the original flags value. */
	flags = save_flags;

	ret = __db_join_get(dbc, key, data, flags);

	if (handle_check)
		__env_db_rep_exit(dbenv);

	return (ret);
}

static int
__db_join_get(dbc, key_arg, data_arg, flags)
	DBC *dbc;
	DBT *key_arg, *data_arg;
	u_int32_t flags;
{
	DBT *key_n, key_n_mem;
	DB *dbp;
	DBC *cp;
	JOIN_CURSOR *jc;
	int db_manage_data, ret;
	u_int32_t i, j, operation, opmods;

	dbp = dbc->dbp;
	jc = (JOIN_CURSOR *)dbc->internal;

	operation = LF_ISSET(DB_OPFLAGS_MASK);

	/* !!!
	 * If the set of flags here changes, check that __db_join_primget
	 * is updated to handle them properly.
	 */
	opmods = LF_ISSET(DB_RMW | DB_DEGREE_2 | DB_DIRTY_READ);

	/*
	 * Since we are fetching the key as a datum in the secondary indices,
	 * we must be careful of caller-specified DB_DBT_* memory
	 * management flags.  If necessary, use a stack-allocated DBT;
	 * we'll appropriately copy and/or allocate the data later.
	 */
	if (F_ISSET(key_arg, DB_DBT_USERMEM) ||
	    F_ISSET(key_arg, DB_DBT_MALLOC)) {
		/* We just use the default buffer;  no need to go malloc. */
		key_n = &key_n_mem;
		memset(key_n, 0, sizeof(DBT));
	} else {
		/*
		 * Either DB_DBT_REALLOC or the default buffer will work
		 * fine if we have to reuse it, as we do.
		 */
		key_n = key_arg;
	}

	/*
	 * If our last attempt to do a get on the primary key failed,
	 * short-circuit the join and try again with the same key.
	 */
	if (F_ISSET(jc, JOIN_RETRY))
		goto samekey;
	F_CLR(jc, JOIN_RETRY);

retry:	ret = __db_c_get(jc->j_workcurs[0], &jc->j_key, key_n,
	    opmods | (jc->j_exhausted[0] ? DB_NEXT_DUP : DB_CURRENT));

	if (ret == DB_BUFFER_SMALL) {
		jc->j_key.ulen <<= 1;
		if ((ret = __os_realloc(dbp->dbenv,
		    jc->j_key.ulen, &jc->j_key.data)) != 0)
			goto mem_err;
		goto retry;
	}

	/*
	 * If ret == DB_NOTFOUND, we're out of elements of the first
	 * secondary cursor.  This is how we finally finish the join
	 * if all goes well.
	 */
	if (ret != 0)
		goto err;

	/*
	 * If jc->j_exhausted[0] == 1, we've just advanced the first cursor,
	 * and we're going to want to advance all the cursors that point to
	 * the first member of a duplicate duplicate set (j_fdupcurs[1..N]).
	 * Close all the cursors in j_fdupcurs;  we'll reopen them the
	 * first time through the upcoming loop.
	 */
	for (i = 1; i < jc->j_ncurs; i++) {
		if (jc->j_fdupcurs[i] != NULL &&
		    (ret = __db_c_close(jc->j_fdupcurs[i])) != 0)
			goto err;
		jc->j_fdupcurs[i] = NULL;
	}

	/*
	 * If jc->j_curslist[1] == NULL, we have only one cursor in the join.
	 * Thus, we can safely increment that one cursor on each call
	 * to __db_join_get, and we signal this by setting jc->j_exhausted[0]
	 * right away.
	 *
	 * Otherwise, reset jc->j_exhausted[0] to 0, so that we don't
	 * increment it until we know we're ready to.
	 */
	if (jc->j_curslist[1] == NULL)
		jc->j_exhausted[0] = 1;
	else