erl_db_tree.c

OTP是开放电信平台的简称

/* ``The contents of this file are subject to the Erlang Public License,
 * Version 1.1, (the "License"); you may not use this file except in
 * compliance with the License. You should have received a copy of the
 * Erlang Public License along with this software. If not, it can be
 * retrieved via the world wide web at http://www.erlang.org/.
 * 
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Initial Developer of the Original Code is Ericsson Utvecklings AB.
 * Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
 * AB. All Rights Reserved.''
 * 
 *     $Id$
 */
/*
** Implementation of ordered ETS tables.
** The tables are implemented as AVL trees (Published by Adelson-Velski 
** and Landis). A nice source for learning about these trees is
** Wirth's Algorithms + Datastructures = Programs.
** The implementation here is however not made with recursion
** as the examples in Wirths book are.
*/

/*
#ifdef DEBUG
#define HARDDEBUG 1
#endif
*/
#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#include "sys.h"
#include "erl_vm.h"
#include "global.h"
#include "erl_process.h"
#include "error.h"
#define ERTS_WANT_DB_INTERNAL__
#include "erl_db.h"
#include "bif.h"
#include "big.h"

#include "erl_db_tree.h"



#define GETKEY(dtt, tplp)   (*((tplp) +  (dtt)->common.keypos))
#define GETKEY_WITH_POS(Keypos, Tplp) (*((Tplp) + Keypos))

/*
** A stack of this size is enough for an AVL tree with more than
** 0xFFFFFFFF elements. May be subject to change if
** the datatype of the element counter is changed to a 64 bit integer.
** The Maximal height of an AVL tree is calculated as:
** h(n) <= 1.4404 * log(n + 2) - 0.328
** Where n denotes the number of nodes, h(n) the height of the tree
** with n nodes and log is the binary logarithm.
*/

#define STACK_NEED 50
#define TREE_MAX_ELEMENTS 0xFFFFFFFFUL

#define PUSH_NODE(Dtt, Tdt)                             \
((Dtt)->stack[(Dtt)->stack_pos++] = Tdt)

#define POP_NODE(Dtt)				\
     ((Dtt->stack_pos) ? 			\
      (Dtt)->stack[--((Dtt)->stack_pos)] : NULL)

#define TOP_NODE(Dtt)                           \
     ((Dtt->stack_pos) ? 			\
      (Dtt)->stack[(Dtt)->stack_pos - 1] : NULL)

#define EMPTY_NODE(Dtt) (TOP_NODE(Dtt) == NULL)

     
/*
** Some macros for "direction stacks"
*/
#define DIR_LEFT 0
#define DIR_RIGHT 1
#define DIR_END 2 

/*
 * Special binary flag
 */
#define BIN_FLAG_ALL_OBJECTS         BIN_FLAG_USR1

/*
 * Number of records to delete before trapping.
 */
#define DELETE_RECORD_LIMIT 12000

/* 
** Debugging
*/
#ifdef HARDDEBUG
static TreeDbTerm *traverse_until(TreeDbTerm *t, int *current, int to);
static void check_slot_pos(DbTableTree *tb);
static void check_saved_stack(DbTableTree *tb);
static int check_table_tree(TreeDbTerm *t);

#define TREE_DEBUG
#endif

#ifdef TREE_DEBUG
/*
** Primitive trace macro
*/
#define DBG erts_fprintf(stderr,"%d\n",__LINE__)

/*
** Debugging dump
*/

static void do_dump_tree2(int to, void *to_arg, int show, TreeDbTerm *t,
			  int offset);

#else

#define DBG /* nothing */

#endif

/*
 * Size calculations
 */
#define SIZ_OVERHEAD ((sizeof(TreeDbTerm)/sizeof(Eterm)) - 1)
#define SIZ_DBTERM(TDT) (SIZ_OVERHEAD + (TDT)->dbterm.size)

/*
** Datatypes
*/

/* 
 * This structure is filled in by analyze_pattern() for the select 
 * functions.
 */
struct mp_info {
    int all_objects;		/* True if complete objects are always
				 * returned from the match_spec (can use 
				 * copy_shallow on the return value) */
    int something_can_match;	/* The match_spec is not "impossible" */
    int some_limitation;	/* There is some limitation on the search
				 * area, i. e. least and/or most is set.*/
    int got_partial;		/* The limitation has a partially bound
				 * key */
    Eterm least;		/* The lowest matching key (possibly 
				 * partially bound expression) */
    Eterm most;                 /* The highest matching key (possibly 
				 * partially bound expression) */

    TreeDbTerm *save_term;      /* If the key is completely bound, this
				 * will be the Tree node we're searching
				 * for, otherwise it will be useless */
    Binary *mp;                 /* The compiled match program */
};

/*
 * Used by doit_select(_chunk)
 */
struct select_context {
    Process *p;
    Eterm accum;
    Binary *mp;
    Eterm end_condition;
    Eterm *lastobj;
    Sint32 max;
    int keypos;
    int all_objects;
    Sint got;
    Sint chunk_size;
};

/*
 * Used by doit_select_count
 */
struct select_count_context {
    Process *p;
    Binary *mp;
    Eterm end_condition;
    Eterm *lastobj;
    Sint32 max;
    int keypos;
    int all_objects;
    Sint got;
};

/*
 * Used by doit_select_delete
 */
struct select_delete_context {
    Process *p;
    DbTableTree *tb;
    Uint accum;
    Binary *mp;
    Eterm end_condition;
    int erase_lastterm;
    TreeDbTerm *lastterm;
    Sint32 max;
    int keypos;
};

/*
** Forward declarations 
*/
static TreeDbTerm *linkout_tree(DbTableTree *tb, Eterm key);
static TreeDbTerm *linkout_object_tree(DbTableTree *tb, 
				       Eterm object);
static void do_free_tree(DbTableTree *tb);
static int do_free_tree_cont(DbTableTree *tb, int num_left);
static TreeDbTerm* get_term(DbTableTree *tb,
			    TreeDbTerm* old, 
			    Eterm obj);
static void free_term(DbTableTree *tb, TreeDbTerm* p);
static int balance_left(TreeDbTerm **this); 
static int balance_right(TreeDbTerm **this); 
static int delsub(TreeDbTerm **this); 
static TreeDbTerm *slot_search(Process *p, DbTableTree *tb, Sint slot);
static int realloc_counter(DbTableCommon *tb, TreeDbTerm** bp, Uint sz, 
			   Eterm new_counter, int counterpos);
static TreeDbTerm *find_node(DbTableTree *tb, Eterm key);
static TreeDbTerm **find_node2(DbTableTree *tb, Eterm key);
static TreeDbTerm *find_next(DbTableTree *tb, Eterm key);
static TreeDbTerm *find_prev(DbTableTree *tb, Eterm key);
static TreeDbTerm *find_next_from_pb_key(DbTableTree *tb, Eterm key);
static TreeDbTerm *find_prev_from_pb_key(DbTableTree *tb, Eterm key);
static void traverse_backwards(DbTableTree *tb,
			       Eterm lastkey,
			       int (*doit)(DbTableTree *tb,
					   TreeDbTerm *,
					   void *,
					   int),
			       void *context); 
static void traverse_forward(DbTableTree *tb,
			     Eterm lastkey,
			     int (*doit)(DbTableTree *tb,
					 TreeDbTerm *,
					 void *,
					 int),
			     void *context); 
static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm **ret,
		     Eterm *partly_bound_key);
static Sint cmp_partly_bound(Eterm partly_bound_key, Eterm bound_key);
static Sint do_cmp_partly_bound(Eterm a, Eterm b, int *done);

static int analyze_pattern(DbTableTree *tb, Eterm pattern, 
			   struct mp_info *mpi);
static int doit_select(DbTableTree *tb,
		       TreeDbTerm *this,
		       void *ptr,
		       int forward);
static int doit_select_count(DbTableTree *tb,
			     TreeDbTerm *this,
			     void *ptr,
			     int forward);
static int doit_select_chunk(DbTableTree *tb,
			     TreeDbTerm *this,
			     void *ptr,
			     int forward);
static int doit_select_delete(DbTableTree *tb,
			      TreeDbTerm *this,
			      void *ptr,
			      int forward);
static void do_dump_tree(int to, void *to_arg, TreeDbTerm *t);

static int partly_bound_can_match_lesser(Eterm partly_bound_1, 
					 Eterm partly_bound_2);
static int partly_bound_can_match_greater(Eterm partly_bound_1, 
					  Eterm partly_bound_2); 
static int do_partly_bound_can_match_lesser(Eterm a, Eterm b, 
					    int *done);
static int do_partly_bound_can_match_greater(Eterm a, Eterm b, 
					     int *done);
static BIF_RETTYPE ets_select_reverse(Process *p, Eterm a1, 
				      Eterm a2, Eterm a3);

/* Method interface functions */
static int db_first_tree(Process *p, DbTable *tbl, 
		  Eterm *ret);
static int db_next_tree(Process *p, DbTable *tbl, 
			Eterm key, Eterm *ret);
static int db_last_tree(Process *p, DbTable *tbl, 
			Eterm *ret);
static int db_prev_tree(Process *p, DbTable *tbl, 
			Eterm key,
			Eterm *ret);
static int db_update_counter_tree(Process *p, DbTable *tbl, 
				  Eterm key, Eterm incr,
				  int warp, int counterpos,
				  Eterm *ret);
static int db_put_tree(Process *p, DbTable *tbl, 
		       Eterm obj, Eterm *ret);
static int db_get_tree(Process *p, DbTable *tbl, 
		       Eterm key,  Eterm *ret);
static int db_member_tree(Process *p, DbTable *tbl, 
			  Eterm key,
			  Eterm *ret);
static int db_get_element_tree(Process *p, DbTable *tbl, 
			       Eterm key,int ndex,
			       Eterm *ret);
static int db_erase_tree(Process *p, DbTable *tbl, 
			 Eterm key, Eterm *ret);
static int db_erase_object_tree(Process *p, DbTable *tbl, 
				Eterm object,Eterm *ret);
static int db_slot_tree(Process *p, DbTable *tbl, 
			Eterm slot_term,  Eterm *ret);
static int db_select_tree(Process *p, DbTable *tbl, 
			  Eterm pattern, int reversed, Eterm *ret);
static int db_select_count_tree(Process *p, DbTable *tbl, 
				Eterm pattern,  Eterm *ret);
static int db_select_chunk_tree(Process *p, DbTable *tbl, 
				Eterm pattern, Sint chunk_size,
				int reversed, Eterm *ret);
static int db_select_continue_tree(Process *p, DbTable *tbl,
				   Eterm continuation, Eterm *ret);
static int db_select_count_continue_tree(Process *p, DbTable *tbl,
					 Eterm continuation, Eterm *ret);
static int db_select_delete_tree(Process *p, DbTable *tbl, 
				 Eterm pattern,  Eterm *ret);
static int db_select_delete_continue_tree(Process *p, DbTable *tbl, 
					  Eterm continuation, Eterm *ret);
static void db_print_tree(int to, void *to_arg,
			  int show, DbTable *tbl);
static int db_free_table_tree(DbTable *tbl);

static int db_free_table_continue_tree(DbTable *tbl, int first);

static void db_foreach_offheap_tree(DbTable *,
				    void (*)(ErlOffHeap *, void *),
				    void *);

static int db_delete_all_objects_tree(Process* p, DbTable* tbl);

#ifdef HARDDEBUG
static void db_check_table_tree(DbTable *tbl);
#endif

/*
** Static variables
*/

Export ets_select_reverse_exp;

/*
** External interface 
*/
DbTableMethod db_tree =
{
    db_create_tree,
    db_first_tree,
    db_next_tree,
    db_last_tree,
    db_prev_tree,
    db_put_tree,
    db_get_tree,
    db_get_element_tree,
    db_member_tree,
    db_erase_tree,
    db_erase_object_tree,
    db_slot_tree,
    db_update_counter_tree,
    db_select_chunk_tree,
    db_select_tree, /* why not chunk size=0 ??? */
    db_select_delete_tree,
    db_select_continue_tree,
    db_select_delete_continue_tree,
    db_select_count_tree,
    db_select_count_continue_tree,
    db_delete_all_objects_tree,
    db_free_table_tree,
    db_free_table_continue_tree,
    db_print_tree,
    db_foreach_offheap_tree,
#ifdef HARDDEBUG
    db_check_table_tree,
#else
    NULL,
#endif
};





void db_initialize_tree(void)
{
    memset(&ets_select_reverse_exp, 0, sizeof(Export));
    ets_select_reverse_exp.address = 
	&ets_select_reverse_exp.code[3];
    ets_select_reverse_exp.code[0] = am_ets;
    ets_select_reverse_exp.code[1] = am_reverse;
    ets_select_reverse_exp.code[2] = 3;
    ets_select_reverse_exp.code[3] =
	(Eterm) em_apply_bif;
    ets_select_reverse_exp.code[4] = 
	(Eterm) &ets_select_reverse;
    return;
};

/*
** Table interface routines ie what's called by the bif's 
*/

int db_create_tree(Process *p, DbTable *tbl)
{
    DbTableTree *tb = &tbl->tree;
    tb->root = NULL;
    tb->stack = erts_db_alloc(ERTS_ALC_T_DB_STK,
			      (DbTable *) tb,
			      sizeof(TreeDbTerm *) * STACK_NEED);
    tb->stack_pos = 0;
    tb->slot_pos = 0;
    tb->deletion = 0;
    return DB_ERROR_NONE;
}

static int db_first_tree(Process *p, DbTable *tbl, Eterm *ret)
{
    DbTableTree *tb = &tbl->tree;
    TreeDbTerm *this;
    Eterm e;
    Eterm *hp;
    Uint sz;

    /* Walk down to the tree to the left */
    tb->stack_pos = tb->slot_pos = 0;
    if (( this = tb->root ) == NULL) {
	*ret = am_EOT;
	return DB_ERROR_NONE;
    }
    while (this != NULL) {
	PUSH_NODE(tb, this);
	this = this->left;
    }
    this = TOP_NODE(tb);
    e = GETKEY(tb, this->dbterm.tpl);
    sz = size_object(e);

    hp = HAlloc(p, sz);

    tb->slot_pos = 1; /* Slot pos is position in order, 
			 if it's other than 0 */

    *ret = copy_struct(e,sz,&hp,&MSO(p));
    
    return DB_ERROR_NONE;
}

static int db_next_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
{
    DbTableTree *tb = &tbl->tree;
    TreeDbTerm *this;
    Eterm e;
    Eterm *hp;
    Uint sz;

    if (is_atom(key) && key == am_EOT)
	return DB_ERROR_BADKEY;
    if (( this = find_next(tb, key) ) == NULL) {
	*ret = am_EOT;
	return DB_ERROR_NONE;
    }
    
    e = GETKEY(tb, this->dbterm.tpl);
    sz = size_object(e);

    hp = HAlloc(p, sz);

    *ret = copy_struct(e,sz,&hp,&MSO(p));
    
    return DB_ERROR_NONE;
}

static int db_last_tree(Process *p, DbTable *tbl, Eterm *ret)
{
    DbTableTree *tb = &tbl->tree;
    TreeDbTerm *this;
    Eterm e;
    Eterm *hp;
    Uint sz;

    /* Walk down to the tree to the left */
    tb->stack_pos = tb->slot_pos = 0;
    if (( this = tb->root ) == NULL) {
	*ret = am_EOT;
	return DB_ERROR_NONE;
    }
    while (this != NULL) {
	PUSH_NODE(tb, this);
	this = this->right;
    }
    this = TOP_NODE(tb);
    e = GETKEY(tb, this->dbterm.tpl);
    sz = size_object(e);

    hp = HAlloc(p, sz);

    tb->slot_pos = tb->common.nitems; /* Slot pos is position in order, 
				  if it's other than 0 */

    *ret = copy_struct(e,sz,&hp,&MSO(p));
    
    return DB_ERROR_NONE;
}

static int db_prev_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
{
    DbTableTree *tb = &tbl->tree;
    TreeDbTerm *this;
    Eterm e;
    Eterm *hp;
    Uint sz;

    if (is_atom(key) && key == am_EOT)
	return DB_ERROR_BADKEY;
    if (( this = find_prev(tb, key) ) == NULL) {
	*ret = am_EOT;
	return DB_ERROR_NONE;
    }
    
    e = GETKEY(tb, this->dbterm.tpl);
    sz = size_object(e);

    hp = HAlloc(p, sz);

    *ret = copy_struct(e,sz,&hp,&MSO(p));
    
    return DB_ERROR_NONE;
}

static int db_update_counter_tree(Process *p,
				  DbTable *tbl, 
				  Eterm key,
				  Eterm incr,
				  int warp,
				  int counterpos,
				  Eterm *ret)
{
    DbTableTree *tb = &tbl->tree;
    TreeDbTerm **bp = find_node2(tb, key);
    TreeDbTerm *b;
    int res;
    if (bp == NULL)
	return DB_ERROR_BADKEY;
    b = *bp;
    if (counterpos <= 0)