The R13B03 release.OTP_R13B03

1 files changed, 3289 insertions, 0 deletions

diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
new file mode 100644
index 0000000000..d3a916d2d9
--- /dev/null
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -0,0 +1,3289 @@
+/*
+ * %CopyrightBegin%
+ * 
+ * Copyright Ericsson AB 1998-2009. All Rights Reserved.
+ * 
+ * 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 online 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.
+ * 
+ * %CopyrightEnd%
+ */
+
+/*
+** 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_binary.h"
+
+#include "erl_db_tree.h"
+
+
+
+#define GETKEY(dtt, tplp)   (*((tplp) +  (dtt)->common.keypos))
+#define GETKEY_WITH_POS(Keypos, Tplp) (*((Tplp) + Keypos))
+#define NITEMS(tb) ((int)erts_smp_atomic_read(&(tb)->common.nitems))
+
+/*
+** 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)->array[(Dtt)->pos++] = Tdt)
+
+#define POP_NODE(Dtt)			\
+     (((Dtt)->pos) ? 			\
+      (Dtt)->array[--((Dtt)->pos)] : NULL)
+
+#define TOP_NODE(Dtt)                   \
+     ((Dtt->pos) ? 			\
+      (Dtt)->array[(Dtt)->pos - 1] : NULL)
+
+#define EMPTY_NODE(Dtt) (TOP_NODE(Dtt) == NULL)
+
+
+
+/* Obtain table static stack if available. NULL if not.
+** Must be released with release_stack()
+*/
+static DbTreeStack* get_static_stack(DbTableTree* tb)
+{
+    if (!erts_smp_atomic_xchg(&tb->is_stack_busy, 1)) {
+	return &tb->static_stack;
+    }
+    return NULL;
+}
+
+/* Obtain static stack if available, otherwise empty dynamic stack.
+** Must be released with release_stack()
+*/
+static DbTreeStack* get_any_stack(DbTableTree* tb)
+{
+    DbTreeStack* stack;
+    if (!erts_smp_atomic_xchg(&tb->is_stack_busy, 1)) {
+	return &tb->static_stack;
+    }
+    stack = erts_db_alloc(ERTS_ALC_T_DB_STK, (DbTable *) tb,
+			  sizeof(DbTreeStack) + sizeof(TreeDbTerm*) * STACK_NEED);
+    stack->pos = 0;
+    stack->slot = 0;
+    stack->array = (TreeDbTerm**) (stack + 1);
+    return stack;
+}
+
+static void release_stack(DbTableTree* tb, DbTreeStack* stack)
+{
+    if (stack == &tb->static_stack) {
+	ASSERT(erts_smp_atomic_read(&tb->is_stack_busy) == 1);
+	erts_smp_atomic_set(&tb->is_stack_busy, 0);
+    }
+    else {
+	erts_db_free(ERTS_ALC_T_DB_STK, (DbTable *) tb,
+		     (void *) stack, sizeof(DbTreeStack) + sizeof(TreeDbTerm*) * STACK_NEED);
+    }
+}
+
+static void reset_static_stack(DbTableTree* tb)
+{
+    tb->static_stack.pos = 0;
+    tb->static_stack.slot = 0;
+}
+
+
+/*
+** 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 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 TreeDbTerm *find_node(DbTableTree *tb, Eterm key);
+static TreeDbTerm **find_node2(DbTableTree *tb, Eterm key);
+static TreeDbTerm *find_next(DbTableTree *tb, DbTreeStack*, Eterm key);
+static TreeDbTerm *find_prev(DbTableTree *tb, DbTreeStack*, Eterm key);
+static TreeDbTerm *find_next_from_pb_key(DbTableTree *tb, DbTreeStack*,
+					 Eterm key);
+static TreeDbTerm *find_prev_from_pb_key(DbTableTree *tb, DbTreeStack*,
+					 Eterm key);
+static void traverse_backwards(DbTableTree *tb,
+			       DbTreeStack*,
+			       Eterm lastkey,
+			       int (*doit)(DbTableTree *tb,
+					   TreeDbTerm *,
+					   void *,
+					   int),
+			       void *context); 
+static void traverse_forward(DbTableTree *tb,
+			     DbTreeStack*,
+			     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_put_tree(DbTable *tbl, Eterm obj, int key_clash_fail);
+static int db_get_tree(Process *p, DbTable *tbl, 
+		       Eterm key,  Eterm *ret);
+static int db_member_tree(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(DbTable *tbl, Eterm key, Eterm *ret);
+static int db_erase_object_tree(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);
+
+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 int db_lookup_dbterm_tree(DbTable *, Eterm key, DbUpdateHandle*);
+static void db_finalize_dbterm_tree(DbUpdateHandle*);
+
+/*
+** 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_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
+    db_lookup_dbterm_tree,
+    db_finalize_dbterm_tree
+
+};
+
+
+
+
+
+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->static_stack.array = erts_db_alloc(ERTS_ALC_T_DB_STK,
+					   (DbTable *) tb,
+					   sizeof(TreeDbTerm *) * STACK_NEED);
+    tb->static_stack.pos = 0;
+    tb->static_stack.slot = 0;
+    erts_smp_atomic_init(&tb->is_stack_busy, 0);
+    tb->deletion = 0;
+    return DB_ERROR_NONE;
+}
+
+static int db_first_tree(Process *p, DbTable *tbl, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    TreeDbTerm *this;
+    Eterm e;
+    Eterm *hp;
+    Uint sz;
+
+    if (( this = tb->root ) == NULL) {
+	*ret = am_EOT;
+	return DB_ERROR_NONE;
+    }
+    /* Walk down to the tree to the left */
+    if ((stack = get_static_stack(tb)) != NULL) {
+	stack->pos = stack->slot = 0;
+    }
+    while (this->left != NULL) {
+	if (stack) PUSH_NODE(stack, this);
+	this = this->left;
+    }
+    if (stack) {
+	PUSH_NODE(stack, this);
+	stack->slot = 1;
+	release_stack(tb,stack);
+    }
+    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_next_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    TreeDbTerm *this;
+    Eterm e;
+    Eterm *hp;
+    Uint sz;
+
+    if (is_atom(key) && key == am_EOT)
+	return DB_ERROR_BADKEY;
+    stack = get_any_stack(tb);
+    this = find_next(tb, stack, key);
+    release_stack(tb,stack);
+    if (this == 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;
+    DbTreeStack* stack;
+    Eterm e;
+    Eterm *hp;
+    Uint sz;
+
+    if (( this = tb->root ) == NULL) {
+	*ret = am_EOT;
+	return DB_ERROR_NONE;
+    }
+    /* Walk down to the tree to the left */
+    if ((stack = get_static_stack(tb)) != NULL) {
+	stack->pos = stack->slot = 0;
+    }    
+    while (this->right != NULL) {
+	if (stack) PUSH_NODE(stack, this);
+	this = this->right;
+    }
+    if (stack) {
+	PUSH_NODE(stack, this);
+	stack->slot = NITEMS(tb);
+	release_stack(tb,stack);
+    }
+    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_prev_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    TreeDbTerm *this;
+    DbTreeStack* stack;
+    Eterm e;
+    Eterm *hp;
+    Uint sz;
+
+    if (is_atom(key) && key == am_EOT)
+	return DB_ERROR_BADKEY;
+    stack = get_any_stack(tb);
+    this = find_prev(tb, stack, key);
+    release_stack(tb,stack);
+    if (this == 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_put_tree(DbTable *tbl, Eterm obj, int key_clash_fail)
+{
+    DbTableTree *tb = &tbl->tree;
+    /* Non recursive insertion in AVL tree, building our own stack */
+    TreeDbTerm **tstack[STACK_NEED];
+    int tpos = 0;
+    int dstack[STACK_NEED+1];
+    int dpos = 0;
+    int state = 0;
+    TreeDbTerm **this = &tb->root;
+    Sint c;
+    Eterm key;
+    int dir;
+    TreeDbTerm *p1, *p2, *p;
+
+    key = GETKEY(tb, tuple_val(obj));
+
+    reset_static_stack(tb);
+
+    dstack[dpos++] = DIR_END;
+    for (;;)
+	if (!*this) { /* Found our place */
+	    state = 1;
+	    if (erts_smp_atomic_inctest(&tb->common.nitems) >= TREE_MAX_ELEMENTS) {
+		erts_smp_atomic_dec(&tb->common.nitems);
+		return DB_ERROR_SYSRES;
+	    }
+	    *this = get_term(tb, NULL, obj);
+	    (*this)->balance = 0;
+	    (*this)->left = (*this)->right = NULL;
+	    break;
+	} else if ((c = cmp(key,GETKEY(tb,(*this)->dbterm.tpl))) < 0) { 
+	    /* go left */
+	    dstack[dpos++] = DIR_LEFT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->left);
+	} else if (c > 0) { /* go right */
+	    dstack[dpos++] = DIR_RIGHT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->right);
+	} else if (!key_clash_fail) { /* Equal key and this is a set, replace. */
+	    *this = get_term(tb, *this, obj);
+	    break;
+	} else {
+	    return DB_ERROR_BADKEY; /* key already exists */
+	}
+
+    while (state && ( dir = dstack[--dpos] ) != DIR_END) {
+	this = tstack[--tpos];
+	p = *this;
+	if (dir == DIR_LEFT) {
+	    switch (p->balance) {
+	    case 1:
+		p->balance = 0;
+		state = 0;
+		break;
+	    case 0:
+		p->balance = -1;
+		break;
+	    case -1: /* The icky case */
+		p1 = p->left;
+		if (p1->balance == -1) { /* Single LL rotation */
+		    p->left = p1->right;
+		    p1->right = p;
+		    p->balance = 0;
+		    (*this) = p1;
+		} else { /* Double RR rotation */
+		    p2 = p1->right;
+		    p1->right = p2->left;
+		    p2->left = p1;
+		    p->left = p2->right;
+		    p2->right = p;
+		    p->balance = (p2->balance == -1) ? +1 : 0;
+		    p1->balance = (p2->balance == 1) ? -1 : 0;
+		    (*this) = p2;
+		}
+		(*this)->balance = 0;
+		state = 0;
+		break;
+	    }
+	} else { /* dir == DIR_RIGHT */
+	    switch (p->balance) {
+	    case -1:
+		p->balance = 0;
+		state = 0;
+		break;
+	    case 0:
+		p->balance = 1;
+		break;
+	    case 1:
+		p1 = p->right;
+		if (p1->balance == 1) { /* Single RR rotation */
+		    p->right = p1->left;
+		    p1->left = p;
+		    p->balance = 0;
+		    (*this) = p1;
+		} else { /* Double RL rotation */
+		    p2 = p1->left;
+		    p1->left = p2->right;
+		    p2->right = p1;
+		    p->right = p2->left;
+		    p2->left = p;
+		    p->balance = (p2->balance == 1) ? -1 : 0;
+		    p1->balance = (p2->balance == -1) ? 1 : 0;
+		    (*this) = p2;
+		}
+		(*this)->balance = 0; 
+		state = 0;
+		break;
+	    }
+	}
+    }
+    return DB_ERROR_NONE;
+}
+
+static int db_get_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    Eterm copy;
+    Eterm *hp;
+    TreeDbTerm *this;
+
+    /*
+     * This is always a set, so we know exactly how large
+     * the data is when we have found it.
+     * The list created around it is purely for interface conformance.
+     */
+    
+    this = find_node(tb,key);
+    if (this == NULL) {
+	*ret = NIL;
+    } else {
+	hp = HAlloc(p, this->dbterm.size + 2);
+	copy = copy_shallow(DBTERM_BUF(&this->dbterm), 
+			    this->dbterm.size, 
+			    &hp, 
+			    &MSO(p));
+	*ret = CONS(hp, copy, NIL);
+    }
+    return DB_ERROR_NONE;
+}
+
+static int db_member_tree(DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+
+    *ret = (find_node(tb,key) == NULL) ? am_false : am_true;
+    return DB_ERROR_NONE;
+}
+
+static int db_get_element_tree(Process *p, DbTable *tbl,
+			       Eterm key, int ndex, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    /*
+     * Look the node up:
+     */
+    Eterm *hp;
+    TreeDbTerm *this;
+
+    /*
+     * This is always a set, so we know exactly how large
+     * the data is when we have found it.
+     * No list is created around elements in set's so there are no list
+     * around the element here either.
+     */
+    
+    this = find_node(tb,key);
+    if (this == NULL) {
+	return DB_ERROR_BADKEY;
+    } else {
+	Eterm element;
+	Uint sz;
+	if (ndex > arityval(this->dbterm.tpl[0])) {
+	    return DB_ERROR_BADPARAM;
+	}
+	element = this->dbterm.tpl[ndex];
+	sz = size_object(element);
+	hp = HAlloc(p, sz);
+	*ret = copy_struct(element, 
+			   sz, 
+			   &hp, 
+			   &MSO(p));
+    }
+    return DB_ERROR_NONE;
+}
+
+static int db_erase_tree(DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    TreeDbTerm *res;
+
+    *ret = am_true;
+
+    if ((res = linkout_tree(tb, key)) != NULL) {
+	free_term(tb, res);
+    }
+    return DB_ERROR_NONE;
+}
+
+static int db_erase_object_tree(DbTable *tbl, Eterm object, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    TreeDbTerm *res;
+
+    *ret = am_true;
+
+    if ((res = linkout_object_tree(tb, object)) != NULL) {
+	free_term(tb, res);
+    }
+    return DB_ERROR_NONE;
+}
+
+
+static int db_slot_tree(Process *p, DbTable *tbl, 
+			Eterm slot_term, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    Sint slot;
+    TreeDbTerm *st;
+    Eterm *hp;
+    Eterm copy;
+
+    /*
+     * The notion of a "slot" is not natural in a tree, but we try to
+     * simulate it by giving the n'th node in the tree instead.
+     * Traversing a tree in this way is not very convenient, but by
+     * using the saved stack we at least sometimes will get acceptable 
+     * performance.
+     */
+
+    if (is_not_small(slot_term) ||
+	((slot = signed_val(slot_term)) < 0) ||
+	(slot > NITEMS(tb)))
+	return DB_ERROR_BADPARAM;
+
+    if (slot == NITEMS(tb)) {
+	*ret = am_EOT;
+	return DB_ERROR_NONE;
+    }
+
+    /* 
+     * We use the slot position and search from there, slot positions 
+     * are counted from 1 and up.
+     */
+    ++slot;
+    st = slot_search(p, tb, slot); 
+    if (st == NULL) {
+	*ret = am_false;
+	return DB_ERROR_UNSPEC;
+    }
+    hp = HAlloc(p, st->dbterm.size + 2);
+    copy = copy_shallow(DBTERM_BUF(&st->dbterm), 
+			st->dbterm.size, 
+			&hp, 
+			&MSO(p));
+    *ret = CONS(hp, copy, NIL);
+    return DB_ERROR_NONE;
+}
+
+
+
+static BIF_RETTYPE ets_select_reverse(Process *p, Eterm a1, Eterm a2, Eterm a3)
+{
+    Eterm list;
+    Eterm result;
+    Eterm* hp;
+    Eterm* hend;
+
+    int max_iter = CONTEXT_REDS * 10;
+
+    if (is_nil(a1)) {
+	hp = HAlloc(p, 3);
+	BIF_RET(TUPLE2(hp,a2,a3));
+    } else if (is_not_list(a1)) {
+    error:
+	BIF_ERROR(p, BADARG);
+    }
+    
+    list = a1;
+    result = a2;
+    hp = hend = NULL;
+    while (is_list(list)) {
+	Eterm* pair = list_val(list);
+	if (--max_iter == 0) {
+	    BUMP_ALL_REDS(p);
+	    HRelease(p, hend, hp);
+	    BIF_TRAP3(&ets_select_reverse_exp, p, list, result, a3);
+	}
+	if (hp == hend) {
+	    hp = HAlloc(p, 64);
+	    hend = hp + 64;
+	}
+	result = CONS(hp, CAR(pair), result);
+	hp += 2;
+	list = CDR(pair);
+    }
+    if (is_not_nil(list))  {
+	goto error;
+    }
+    HRelease(p, hend, hp);
+    BUMP_REDS(p,CONTEXT_REDS - max_iter / 10);
+    hp = HAlloc(p,3);
+    BIF_RET(TUPLE2(hp, result, a3));
+}
+
+static BIF_RETTYPE bif_trap1(Export *bif,
+			     Process *p, 
+			     Eterm p1) 
+{
+    BIF_TRAP1(bif, p, p1);
+}
+    
+static BIF_RETTYPE bif_trap3(Export *bif,
+			     Process *p, 
+			     Eterm p1, 
+			     Eterm p2,
+			     Eterm p3) 
+{
+    BIF_TRAP3(bif, p, p1, p2, p3);
+}
+    
+/*
+** This is called either when the select bif traps or when ets:select/1 
+** is called. It does mostly the same as db_select_tree and may in either case
+** trap to itself again (via the ets:select/1 bif).
+** Note that this is common for db_select_tree and db_select_chunk_tree.
+*/
+static int db_select_continue_tree(Process *p, 
+				   DbTable *tbl,
+				   Eterm continuation,
+				   Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_context sc;
+    unsigned sz;
+    Eterm *hp; 
+    Eterm lastkey;
+    Eterm end_condition; 
+    Binary *mp;
+    Eterm key;
+    Eterm *tptr;
+    Sint chunk_size;
+    Sint reverse;
+
+
+#define RET_TO_BIF(Term, State) do { *ret = (Term); return State; } while(0);
+
+    /* Decode continuation. We know it's a tuple but not the arity or 
+       anything else */
+
+    tptr = tuple_val(continuation);
+
+    if (arityval(*tptr) != 8)
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    
+    if (!is_small(tptr[4]) || !is_binary(tptr[5]) || 
+	!(is_list(tptr[6]) || tptr[6] == NIL) || !is_small(tptr[7]) ||
+	!is_small(tptr[8]))
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    
+    lastkey = tptr[2];
+    end_condition = tptr[3];
+    if (!(thing_subtag(*binary_val(tptr[5])) == REFC_BINARY_SUBTAG))
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    mp = ((ProcBin *) binary_val(tptr[5]))->val;
+    if (!IsMatchProgBinary(mp))
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    chunk_size = signed_val(tptr[4]);
+
+    sc.p = p;
+    sc.accum = tptr[6];
+    sc.mp = mp;
+    sc.end_condition = NIL;
+    sc.lastobj = NULL;
+    sc.max = 1000;
+    sc.keypos = tb->common.keypos;
+    sc.all_objects = mp->flags & BIN_FLAG_ALL_OBJECTS;
+    sc.chunk_size = chunk_size;
+    reverse = unsigned_val(tptr[7]);
+    sc.got = signed_val(tptr[8]);
+
+    stack = get_any_stack(tb);
+    if (chunk_size) {
+	if (reverse) {
+	    traverse_backwards(tb, stack, lastkey, &doit_select_chunk, &sc); 
+	} else {
+	    traverse_forward(tb, stack, lastkey, &doit_select_chunk, &sc); 
+	}
+    } else {
+	if (reverse) {
+	    traverse_forward(tb, stack, lastkey, &doit_select, &sc);
+	} else {
+	    traverse_backwards(tb, stack, lastkey, &doit_select, &sc);
+	}
+    }
+    release_stack(tb,stack);
+
+    BUMP_REDS(p, 1000 - sc.max);
+
+    if (sc.max > 0 || (chunk_size && sc.got == chunk_size)) {
+	if (chunk_size) {
+	    Eterm *hp; 
+	    unsigned sz;
+
+	    if (sc.got < chunk_size || sc.lastobj == NULL) { 
+		/* end of table, sc.lastobj may be NULL as we may have been
+		   at the very last object in the table when trapping. */
+		if (!sc.got) {
+		    RET_TO_BIF(am_EOT, DB_ERROR_NONE);
+		} else {
+		    RET_TO_BIF(bif_trap3(&ets_select_reverse_exp, p,
+					 sc.accum, NIL, am_EOT), 
+			       DB_ERROR_NONE);
+		}
+	    }
+
+	    key = GETKEY(tb, sc.lastobj);
+
+	    sz = size_object(key);
+	    hp = HAlloc(p, 9 + sz);
+	    key = copy_struct(key, sz, &hp, &MSO(p));
+	    continuation = TUPLE8
+		(hp,
+		 tptr[1],
+		 key,
+		 tptr[3], 
+		 tptr[4],
+		 tptr[5],
+		 NIL,
+		 tptr[7],
+		 make_small(0));
+	    RET_TO_BIF(bif_trap3(&ets_select_reverse_exp, p,
+				 sc.accum, NIL, continuation), 
+		       DB_ERROR_NONE);
+	} else {
+	    RET_TO_BIF(sc.accum, DB_ERROR_NONE);
+	}
+    }	
+    key = GETKEY(tb, sc.lastobj);
+    if (chunk_size) {
+	if (end_condition != NIL && 
+	    ((!reverse && cmp_partly_bound(end_condition,key) < 0) ||
+	     (reverse && cmp_partly_bound(end_condition,key) > 0))) { 
+	    /* done anyway */
+	    if (!sc.got) {
+		RET_TO_BIF(am_EOT, DB_ERROR_NONE);
+	    } else {
+		RET_TO_BIF(bif_trap3(&ets_select_reverse_exp, p, 
+				     sc.accum, NIL, am_EOT), 
+			   DB_ERROR_NONE);
+	    }
+	}
+    } else {
+	if (end_condition != NIL && 
+	    ((!reverse && cmp_partly_bound(end_condition,key) > 0) ||
+	     (reverse && cmp_partly_bound(end_condition,key) < 0))) { 
+	    /* done anyway */
+	    RET_TO_BIF(sc.accum,DB_ERROR_NONE);
+	}
+    }
+    /* Not done yet, let's trap. */
+    sz = size_object(key);
+    hp = HAlloc(p, 9 + sz);
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    continuation = TUPLE8
+	(hp,
+	 tptr[1],
+	 key,
+	 tptr[3], 
+	 tptr[4],
+	 tptr[5],
+	 sc.accum,
+	 tptr[7],
+	 make_small(sc.got));
+    RET_TO_BIF(bif_trap1(bif_export[BIF_ets_select_1], p, continuation), 
+	       DB_ERROR_NONE);
+
+#undef RET_TO_BIF
+}
+
+
+static int db_select_tree(Process *p, DbTable *tbl, 
+			  Eterm pattern, int reverse, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_context sc;
+    struct mp_info mpi;
+    Eterm lastkey = NIL;
+    Eterm key;
+    Eterm continuation;
+    unsigned sz;
+    Eterm *hp; 
+    TreeDbTerm *this;
+    int errcode;
+    Eterm mpb;
+
+
+#define RET_TO_BIF(Term,RetVal) do { 	       	\
+	if (mpi.mp != NULL) {			\
+	    erts_bin_free(mpi.mp);       	\
+	}					\
+	*ret = (Term); 				\
+	return RetVal; 			        \
+    } while(0)
+
+    mpi.mp = NULL;
+
+    sc.accum = NIL;
+    sc.lastobj = NULL;
+    sc.p = p;
+    sc.max = 1000; 
+    sc.end_condition = NIL;
+    sc.keypos = tb->common.keypos;
+    sc.got = 0;
+    sc.chunk_size = 0;
+
+    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+	RET_TO_BIF(NIL,errcode);
+    }
+
+    if (!mpi.something_can_match) {
+	RET_TO_BIF(NIL,DB_ERROR_NONE);  
+	/* can't possibly match anything */
+    }
+
+    sc.mp = mpi.mp;
+    sc.all_objects = mpi.all_objects;
+
+    if (!mpi.got_partial && mpi.some_limitation && 
+	cmp(mpi.least,mpi.most) == 0) {
+	doit_select(tb,mpi.save_term,&sc,0 /* direction doesn't matter */);
+	RET_TO_BIF(sc.accum,DB_ERROR_NONE);
+    }
+
+    stack = get_any_stack(tb);
+    if (reverse) {
+	if (mpi.some_limitation) {
+	    if ((this = find_prev_from_pb_key(tb, stack, mpi.least)) != NULL) {
+		lastkey = GETKEY(tb, this->dbterm.tpl);
+	    }
+	    sc.end_condition = mpi.most;
+	}
+	
+	traverse_forward(tb, stack, lastkey, &doit_select, &sc);
+    } else {
+	if (mpi.some_limitation) {
+	    if ((this = find_next_from_pb_key(tb, stack, mpi.most)) != NULL) {
+		lastkey = GETKEY(tb, this->dbterm.tpl);
+	    }
+	    sc.end_condition = mpi.least;
+	}
+	
+	traverse_backwards(tb, stack, lastkey, &doit_select, &sc);
+    }
+    release_stack(tb,stack);
+#ifdef HARDDEBUG
+	erts_fprintf(stderr,"Least: %T\n", mpi.least);
+	erts_fprintf(stderr,"Most: %T\n", mpi.most);
+#endif
+    BUMP_REDS(p, 1000 - sc.max);
+    if (sc.max > 0) {
+	RET_TO_BIF(sc.accum,DB_ERROR_NONE);
+    }
+
+    key = GETKEY(tb, sc.lastobj);
+    sz = size_object(key);
+    hp = HAlloc(p, 9 + sz + PROC_BIN_SIZE);
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    if (mpi.all_objects)
+	(mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
+    mpb=db_make_mp_binary(p,mpi.mp,&hp);
+	    
+    continuation = TUPLE8
+	(hp,
+	 tb->common.id,
+	 key,
+	 sc.end_condition, /* From the match program, needn't be copied */
+	 make_small(0), /* Chunk size of zero means not chunked to the
+			   continuation BIF */
+	 mpb,
+	 sc.accum,
+	 make_small(reverse),
+	 make_small(sc.got));
+
+    /* Don't free mpi.mp, so don't use macro */
+    *ret = bif_trap1(bif_export[BIF_ets_select_1], p, continuation); 
+    return DB_ERROR_NONE;
+
+#undef RET_TO_BIF
+
+}
+
+    
+/*
+** This is called either when the select_count bif traps.
+*/
+static int db_select_count_continue_tree(Process *p, 
+					 DbTable *tbl,
+					 Eterm continuation,
+					 Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_count_context sc;
+    unsigned sz;
+    Eterm *hp; 
+    Eterm lastkey;
+    Eterm end_condition; 
+    Binary *mp;
+    Eterm key;
+    Eterm *tptr;
+    Eterm egot;
+
+
+#define RET_TO_BIF(Term, State) do { *ret = (Term); return State; } while(0);
+
+    /* Decode continuation. We know it's a tuple and everything else as
+     this is only called by ourselves */
+
+    /* continuation: 
+       {Table, Lastkey, EndCondition, MatchProgBin, HowManyGot}*/
+
+    tptr = tuple_val(continuation);
+
+    if (arityval(*tptr) != 5)
+	erl_exit(1,"Internal error in ets:select_count/1");
+    
+    lastkey = tptr[2];
+    end_condition = tptr[3];
+    if (!(thing_subtag(*binary_val(tptr[4])) == REFC_BINARY_SUBTAG))
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    mp = ((ProcBin *) binary_val(tptr[4]))->val;
+    if (!IsMatchProgBinary(mp))
+	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+
+    sc.p = p;
+    sc.mp = mp;
+    sc.end_condition = NIL;
+    sc.lastobj = NULL;
+    sc.max = 1000;
+    sc.keypos = tb->common.keypos;
+    if (is_big(tptr[5])) {
+	sc.got = big_to_uint32(tptr[5]);
+    } else {
+	sc.got = unsigned_val(tptr[5]);
+    }
+
+    stack = get_any_stack(tb);
+    traverse_backwards(tb, stack, lastkey, &doit_select_count, &sc);
+    release_stack(tb,stack);
+
+    BUMP_REDS(p, 1000 - sc.max);
+
+    if (sc.max > 0) {
+	RET_TO_BIF(erts_make_integer(sc.got,p), DB_ERROR_NONE);
+    }	
+    key = GETKEY(tb, sc.lastobj);
+    if (end_condition != NIL && 
+	(cmp_partly_bound(end_condition,key) > 0)) { 
+	/* done anyway */
+	RET_TO_BIF(make_small(sc.got),DB_ERROR_NONE);
+    }
+    /* Not done yet, let's trap. */
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.got)) {
+	hp = HAlloc(p, sz + 6);
+	egot = make_small(sc.got);
+    }
+    else {
+	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + 6);
+	egot = uint_to_big(sc.got, hp);
+	hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    continuation = TUPLE5
+	(hp,
+	 tptr[1],
+	 key,
+	 tptr[3], 
+	 tptr[4],
+	 egot);
+    RET_TO_BIF(bif_trap1(&ets_select_count_continue_exp, p, continuation), 
+	       DB_ERROR_NONE);
+
+#undef RET_TO_BIF
+}
+
+
+static int db_select_count_tree(Process *p, DbTable *tbl, 
+				Eterm pattern, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_count_context sc;
+    struct mp_info mpi;
+    Eterm lastkey = NIL;
+    Eterm key;
+    Eterm continuation;
+    unsigned sz;
+    Eterm *hp; 
+    TreeDbTerm *this;
+    int errcode;
+    Eterm egot;
+    Eterm mpb;
+
+
+#define RET_TO_BIF(Term,RetVal) do { 	       	\
+	if (mpi.mp != NULL) {			\
+	    erts_bin_free(mpi.mp);       	\
+	}					\
+	*ret = (Term); 				\
+	return RetVal; 			        \
+    } while(0)
+
+    mpi.mp = NULL;
+
+    sc.lastobj = NULL;
+    sc.p = p;
+    sc.max = 1000; 
+    sc.end_condition = NIL;
+    sc.keypos = tb->common.keypos;
+    sc.got = 0;
+
+    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+	RET_TO_BIF(NIL,errcode);
+    }
+
+    if (!mpi.something_can_match) {
+	RET_TO_BIF(make_small(0),DB_ERROR_NONE);  
+	/* can't possibly match anything */
+    }
+
+    sc.mp = mpi.mp;
+    sc.all_objects = mpi.all_objects;
+
+    if (!mpi.got_partial && mpi.some_limitation && 
+	cmp(mpi.least,mpi.most) == 0) {
+	doit_select_count(tb,mpi.save_term,&sc,0 /* dummy */);
+	RET_TO_BIF(erts_make_integer(sc.got,p),DB_ERROR_NONE);
+    }
+
+    stack = get_any_stack(tb);
+    if (mpi.some_limitation) {
+	if ((this = find_next_from_pb_key(tb, stack, mpi.most)) != NULL) {
+	    lastkey = GETKEY(tb, this->dbterm.tpl);
+	}
+	sc.end_condition = mpi.least;
+    }
+    
+    traverse_backwards(tb, stack, lastkey, &doit_select_count, &sc);
+    release_stack(tb,stack);
+    BUMP_REDS(p, 1000 - sc.max);
+    if (sc.max > 0) {
+	RET_TO_BIF(erts_make_integer(sc.got,p),DB_ERROR_NONE);
+    }
+
+    key = GETKEY(tb, sc.lastobj);
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.got)) {
+	hp = HAlloc(p, sz + PROC_BIN_SIZE + 6);
+	egot = make_small(sc.got);
+    }
+    else {
+	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + PROC_BIN_SIZE + 6);
+	egot = uint_to_big(sc.got, hp);
+	hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    if (mpi.all_objects)
+	(mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
+    mpb = db_make_mp_binary(p,mpi.mp,&hp);
+	    
+    continuation = TUPLE5
+	(hp,
+	 tb->common.id,
+	 key,
+	 sc.end_condition, /* From the match program, needn't be copied */
+	 mpb,
+	 egot);
+
+    /* Don't free mpi.mp, so don't use macro */
+    *ret = bif_trap1(&ets_select_count_continue_exp, p, continuation); 
+    return DB_ERROR_NONE;
+
+#undef RET_TO_BIF
+
+}
+
+static int db_select_chunk_tree(Process *p, DbTable *tbl, 
+				Eterm pattern, Sint chunk_size,
+				int reverse,
+				Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_context sc;
+    struct mp_info mpi;
+    Eterm lastkey = NIL;
+    Eterm key;
+    Eterm continuation;
+    unsigned sz;
+    Eterm *hp; 
+    TreeDbTerm *this;
+    int errcode;
+    Eterm mpb;
+
+
+#define RET_TO_BIF(Term,RetVal) do { 		\
+	if (mpi.mp != NULL) {			\
+	    erts_bin_free(mpi.mp);		\
+	}					\
+	*ret = (Term); 				\
+	return RetVal; 			        \
+    } while(0)
+
+    mpi.mp = NULL;
+
+    sc.accum = NIL;
+    sc.lastobj = NULL;
+    sc.p = p;
+    sc.max = 1000; 
+    sc.end_condition = NIL;
+    sc.keypos = tb->common.keypos;
+    sc.got = 0;
+    sc.chunk_size = chunk_size;
+
+    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+	RET_TO_BIF(NIL,errcode);
+    }
+
+    if (!mpi.something_can_match) {
+	RET_TO_BIF(am_EOT,DB_ERROR_NONE);
+	/* can't possibly match anything */
+    }
+
+    sc.mp = mpi.mp;
+    sc.all_objects = mpi.all_objects;
+
+    if (!mpi.got_partial && mpi.some_limitation && 
+	cmp(mpi.least,mpi.most) == 0) {
+	doit_select(tb,mpi.save_term,&sc, 0 /* direction doesn't matter */);
+	if (sc.accum != NIL) {
+	    hp=HAlloc(p, 3);
+	    RET_TO_BIF(TUPLE2(hp,sc.accum,am_EOT),DB_ERROR_NONE);
+	} else {
+	    RET_TO_BIF(am_EOT,DB_ERROR_NONE);
+	}
+    }
+
+    stack = get_any_stack(tb);
+    if (reverse) {
+	if (mpi.some_limitation) {
+	    if ((this = find_next_from_pb_key(tb, stack, mpi.most)) != NULL) {
+		lastkey = GETKEY(tb, this->dbterm.tpl);
+	    }
+	    sc.end_condition = mpi.least;
+	}
+
+	traverse_backwards(tb, stack, lastkey, &doit_select_chunk, &sc);
+    } else {
+	if (mpi.some_limitation) {
+	    if ((this = find_prev_from_pb_key(tb, stack, mpi.least)) != NULL) {
+		lastkey = GETKEY(tb, this->dbterm.tpl);
+	    }
+	    sc.end_condition = mpi.most;
+	}
+
+	traverse_forward(tb, stack, lastkey, &doit_select_chunk, &sc);
+    }
+    release_stack(tb,stack);
+
+    BUMP_REDS(p, 1000 - sc.max);
+    if (sc.max > 0 || sc.got == chunk_size) {
+	Eterm *hp; 
+	unsigned sz;
+
+	if (sc.got < chunk_size ||
+	    sc.lastobj == NULL) { 
+	    /* We haven't got all and we haven't trapped 
+	       which should mean we are at the end of the 
+	       table, sc.lastobj may be NULL if the table was empty */
+	    
+	    if (!sc.got) {
+		RET_TO_BIF(am_EOT, DB_ERROR_NONE);
+	    } else {
+		RET_TO_BIF(bif_trap3(&ets_select_reverse_exp, p,
+				     sc.accum, NIL, am_EOT), 
+			   DB_ERROR_NONE);
+	    }
+	}
+
+	key = GETKEY(tb, sc.lastobj);
+	sz = size_object(key);
+	hp = HAlloc(p, 9 + sz + PROC_BIN_SIZE);
+	key = copy_struct(key, sz, &hp, &MSO(p));
+	if (mpi.all_objects)
+	    (mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
+	mpb = db_make_mp_binary(p,mpi.mp,&hp);
+	
+	continuation = TUPLE8
+	    (hp,
+	     tb->common.id,
+	     key,
+	     sc.end_condition, /* From the match program, 
+				  needn't be copied */
+	     make_small(chunk_size),
+	     mpb,
+	     NIL,
+	     make_small(reverse),
+	     make_small(0));
+	/* Don't let RET_TO_BIF macro free mpi.mp*/
+	*ret = bif_trap3(&ets_select_reverse_exp, p,
+			 sc.accum, NIL, continuation);
+	return DB_ERROR_NONE; 
+    }
+
+    key = GETKEY(tb, sc.lastobj);
+    sz = size_object(key);
+    hp = HAlloc(p, 9 + sz + PROC_BIN_SIZE);
+    key = copy_struct(key, sz, &hp, &MSO(p));
+
+    if (mpi.all_objects)
+	(mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
+    mpb = db_make_mp_binary(p,mpi.mp,&hp);    
+    continuation = TUPLE8
+	(hp,
+	 tb->common.id,
+	 key,
+	 sc.end_condition, /* From the match program, needn't be copied */
+	 make_small(chunk_size),
+	 mpb,
+	 sc.accum,
+	 make_small(reverse),
+	 make_small(sc.got));
+    /* Don't let RET_TO_BIF macro free mpi.mp*/
+    *ret = bif_trap1(bif_export[BIF_ets_select_1], p, continuation);
+    return DB_ERROR_NONE;
+
+#undef RET_TO_BIF
+
+}
+
+/*
+** This is called when select_delete traps
+*/
+static int db_select_delete_continue_tree(Process *p, 
+					  DbTable *tbl,
+					  Eterm continuation,
+					  Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    struct select_delete_context sc;
+    unsigned sz;
+    Eterm *hp; 
+    Eterm lastkey;
+    Eterm end_condition; 
+    Binary *mp;
+    Eterm key;
+    Eterm *tptr;
+    Eterm eaccsum;
+
+
+#define RET_TO_BIF(Term, State) do { 		\
+	if (sc.erase_lastterm) {		\
+	    free_term(tb, sc.lastterm);		\
+	}					\
+	*ret = (Term); 				\
+	return State; 				\
+    } while(0);
+
+    /* Decode continuation. We know it's correct, this can only be called
+       by trapping */
+
+    tptr = tuple_val(continuation);
+
+    lastkey = tptr[2];
+    end_condition = tptr[3];
+
+    sc.erase_lastterm = 0; /* Before first RET_TO_BIF */
+    sc.lastterm = NULL;
+
+    mp = ((ProcBin *) binary_val(tptr[4]))->val;
+    sc.p = p;
+    sc.tb = tb;
+    if (is_big(tptr[5])) {
+	sc.accum = big_to_uint32(tptr[5]);
+    } else {
+	sc.accum = unsigned_val(tptr[5]);
+    }
+    sc.mp = mp;
+    sc.end_condition = NIL;
+    sc.max = 1000;
+    sc.keypos = tb->common.keypos;
+
+    ASSERT(!erts_smp_atomic_read(&tb->is_stack_busy));
+    traverse_backwards(tb, &tb->static_stack, lastkey, &doit_select_delete, &sc);
+
+    BUMP_REDS(p, 1000 - sc.max);
+
+    if (sc.max > 0) {
+	RET_TO_BIF(erts_make_integer(sc.accum, p), DB_ERROR_NONE);
+    }	
+    key = GETKEY(tb, (sc.lastterm)->dbterm.tpl);
+    if (end_condition != NIL && 
+	cmp_partly_bound(end_condition,key) > 0) { /* done anyway */
+	RET_TO_BIF(erts_make_integer(sc.accum,p),DB_ERROR_NONE);
+    }
+    /* Not done yet, let's trap. */
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.accum)) {
+	hp = HAlloc(p, sz + 6);
+	eaccsum = make_small(sc.accum);
+    }
+    else {
+	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + 6);
+	eaccsum = uint_to_big(sc.accum, hp);
+	hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    continuation = TUPLE5
+	(hp,
+	 tptr[1],
+	 key,
+	 tptr[3], 
+	 tptr[4],
+	 eaccsum);
+    RET_TO_BIF(bif_trap1(&ets_select_delete_continue_exp, p, continuation), 
+	       DB_ERROR_NONE);
+
+#undef RET_TO_BIF
+}
+
+static int db_select_delete_tree(Process *p, DbTable *tbl, 
+				 Eterm pattern, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    struct select_delete_context sc;
+    struct mp_info mpi;
+    Eterm lastkey = NIL;
+    Eterm key;
+    Eterm continuation;
+    unsigned sz;
+    Eterm *hp; 
+    TreeDbTerm *this;
+    int errcode;
+    Eterm mpb;
+    Eterm eaccsum;
+
+#define RET_TO_BIF(Term,RetVal) do { 	       	\
+	if (mpi.mp != NULL) {			\
+	    erts_bin_free(mpi.mp);       	\
+	}					\
+	if (sc.erase_lastterm) {                \
+	    free_term(tb, sc.lastterm);         \
+	}                                       \
+	*ret = (Term); 				\
+	return RetVal; 			        \
+    } while(0)
+
+    mpi.mp = NULL;
+
+    sc.accum = 0;
+    sc.erase_lastterm = 0;
+    sc.lastterm = NULL;
+    sc.p = p;
+    sc.max = 1000; 
+    sc.end_condition = NIL;
+    sc.keypos = tb->common.keypos;
+    sc.tb = tb;
+    
+    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+	RET_TO_BIF(0,errcode);
+    }
+
+    if (!mpi.something_can_match) {
+	RET_TO_BIF(make_small(0),DB_ERROR_NONE);  
+	/* can't possibly match anything */
+    }
+
+    sc.mp = mpi.mp;
+
+    if (!mpi.got_partial && mpi.some_limitation && 
+	cmp(mpi.least,mpi.most) == 0) {
+	doit_select_delete(tb,mpi.save_term,&sc, 0 /* direction doesn't 
+						      matter */);
+	RET_TO_BIF(erts_make_integer(sc.accum,p),DB_ERROR_NONE);
+    }
+
+    if (mpi.some_limitation) {
+	if ((this = find_next_from_pb_key(tb, &tb->static_stack, mpi.most)) != NULL) {
+	    lastkey = GETKEY(tb, this->dbterm.tpl);
+	}
+	sc.end_condition = mpi.least;
+    }
+
+    traverse_backwards(tb, &tb->static_stack, lastkey, &doit_select_delete, &sc);
+    BUMP_REDS(p, 1000 - sc.max);
+
+    if (sc.max > 0) {
+	RET_TO_BIF(erts_make_integer(sc.accum,p), DB_ERROR_NONE);
+    }
+
+    key = GETKEY(tb, (sc.lastterm)->dbterm.tpl);
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.accum)) {
+	hp = HAlloc(p, sz + PROC_BIN_SIZE + 6);
+	eaccsum = make_small(sc.accum);
+    }
+    else {
+	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + PROC_BIN_SIZE + 6);
+	eaccsum = uint_to_big(sc.accum, hp);
+	hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    mpb = db_make_mp_binary(p,mpi.mp,&hp);
+    
+    continuation = TUPLE5
+	(hp,
+	 tb->common.id,
+	 key,
+	 sc.end_condition, /* From the match program, needn't be copied */
+	 mpb,
+	 eaccsum);
+
+    /* Don't free mpi.mp, so don't use macro */
+    if (sc.erase_lastterm) {
+	free_term(tb, sc.lastterm);
+    }
+    *ret = bif_trap1(&ets_select_delete_continue_exp, p, continuation); 
+    return DB_ERROR_NONE;
+
+#undef RET_TO_BIF
+
+}
+
+/*
+** Other interface routines (not directly coupled to one bif)
+*/
+
+/* Display hash table contents (for dump) */
+static void db_print_tree(int to, void *to_arg, 
+			  int show,
+			  DbTable *tbl)
+{
+    DbTableTree *tb = &tbl->tree;
+#ifdef TREE_DEBUG
+    if (show)
+	erts_print(to, to_arg, "\nTree data dump:\n"
+		   "------------------------------------------------\n");
+    do_dump_tree2(to, to_arg, show, tb->root, 0);
+    if (show)
+	erts_print(to, to_arg, "\n"
+		   "------------------------------------------------\n");
+#else
+    erts_print(to, to_arg, "Ordered set (AVL tree), Elements: %d\n", NITEMS(tb));
+    do_dump_tree(to, to_arg, tb->root);
+#endif
+}
+
+/* release all memory occupied by a single table */
+static int db_free_table_tree(DbTable *tbl)
+{
+    while (!db_free_table_continue_tree(tbl))
+	;
+    return 1;
+}
+
+static int db_free_table_continue_tree(DbTable *tbl)
+{
+    DbTableTree *tb = &tbl->tree;
+    int result;
+
+    if (!tb->deletion) {
+	tb->static_stack.pos = 0;
+	tb->deletion = 1;
+	PUSH_NODE(&tb->static_stack, tb->root);
+    }
+    result = do_free_tree_cont(tb, DELETE_RECORD_LIMIT);
+    if (result) {		/* Completely done. */
+	erts_db_free(ERTS_ALC_T_DB_STK,
+		     (DbTable *) tb,
+		     (void *) tb->static_stack.array,
+		     sizeof(TreeDbTerm *) * STACK_NEED);
+	ASSERT(erts_smp_atomic_read(&tb->common.memory_size)
+	       == sizeof(DbTable));
+    }
+    return result;
+}
+
+static int db_delete_all_objects_tree(Process* p, DbTable* tbl)
+{
+    db_free_table_tree(tbl);
+    db_create_tree(p, tbl);
+    erts_smp_atomic_set(&tbl->tree.common.nitems, 0);
+    return 0;
+}
+
+static void do_db_tree_foreach_offheap(TreeDbTerm *,
+				       void (*)(ErlOffHeap *, void *),
+				       void *);
+
+static void db_foreach_offheap_tree(DbTable *tbl,
+				    void (*func)(ErlOffHeap *, void *),
+				    void * arg)
+{
+    do_db_tree_foreach_offheap(tbl->tree.root, func, arg);
+}
+
+
+/*
+** Functions for internal use
+*/
+
+
+static void
+do_db_tree_foreach_offheap(TreeDbTerm *tdbt,
+			   void (*func)(ErlOffHeap *, void *),
+			   void * arg)
+{
+    if(!tdbt)
+	return;
+    do_db_tree_foreach_offheap(tdbt->left, func, arg);
+    (*func)(&(tdbt->dbterm.off_heap), arg);
+    do_db_tree_foreach_offheap(tdbt->right, func, arg);
+}
+
+static TreeDbTerm *linkout_tree(DbTableTree *tb, 
+				Eterm key)
+{
+    TreeDbTerm **tstack[STACK_NEED];
+    int tpos = 0;
+    int dstack[STACK_NEED+1];
+    int dpos = 0;
+    int state = 0;
+    TreeDbTerm **this = &tb->root;
+    Sint c;
+    int dir;
+    TreeDbTerm *q = NULL;
+
+    /*
+     * Somewhat complicated, deletion in an AVL tree,
+     * The two helpers balance_left and balance_right are used to
+     * keep the balance. As in insert, we do the stacking ourselves.
+     */
+
+    reset_static_stack(tb);
+    dstack[dpos++] = DIR_END;
+    for (;;) {
+	if (!*this) { /* Failure */
+	    return NULL;
+	} else if ((c = cmp(key,GETKEY(tb,(*this)->dbterm.tpl))) < 0) { 
+	    dstack[dpos++] = DIR_LEFT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->left);
+	} else if (c > 0) { /* go right */
+	    dstack[dpos++] = DIR_RIGHT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->right);
+	} else { /* Equal key, found the one to delete*/
+	    q = (*this);
+	    if (q->right == NULL) {
+		(*this) = q->left;
+		state = 1;
+	    } else if (q->left == NULL) {
+		(*this) = q->right;
+		state = 1;
+	    } else {
+		dstack[dpos++] = DIR_LEFT;
+		tstack[tpos++] = this;
+		state = delsub(this);
+	    }
+	    erts_smp_atomic_dec(&tb->common.nitems);
+	    break;
+	}
+    }
+    while (state && ( dir = dstack[--dpos] ) != DIR_END) {
+	this = tstack[--tpos];
+	if (dir == DIR_LEFT) {
+	    state = balance_left(this);
+	} else {
+	    state = balance_right(this);
+	}
+    }
+    return q;
+}
+
+static TreeDbTerm *linkout_object_tree(DbTableTree *tb, 
+				       Eterm object)
+{
+    TreeDbTerm **tstack[STACK_NEED];
+    int tpos = 0;
+    int dstack[STACK_NEED+1];
+    int dpos = 0;
+    int state = 0;
+    TreeDbTerm **this = &tb->root;
+    Sint c;
+    int dir;
+    TreeDbTerm *q = NULL;
+    Eterm key;
+
+    /*
+     * Somewhat complicated, deletion in an AVL tree,
+     * The two helpers balance_left and balance_right are used to
+     * keep the balance. As in insert, we do the stacking ourselves.
+     */
+
+    
+    key = GETKEY(tb, tuple_val(object));
+
+    reset_static_stack(tb);
+    dstack[dpos++] = DIR_END;
+    for (;;) {
+	if (!*this) { /* Failure */
+	    return NULL;
+	} else if ((c = cmp(key,GETKEY(tb,(*this)->dbterm.tpl))) < 0) { 
+	    dstack[dpos++] = DIR_LEFT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->left);
+	} else if (c > 0) { /* go right */
+	    dstack[dpos++] = DIR_RIGHT;
+	    tstack[tpos++] = this;
+	    this = &((*this)->right);
+	} else { /* Equal key, found the only possible matching object*/
+	    if (!eq(object,make_tuple((*this)->dbterm.tpl))) {
+		return NULL;
+	    }
+	    q = (*this);
+	    if (q->right == NULL) {
+		(*this) = q->left;
+		state = 1;
+	    } else if (q->left == NULL) {
+		(*this) = q->right;
+		state = 1;
+	    } else {
+		dstack[dpos++] = DIR_LEFT;
+		tstack[tpos++] = this;
+		state = delsub(this);
+	    }
+	    erts_smp_atomic_dec(&tb->common.nitems);
+	    break;
+	}
+    }
+    while (state && ( dir = dstack[--dpos] ) != DIR_END) {
+	this = tstack[--tpos];
+	if (dir == DIR_LEFT) {
+	    state = balance_left(this);
+	} else {
+	    state = balance_right(this);
+	}
+    }
+    return q;
+}
+
+/*
+** For the select functions, analyzes the pattern and determines which
+** part of the tree should be searched. Also compiles the match program
+*/
+static int analyze_pattern(DbTableTree *tb, Eterm pattern, 
+			   struct mp_info *mpi)
+{
+    Eterm lst, tpl, ttpl;
+    Eterm *matches,*guards, *bodies;
+    Eterm sbuff[30];
+    Eterm *buff = sbuff;
+    Eterm *ptpl;
+    int i;
+    int num_heads = 0;
+    Eterm key;
+    Eterm partly_bound;
+    int res;
+    Eterm least = 0;
+    Eterm most = 0;
+
+    mpi->some_limitation = 1;
+    mpi->got_partial = 0;
+    mpi->something_can_match = 0;
+    mpi->mp = NULL;
+    mpi->all_objects = 1;
+    mpi->save_term = NULL;
+
+    for (lst = pattern; is_list(lst); lst = CDR(list_val(lst)))
+	++num_heads;
+
+    if (lst != NIL) {/* proper list... */
+	return DB_ERROR_BADPARAM;
+    }
+    if (num_heads > 10) {
+	buff = erts_alloc(ERTS_ALC_T_DB_TMP, sizeof(Eterm) * num_heads * 3);
+    }
+
+    matches = buff;
+    guards = buff + num_heads;
+    bodies = buff + (num_heads * 2);
+
+    i = 0;
+    for(lst = pattern; is_list(lst); lst = CDR(list_val(lst))) {
+	Eterm body;
+	ttpl = CAR(list_val(lst));
+	if (!is_tuple(ttpl)) {
+	    if (buff != sbuff) { 
+		erts_free(ERTS_ALC_T_DB_TMP, buff);
+	    }
+	    return DB_ERROR_BADPARAM;
+	}
+	ptpl = tuple_val(ttpl);
+	if (ptpl[0] != make_arityval(3U)) {
+	    if (buff != sbuff) { 
+		erts_free(ERTS_ALC_T_DB_TMP, buff);
+	    }
+	    return DB_ERROR_BADPARAM;
+	}
+	matches[i] = tpl = ptpl[1];
+	guards[i] = ptpl[2];
+	bodies[i] = body = ptpl[3];
+	if (!is_list(body) || CDR(list_val(body)) != NIL ||
+	    CAR(list_val(body)) != am_DollarUnderscore) {
+	    mpi->all_objects = 0;
+	}
+	++i;
+
+	partly_bound = NIL;
+	res = key_given(tb, tpl, &mpi->save_term, &partly_bound);
+	if ( res >= 0 ) {   /* Can match something */
+	    key = 0;
+	    mpi->something_can_match = 1;
+	    if (res > 0) {
+		key = GETKEY(tb,tuple_val(tpl)); 
+	    } else if (partly_bound != NIL) {
+		mpi->got_partial = 1;
+		key = partly_bound;
+	    } else {
+		mpi->some_limitation = 0;
+	    }
+	    if (key != 0) {
+		if (least == 0 || 
+		    partly_bound_can_match_lesser(key,least)) {
+		    least = key;
+		}
+		if (most == 0 || 
+		    partly_bound_can_match_greater(key,most)) {
+		    most = key;
+		}
+	    }
+	}
+    }
+    mpi->least = least;
+    mpi->most = most;
+
+    /*
+     * It would be nice not to compile the match_spec if nothing could match,
+     * but then the select calls would not fail like they should on bad 
+     * match specs that happen to specify non existent keys etc.
+     */
+    if ((mpi->mp = db_match_compile(matches, guards, bodies,
+				    num_heads, DCOMP_TABLE, NULL)) 
+	== NULL) {
+	if (buff != sbuff) { 
+	    erts_free(ERTS_ALC_T_DB_TMP, buff);
+	}
+	return DB_ERROR_BADPARAM;
+    }
+    if (buff != sbuff) { 
+	erts_free(ERTS_ALC_T_DB_TMP, buff);
+    }
+    return DB_ERROR_NONE;
+}
+
+static void do_dump_tree(int to, void *to_arg, TreeDbTerm *t)
+{
+    if (t != NULL) {
+	do_dump_tree(to, to_arg, t->left);
+	erts_print(to, to_arg, "%T\n", make_tuple(t->dbterm.tpl));
+	do_dump_tree(to, to_arg, t->right); 
+    }
+}
+
+static void free_term(DbTableTree *tb, TreeDbTerm* p)
+{
+    db_free_term_data(&(p->dbterm));
+    erts_db_free(ERTS_ALC_T_DB_TERM,
+		 (DbTable *) tb,
+		 (void *) p,
+		 SIZ_DBTERM(p)*sizeof(Uint));
+}
+
+static int do_free_tree_cont(DbTableTree *tb, int num_left)
+{
+    TreeDbTerm *root;
+    TreeDbTerm *p;
+
+    for (;;) {
+	root = POP_NODE(&tb->static_stack);
+	if (root == NULL) break;
+	for (;;) {
+	    if ((p = root->left) != NULL) {
+		root->left = NULL;
+		PUSH_NODE(&tb->static_stack, root);
+		root = p;
+	    } else if ((p = root->right) != NULL) {
+		root->right = NULL;
+		PUSH_NODE(&tb->static_stack, root);
+		root = p;
+	    } else {
+		free_term(tb, root);
+		if (--num_left > 0) {
+		    break;
+		} else {
+		    return 0;	/* Done enough for now */
+		}
+	    }
+	}
+    }
+    return 1;
+}
+
+static TreeDbTerm* get_term(DbTableTree *tb,
+			    TreeDbTerm* old, 
+			    Eterm obj) 
+{
+    TreeDbTerm* p = db_get_term((DbTableCommon *) tb,
+				(old != NULL) ? &(old->dbterm) : NULL, 
+				((char *) &(old->dbterm)) - ((char *) old),
+				obj);
+    return p;
+}
+
+/*
+ * Deletion helpers
+ */
+static int balance_left(TreeDbTerm **this) 
+{
+    TreeDbTerm *p, *p1, *p2;
+    int b1, b2, h = 1;
+    
+    p = *this;
+    switch (p->balance) {
+    case -1:
+	p->balance = 0;
+	break;
+    case 0:
+	p->balance = 1;
+	h = 0;
+	break;
+    case 1:
+	p1 = p->right;
+	b1 = p1->balance;
+	if (b1 >= 0) { /* Single RR rotation */
+	    p->right = p1->left;
+	    p1->left = p;
+	    if (b1 == 0) {
+		p->balance = 1;
+		p1->balance = -1;
+		h = 0;
+	    } else {
+		p->balance = p1->balance = 0;
+	    }
+	    (*this) = p1;
+	} else { /* Double RL rotation */
+	    p2 = p1->left;
+	    b2 = p2->balance;
+	    p1->left = p2->right;
+	    p2->right = p1;
+	    p->right = p2->left;
+	    p2->left = p;
+	    p->balance = (b2 == 1) ? -1 : 0;
+	    p1->balance = (b2 == -1) ? 1 : 0;
+	    p2->balance = 0;
+	    (*this) = p2;
+	}
+	break;
+    }
+    return h;
+}
+
+static int balance_right(TreeDbTerm **this) 
+{
+    TreeDbTerm *p, *p1, *p2;
+    int b1, b2, h = 1;
+    
+    p = *this;
+    switch (p->balance) {
+    case 1:
+	p->balance = 0;
+	break;
+    case 0:
+	p->balance = -1;
+	h = 0;
+	break;
+    case -1:
+	p1 = p->left;
+	b1 = p1->balance;
+	if (b1 <= 0) { /* Single LL rotation */
+	    p->left = p1->right;
+	    p1->right = p;
+	    if (b1 == 0) {
+		p->balance = -1;
+		p1->balance = 1;
+		h = 0;
+	    } else {
+		p->balance = p1->balance = 0;
+	    }
+	    (*this) = p1;
+	} else { /* Double LR rotation */
+	    p2 = p1->right;
+	    b2 = p2->balance;
+	    p1->right = p2->left;
+	    p2->left = p1;
+	    p->left = p2->right;
+	    p2->right = p;
+	    p->balance = (b2 == -1) ? 1 : 0;
+	    p1->balance = (b2 == 1) ? -1 : 0;
+	    p2->balance = 0;
+	    (*this) = p2;
+	}
+    }
+    return h;
+}
+
+static int delsub(TreeDbTerm **this) 
+{
+    TreeDbTerm **tstack[STACK_NEED];
+    int tpos = 0;
+    TreeDbTerm *q = (*this);
+    TreeDbTerm **r = &(q->left);
+    int h;
+
+    /*
+     * Walk down the tree to the right and search 
+     * for a void right child, pick that child out
+     * and return it to be put in the deleted 
+     * object's place.
+     */
+    
+    while ((*r)->right != NULL) {
+	tstack[tpos++] = r;
+	r = &((*r)->right);
+    }
+    *this = *r;
+    *r = (*r)->left;
+    (*this)->left = q->left;
+    (*this)->right = q->right;
+    (*this)->balance = q->balance;
+    tstack[0] = &((*this)->left);
+    h = 1;
+    while (tpos && h) {
+	r = tstack[--tpos];
+	h = balance_right(r);
+    }
+    return h;
+}
+
+/*
+ * Helper for db_slot
+ */
+
+static TreeDbTerm *slot_search(Process *p, DbTableTree *tb, Sint slot)
+{
+    TreeDbTerm *this;
+    TreeDbTerm *tmp;
+    DbTreeStack* stack = get_any_stack(tb);
+    ASSERT(stack != NULL);
+
+    if (slot == 1) { /* Don't search from where we are if we are 
+			looking for the first slot */
+	stack->slot = 0;
+    }
+
+    if (stack->slot == 0) { /* clear stack if slot positions 
+				are not recorded */
+	stack->pos = 0;
+    }
+    if (EMPTY_NODE(stack)) {
+	this = tb->root;
+	if (this == NULL)
+	    goto done;
+	while (this->left != NULL){
+	    PUSH_NODE(stack, this);
+	    this = this->left;
+	}
+	PUSH_NODE(stack, this);
+	stack->slot = 1;
+    }
+    this = TOP_NODE(stack);
+    while (stack->slot != slot && this != NULL) {
+	if (slot > stack->slot) {
+	    if (this->right != NULL) {
+		this = this->right;
+		while (this->left != NULL) {
+		    PUSH_NODE(stack, this);
+		    this = this->left;
+		}
+		PUSH_NODE(stack, this);
+	    } else {
+		for (;;) {
+		    tmp = POP_NODE(stack);
+		    this = TOP_NODE(stack);
+		    if (this == NULL || this->left == tmp)
+			break;
+		}
+	    }		
+	    ++(stack->slot);
+	} else {
+	    if (this->left != NULL) {
+		this = this->left;
+		while (this->right != NULL) {
+		    PUSH_NODE(stack, this);
+		    this = this->right;
+		}
+		PUSH_NODE(stack, this);
+	    } else {
+		for (;;) {
+		    tmp = POP_NODE(stack);
+		    this = TOP_NODE(stack);
+		    if (this == NULL || this->right == tmp)
+			break;
+		}
+	    }		
+	    --(stack->slot);
+	}
+    }
+done:
+    release_stack(tb,stack);
+    return this;
+}
+
+/*
+ * Find next and previous in sort order
+ */
+
+static TreeDbTerm *find_next(DbTableTree *tb, DbTreeStack* stack, Eterm key)
+{
+    TreeDbTerm *this;
+    TreeDbTerm *tmp;
+    Sint c;
+
+    if(( this = TOP_NODE(stack)) != NULL) {
+	if (!CMP_EQ(GETKEY(tb, this->dbterm.tpl),key)) {
+	    /* Start from the beginning */
+	    stack->pos = stack->slot = 0;
+	}
+    }
+    if (EMPTY_NODE(stack)) { /* Have to rebuild the stack */
+	if (( this = tb->root ) == NULL)
+	    return NULL;
+	for (;;) {
+	    PUSH_NODE(stack, this);
+	    if (( c = cmp(GETKEY(tb, this->dbterm.tpl),key) ) < 0) {
+		if (this->right == NULL) /* We are at the previos 
+					    and the element does
+					    not exist */
+		    break;
+		else
+		    this = this->right;
+	    } else if (c > 0) {
+		if (this->left == NULL) /* Done */
+		    return this;
+		else
+		    this = this->left;
+	    } else
+		break;
+	}
+    }
+    /* The next element from this... */
+    if (this->right != NULL) {
+	this = this->right;
+	PUSH_NODE(stack,this);
+	while (this->left != NULL) {
+	    this = this->left;
+	    PUSH_NODE(stack, this);
+	}
+	if (stack->slot > 0) 
+	    ++(stack->slot);
+    } else {
+	do {
+	    tmp = POP_NODE(stack);
+	    if (( this = TOP_NODE(stack)) == NULL) {
+		stack->slot = 0;
+		return NULL;
+	    }
+	} while (this->right == tmp);
+	if (stack->slot > 0) 
+	    ++(stack->slot);
+    }
+    return this;
+}
+
+static TreeDbTerm *find_prev(DbTableTree *tb, DbTreeStack* stack, Eterm key)
+{
+    TreeDbTerm *this;
+    TreeDbTerm *tmp;
+    Sint c;
+
+    if(( this = TOP_NODE(stack)) != NULL) {
+	if (!CMP_EQ(GETKEY(tb, this->dbterm.tpl),key)) {
+	    /* Start from the beginning */
+	    stack->pos = stack->slot = 0;
+	}
+    }
+    if (EMPTY_NODE(stack)) { /* Have to rebuild the stack */
+	if (( this = tb->root ) == NULL)
+	    return NULL;
+	for (;;) {
+	    PUSH_NODE(stack, this);
+	    if (( c = cmp(GETKEY(tb, this->dbterm.tpl),key) ) > 0) {
+		if (this->left == NULL) /* We are at the next 
+					   and the element does
+					   not exist */
+		    break;
+		else
+		    this = this->left;
+	    } else if (c < 0) {
+		if (this->right == NULL) /* Done */
+		    return this;
+		else
+		    this = this->right;
+	    } else
+		break;
+	}
+    }
+    /* The previous element from this... */
+    if (this->left != NULL) {
+	this = this->left;
+	PUSH_NODE(stack,this);
+	while (this->right != NULL) {
+	    this = this->right;
+	    PUSH_NODE(stack, this);
+	}
+	if (stack->slot > 0) 
+	    --(stack->slot);
+    } else {
+	do {
+	    tmp = POP_NODE(stack);
+	    if (( this = TOP_NODE(stack)) == NULL) {
+		stack->slot = 0;
+		return NULL;
+	    }
+	} while (this->left == tmp);
+	if (stack->slot > 0) 
+	    --(stack->slot);
+    }
+    return this;
+}
+
+static TreeDbTerm *find_next_from_pb_key(DbTableTree *tb, DbTreeStack* stack,
+					 Eterm key)
+{
+    TreeDbTerm *this;
+    TreeDbTerm *tmp;
+    Sint c;
+
+    /* spool the stack, we have to "re-search" */
+    stack->pos = stack->slot = 0;
+    if (( this = tb->root ) == NULL)
+	return NULL;
+    for (;;) {
+	PUSH_NODE(stack, this);
+	if (( c = cmp_partly_bound(key,GETKEY(tb, this->dbterm.tpl)) ) >= 0) {
+	    if (this->right == NULL) {
+		do {
+		    tmp = POP_NODE(stack);
+		    if (( this = TOP_NODE(stack)) == NULL) {
+			return NULL;
+		    }
+		} while (this->right == tmp);
+		return this;
+	    } else
+		this = this->right;
+	} else /*if (c < 0)*/ {
+	    if (this->left == NULL) /* Done */
+		return this;
+	    else
+		this = this->left;
+	} 
+    }
+}
+
+static TreeDbTerm *find_prev_from_pb_key(DbTableTree *tb, DbTreeStack* stack,
+					 Eterm key)
+{
+    TreeDbTerm *this;
+    TreeDbTerm *tmp;
+    Sint c;
+
+    /* spool the stack, we have to "re-search" */
+    stack->pos = stack->slot = 0;
+    if (( this = tb->root ) == NULL)
+	return NULL;
+    for (;;) {
+	PUSH_NODE(stack, this);
+	if (( c = cmp_partly_bound(key,GETKEY(tb, this->dbterm.tpl)) ) <= 0) {
+	    if (this->left == NULL) {
+		do {
+		    tmp = POP_NODE(stack);
+		    if (( this = TOP_NODE(stack)) == NULL) {
+			return NULL;
+		    }
+		} while (this->left == tmp);
+		return this;
+	    } else
+		this = this->left;
+	} else /*if (c < 0)*/ {
+	    if (this->right == NULL) /* Done */
+		return this;
+	    else
+		this = this->right;
+	} 
+    }
+}
+
+
+/*
+ * Just lookup a node
+ */
+static TreeDbTerm *find_node(DbTableTree *tb, Eterm key)
+{
+    TreeDbTerm *this;
+    Sint res;
+    DbTreeStack* stack = get_static_stack(tb);
+
+    if(!stack || EMPTY_NODE(stack) 
+       || !CMP_EQ(GETKEY(tb, ( this = TOP_NODE(stack) )->dbterm.tpl), key)) {
+
+	this = tb->root;
+	while (this != NULL && 
+	       ( res = cmp(key, GETKEY(tb, this->dbterm.tpl)) ) != 0) {
+	    if (res < 0)
+		this = this->left;
+	    else
+		this = this->right;
+	}
+    }
+    release_stack(tb,stack);
+    return this;
+}
+
+/*
+ * Lookup a node and return the address of the node pointer in the tree
+ */
+static TreeDbTerm **find_node2(DbTableTree *tb, Eterm key)
+{
+    TreeDbTerm **this;
+    Sint res;
+
+    this = &tb->root;
+    while ((*this) != NULL && 
+	   ( res = cmp(key, GETKEY(tb, (*this)->dbterm.tpl)) ) != 0) {
+	if (res < 0)
+	    this = &((*this)->left);
+	else
+	    this = &((*this)->right);
+    }
+    if (*this == NULL)
+	return NULL;
+    return this;
+}
+
+static int db_lookup_dbterm_tree(DbTable *tbl, Eterm key, DbUpdateHandle* handle)
+{
+    DbTableTree *tb = &tbl->tree;
+    TreeDbTerm **pp = find_node2(tb, key);
+
+    if (pp == NULL) return 0;
+
+    handle->tb = tbl;
+    handle->dbterm = &(*pp)->dbterm;
+    handle->bp = (void**) pp;
+    handle->new_size = (*pp)->dbterm.size;
+    handle->mustResize = 0;
+    return 1;
+}
+
+static void db_finalize_dbterm_tree(DbUpdateHandle* handle)
+{
+    if (handle->mustResize) {
+	Eterm* top;
+	Eterm copy;
+	DbTerm* newDbTerm;
+	DbTableTree *tb = &handle->tb->tree;
+	TreeDbTerm* oldp = (TreeDbTerm*) *handle->bp;
+	TreeDbTerm* newp = erts_db_alloc(ERTS_ALC_T_DB_TERM,
+					 handle->tb,
+					 sizeof(TreeDbTerm)+sizeof(Eterm)*(handle->new_size-1));
+	memcpy(newp, oldp, sizeof(TreeDbTerm)-sizeof(DbTerm));  /* copy only tree header */
+	*(handle->bp) = newp;
+	reset_static_stack(tb);
+	newDbTerm = &newp->dbterm;
+    
+	newDbTerm->size = handle->new_size;
+	newDbTerm->off_heap.mso = NULL;
+	newDbTerm->off_heap.externals = NULL;
+    #ifndef HYBRID /* FIND ME! */
+	newDbTerm->off_heap.funs = NULL;
+    #endif
+	newDbTerm->off_heap.overhead = 0;
+	
+	/* make a flat copy */
+	top = DBTERM_BUF(newDbTerm);
+	copy = copy_struct(make_tuple(handle->dbterm->tpl),
+			   handle->new_size,
+			   &top, &newDbTerm->off_heap);
+	DBTERM_SET_TPL(newDbTerm,tuple_val(copy));
+    
+	db_free_term_data(handle->dbterm);
+	erts_db_free(ERTS_ALC_T_DB_TERM,
+		     handle->tb,
+		     (void *) (((char *) handle->dbterm) - (sizeof(TreeDbTerm) - sizeof(DbTerm))),
+		     sizeof(TreeDbTerm) + sizeof(Eterm)*(handle->dbterm->size-1));
+    }
+#ifdef DEBUG
+    handle->dbterm = 0;
+#endif
+    return;
+}   
+
+/*
+ * Traverse the tree with a callback function, used by db_match_xxx
+ */
+static void traverse_backwards(DbTableTree *tb,
+			       DbTreeStack* stack,
+			       Eterm lastkey,
+			       int (*doit)(DbTableTree *,
+					   TreeDbTerm *,
+					   void *,
+					   int),
+			       void *context) 
+{
+    TreeDbTerm *this, *next;
+
+    if (lastkey == NIL) {
+	stack->pos = stack->slot = 0;
+	if (( this = tb->root ) == NULL) {
+	    return;
+	}
+	while (this != NULL) {
+	    PUSH_NODE(stack, this);
+	    this = this->right;
+	}
+	this = TOP_NODE(stack);
+	next = find_prev(tb, stack, GETKEY(tb, this->dbterm.tpl));
+	if (!((*doit)(tb, this, context, 0)))
+	    return;
+    } else {
+	next = find_prev(tb, stack, lastkey);
+    }
+
+    while ((this = next) != NULL) {
+	next = find_prev(tb, stack, GETKEY(tb, this->dbterm.tpl));
+	if (!((*doit)(tb, this, context, 0)))
+	    return;
+    }
+}
+
+/*
+ * Traverse the tree with a callback function, used by db_match_xxx
+ */
+static void traverse_forward(DbTableTree *tb,
+			     DbTreeStack* stack,
+			     Eterm lastkey,
+			     int (*doit)(DbTableTree *,
+					 TreeDbTerm *,
+					 void *,
+					 int),
+			     void *context) 
+{
+    TreeDbTerm *this, *next;
+
+    if (lastkey == NIL) {
+	stack->pos = stack->slot = 0;
+	if (( this = tb->root ) == NULL) {
+	    return;
+	}
+	while (this != NULL) {
+	    PUSH_NODE(stack, this);
+	    this = this->left;
+	}
+	this = TOP_NODE(stack);
+	next = find_next(tb, stack, GETKEY(tb, this->dbterm.tpl));
+	if (!((*doit)(tb, this, context, 1)))
+	    return;
+    } else {
+	next = find_next(tb, stack, lastkey);
+    }
+
+    while ((this = next) != NULL) {
+	next = find_next(tb, stack, GETKEY(tb, this->dbterm.tpl));
+	if (!((*doit)(tb, this, context, 1)))
+	    return;
+    }
+}
+
+/*
+ * Returns 0 if not given 1 if given and -1 on no possible match
+ * if key is given; *ret is set to point to the object concerned.
+ */
+static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm **ret, 
+		     Eterm *partly_bound)
+{
+    TreeDbTerm *this;
+    Eterm key;
+
+    ASSERT(ret != NULL);
+    if (pattern == am_Underscore || db_is_variable(pattern) != -1)
+	return 0;
+    key = db_getkey(tb->common.keypos, pattern);
+    if (is_non_value(key))
+	return -1;  /* can't possibly match anything */
+    if (!db_has_variable(key)) {   /* Bound key */
+	if (( this = find_node(tb, key) ) == NULL) {
+	    return -1;
+	}
+	*ret = this; 
+	return 1;
+    } else if (partly_bound != NULL && key != am_Underscore && 
+	       db_is_variable(key) < 0)
+	*partly_bound = key;
+	
+    return 0;
+}
+
+
+
+static Sint do_cmp_partly_bound(Eterm a, Eterm b, int *done)
+{
+    Eterm* aa;
+    Eterm* bb;
+    Eterm a_hdr;
+    Eterm b_hdr;
+    int i;
+    Sint j;
+
+    /* A variable matches anything */
+    if (is_atom(a) && (a == am_Underscore || (db_is_variable(a) >= 0))) {
+	*done = 1;
+	return 0;
+    }
+    if (a == b)
+	return 0;
+    
+    switch (a & _TAG_PRIMARY_MASK) {
+    case TAG_PRIMARY_LIST:
+	if (!is_list(b)) {
+	    return cmp(a,b);
+	}
+	aa = list_val(a);
+	bb = list_val(b);
+	while (1) {
+	    if ((j = do_cmp_partly_bound(*aa++, *bb++, done)) != 0 || *done) 
+		return j;
+	    if (*aa==*bb)
+		return 0;
+	    if (is_not_list(*aa) || is_not_list(*bb))
+		return do_cmp_partly_bound(*aa, *bb, done);
+	    aa = list_val(*aa);
+	    bb = list_val(*bb);
+	}
+    case TAG_PRIMARY_BOXED:
+	if ((b & _TAG_PRIMARY_MASK) != TAG_PRIMARY_BOXED) {
+	    return cmp(a,b);
+	}
+	a_hdr = ((*boxed_val(a)) & _TAG_HEADER_MASK) >> _TAG_PRIMARY_SIZE;
+	b_hdr = ((*boxed_val(b)) & _TAG_HEADER_MASK) >> _TAG_PRIMARY_SIZE;
+	if (a_hdr != b_hdr) {
+	    return cmp(a, b);
+	}
+	if (a_hdr == (_TAG_HEADER_ARITYVAL >> _TAG_PRIMARY_SIZE)) {
+	    aa = tuple_val(a);
+	    bb = tuple_val(b);
+	    /* compare the arities */
+	    i = arityval(*aa);	/* get the arity*/
+	    if (i < arityval(*bb)) return(-1);
+	    if (i > arityval(*bb)) return(1);
+	    while (i--) {
+		if ((j = do_cmp_partly_bound(*++aa, *++bb, done)) != 0 
+		    || *done) 
+		    return j;
+	    }
+	    return 0;
+	}
+	/* Drop through */
+      default:
+	  return cmp(a, b);
+    }
+}
+
+static Sint cmp_partly_bound(Eterm partly_bound_key, Eterm bound_key) 
+{
+    int done = 0;
+    Sint ret = do_cmp_partly_bound(partly_bound_key, bound_key, &done);
+#ifdef HARDDEBUG
+    erts_fprintf(stderr,"\ncmp_partly_bound: %T", partly_bound_key);
+    if (ret < 0)
+	erts_fprintf(stderr," < ");
+    else if (ret > 0)
+	erts_fprintf(stderr," > ");
+    else
+	erts_fprintf(stderr," == ");
+    erts_fprintf(stderr,"%T\n",bound_key);
+#endif
+    return ret;
+}
+
+/*
+** For partly_bound debugging....
+**
+BIF_RETTYPE ets_testnisse_2(BIF_ALIST_2)
+BIF_ADECL_2
+{
+    Eterm r1 = make_small(partly_bound_can_match_lesser(BIF_ARG_1,
+							BIF_ARG_2));
+    Eterm r2 = make_small(partly_bound_can_match_greater(BIF_ARG_1,
+							 BIF_ARG_2));
+    Eterm *hp = HAlloc(BIF_P,3);
+    Eterm ret;
+
+    ret = TUPLE2(hp,r1,r2);
+    BIF_RET(ret);
+}
+**
+*/
+static int partly_bound_can_match_lesser(Eterm partly_bound_1, 
+					 Eterm partly_bound_2) 
+{
+    int done = 0;
+    int ret = do_partly_bound_can_match_lesser(partly_bound_1, 
+					       partly_bound_2, 
+					       &done);
+#ifdef HARDDEBUG
+    erts_fprintf(stderr,"\npartly_bound_can_match_lesser: %T",partly_bound_1);
+    if (ret)
+	erts_fprintf(stderr," can match lesser than ");
+    else
+	erts_fprintf(stderr," can not match lesser than ");
+    erts_fprintf(stderr,"%T\n",partly_bound_2);
+#endif
+    return ret;
+}
+
+static int partly_bound_can_match_greater(Eterm partly_bound_1, 
+					  Eterm partly_bound_2) 
+{
+    int done = 0;
+    int ret = do_partly_bound_can_match_greater(partly_bound_1, 
+						partly_bound_2, 
+						&done);
+#ifdef HARDDEBUG
+    erts_fprintf(stderr,"\npartly_bound_can_match_greater: %T",partly_bound_1);
+    if (ret)
+	erts_fprintf(stderr," can match greater than ");
+    else
+	erts_fprintf(stderr," can not match greater than ");
+    erts_fprintf(stderr,"%T\n",partly_bound_2);
+#endif
+    return ret;
+}
+
+static int do_partly_bound_can_match_lesser(Eterm a, Eterm b, 
+					    int *done)
+{
+    Eterm* aa;
+    Eterm* bb;
+    Sint i;
+    int j;
+
+    if (is_atom(a) && (a == am_Underscore || 
+		       (db_is_variable(a) >= 0))) {
+	*done = 1;
+	if (is_atom(b) && (b == am_Underscore || 
+			   (db_is_variable(b) >= 0))) {
+	    return 0;
+	} else {
+	    return 1;
+	}
+    } else if (is_atom(b) && (b == am_Underscore || 
+			      (db_is_variable(b) >= 0))) {
+	*done = 1;
+	return 0;
+    }
+
+    if (a == b)
+	return 0;
+
+    if (not_eq_tags(a,b)) {
+	*done = 1;
+	return (cmp(a, b) < 0) ? 1 : 0;
+    }
+
+    /* we now know that tags are the same */
+    switch (tag_val_def(a)) {
+    case TUPLE_DEF:
+	aa = tuple_val(a);
+	bb = tuple_val(b);
+	/* compare the arities */
+	if (arityval(*aa) < arityval(*bb)) return 1;
+	if (arityval(*aa) > arityval(*bb)) return 0;
+	i = arityval(*aa);	/* get the arity*/
+	while (i--) {
+	    if ((j = do_partly_bound_can_match_lesser(*++aa, *++bb, 
+						      done)) != 0 
+		|| *done) 
+		return j;
+	}
+	return 0;
+    case LIST_DEF:
+	aa = list_val(a);
+	bb = list_val(b);
+	while (1) {
+	    if ((j = do_partly_bound_can_match_lesser(*aa++, *bb++, 
+						      done)) != 0 
+		|| *done) 
+		return j;
+	    if (*aa==*bb)
+		return 0;
+	    if (is_not_list(*aa) || is_not_list(*bb))
+		return do_partly_bound_can_match_lesser(*aa, *bb, 
+							done);
+	    aa = list_val(*aa);
+	    bb = list_val(*bb);
+	}
+    default:
+	if((i = cmp(a, b)) != 0) {
+	    *done = 1;
+	}
+	return (i < 0) ? 1 : 0;
+    }
+}
+
+static int do_partly_bound_can_match_greater(Eterm a, Eterm b, 
+					    int *done)
+{
+    Eterm* aa;
+    Eterm* bb;
+    Sint i;
+    int j;
+
+    if (is_atom(a) && (a == am_Underscore || 
+		       (db_is_variable(a) >= 0))) {
+	*done = 1;
+	if (is_atom(b) && (b == am_Underscore || 
+			   (db_is_variable(b) >= 0))) {
+	    return 0;
+	} else {
+	    return 1;
+	}
+    } else if (is_atom(b) && (b == am_Underscore || 
+			      (db_is_variable(b) >= 0))) {
+	*done = 1;
+	return 0;
+    }
+
+    if (a == b)
+	return 0;
+
+    if (not_eq_tags(a,b)) {
+	*done = 1;
+	return (cmp(a, b) > 0) ? 1 : 0;
+    }
+
+    /* we now know that tags are the same */
+    switch (tag_val_def(a)) {
+    case TUPLE_DEF:
+	aa = tuple_val(a);
+	bb = tuple_val(b);
+	/* compare the arities */
+	if (arityval(*aa) < arityval(*bb)) return 0;
+	if (arityval(*aa) > arityval(*bb)) return 1;
+	i = arityval(*aa);	/* get the arity*/
+	while (i--) {
+	    if ((j = do_partly_bound_can_match_greater(*++aa, *++bb, 
+						      done)) != 0 
+		|| *done) 
+		return j;
+	}
+	return 0;
+    case LIST_DEF:
+	aa = list_val(a);
+	bb = list_val(b);
+	while (1) {
+	    if ((j = do_partly_bound_can_match_greater(*aa++, *bb++, 
+						      done)) != 0 
+		|| *done) 
+		return j;
+	    if (*aa==*bb)
+		return 0;
+	    if (is_not_list(*aa) || is_not_list(*bb))
+		return do_partly_bound_can_match_greater(*aa, *bb, 
+							done);
+	    aa = list_val(*aa);
+	    bb = list_val(*bb);
+	}
+    default:
+	if((i = cmp(a, b)) != 0) {
+	    *done = 1;
+	}
+	return (i > 0) ? 1 : 0;
+    }
+}
+
+/*
+ * Callback functions for the different match functions
+ */
+
+static int doit_select(DbTableTree *tb, TreeDbTerm *this, void *ptr,
+		       int forward)
+{
+    struct select_context *sc = (struct select_context *) ptr;
+    Eterm ret;
+    Uint32 dummy;
+
+    sc->lastobj = this->dbterm.tpl;
+    
+    if (sc->end_condition != NIL && 
+	((forward && 
+	  cmp_partly_bound(sc->end_condition, 
+			   GETKEY_WITH_POS(sc->keypos, 
+					   this->dbterm.tpl)) < 0) ||
+	 (!forward && 
+	  cmp_partly_bound(sc->end_condition, 
+			   GETKEY_WITH_POS(sc->keypos, 
+					   this->dbterm.tpl)) > 0))) {
+	return 0;
+    }
+    ret = db_prog_match(sc->p, sc->mp,
+			make_tuple(this->dbterm.tpl), 
+			0, &dummy);
+    if (is_value(ret)) {
+	Uint sz;
+	Eterm *hp;
+	if (sc->all_objects) {
+	    hp = HAlloc(sc->p, this->dbterm.size + 2);
+	    ret = copy_shallow(DBTERM_BUF(&this->dbterm),
+				     this->dbterm.size,
+				     &hp,
+			             &MSO(sc->p));
+	} else {
+	    sz = size_object(ret);
+	    hp = HAlloc(sc->p, sz + 2);
+	    ret = copy_struct(ret, sz, 
+			      &hp, &MSO(sc->p));
+	}
+	sc->accum = CONS(hp, ret, sc->accum);
+    }
+    if (MBUF(sc->p)) {
+	/*
+	 * Force a trap and GC if a heap fragment was created. Many heap fragments
+	 * make the GC slow.
+	 */
+	sc->max = 0;
+    }
+    if (--(sc->max) <= 0) {
+	return 0;
+    }
+    return 1;
+}
+
+static int doit_select_count(DbTableTree *tb, TreeDbTerm *this, void *ptr,
+			     int forward)
+{
+    struct select_count_context *sc = (struct select_count_context *) ptr;
+    Eterm ret;
+    Uint32 dummy;
+
+    sc->lastobj = this->dbterm.tpl;
+    
+    /* Always backwards traversing */
+    if (sc->end_condition != NIL && 
+	(cmp_partly_bound(sc->end_condition, 
+			  GETKEY_WITH_POS(sc->keypos, 
+					  this->dbterm.tpl)) > 0)) {
+	return 0;
+    }
+    ret = db_prog_match(sc->p, sc->mp,
+			make_tuple(this->dbterm.tpl), 
+			0, &dummy);
+    if (ret == am_true) {
+	++(sc->got);
+    }
+    if (--(sc->max) <= 0) {
+	return 0;
+    }
+    return 1;
+}
+
+static int doit_select_chunk(DbTableTree *tb, TreeDbTerm *this, void *ptr,
+			     int forward)
+{
+    struct select_context *sc = (struct select_context *) ptr;
+    Eterm ret;
+    Uint32 dummy;
+
+    sc->lastobj = this->dbterm.tpl;
+    
+    if (sc->end_condition != NIL && 
+	((forward && 
+	  cmp_partly_bound(sc->end_condition, 
+			   GETKEY_WITH_POS(sc->keypos, 
+					   this->dbterm.tpl)) < 0) ||
+	 (!forward && 
+	  cmp_partly_bound(sc->end_condition, 
+			   GETKEY_WITH_POS(sc->keypos, 
+					   this->dbterm.tpl)) > 0))) {
+	return 0;
+    }
+
+    ret = db_prog_match(sc->p, sc->mp,
+			make_tuple(this->dbterm.tpl), 
+			0, &dummy);
+    if (is_value(ret)) {
+	Uint sz;
+	Eterm *hp;
+
+	++(sc->got);
+	if (sc->all_objects) {
+	    hp = HAlloc(sc->p, this->dbterm.size + 2);
+	    ret = copy_shallow(DBTERM_BUF(&this->dbterm),
+				     this->dbterm.size,
+				     &hp,
+			             &MSO(sc->p));
+	} else {
+	    sz = size_object(ret);
+	    hp = HAlloc(sc->p, sz + 2);
+	    ret = copy_struct(ret, sz, &hp, &MSO(sc->p));
+	}
+	sc->accum = CONS(hp, ret, sc->accum);
+    }
+    if (MBUF(sc->p)) {
+	/*
+	 * Force a trap and GC if a heap fragment was created. Many heap fragments
+	 * make the GC slow.
+	 */
+	sc->max = 0;
+    }
+    if (--(sc->max) <= 0 || sc->got == sc->chunk_size) {
+	return 0;
+    }
+    return 1;
+}
+
+
+static int doit_select_delete(DbTableTree *tb, TreeDbTerm *this, void *ptr,
+			      int forward)
+{
+    struct select_delete_context *sc = (struct select_delete_context *) ptr;
+    Eterm ret;
+    Uint32 dummy;
+    Eterm key;
+
+    if (sc->erase_lastterm)
+	free_term(tb, sc->lastterm);
+    sc->erase_lastterm = 0;
+    sc->lastterm = this;
+    
+    if (sc->end_condition != NIL && 
+	cmp_partly_bound(sc->end_condition, 
+			 GETKEY_WITH_POS(sc->keypos, 
+					 this->dbterm.tpl)) > 0)
+	return 0;
+    ret = db_prog_match(sc->p, sc->mp,
+			make_tuple(this->dbterm.tpl), 
+			0, &dummy);
+    if (ret == am_true) {
+	key = GETKEY(sc->tb, this->dbterm.tpl);
+	linkout_tree(sc->tb, key);
+	sc->erase_lastterm = 1;
+	++sc->accum;
+    }
+    if (--(sc->max) <= 0) {
+	return 0;
+    }
+    return 1;
+}
+
+#ifdef TREE_DEBUG
+static void do_dump_tree2(int to, void *to_arg, int show, TreeDbTerm *t,
+			  int offset)
+{
+    if (t == NULL)
+	return 0;
+    do_dump_tree2(to, to_arg, show, t->right, offset + 4);
+    if (show) {
+	erts_print(to, to_arg, "%*s%T (addr = %p, bal = %d)\n"
+		   offset, "", make_tuple(t->dbterm.tpl),
+		   t, t->balance);
+    }
+    do_dump_tree2(to, to_arg, show, t->left, offset + 4); 
+    return sum;
+}
+
+#endif
+
+#ifdef HARDDEBUG
+
+void db_check_table_tree(DbTable *tbl)
+{
+    DbTableTree *tb = &tbl->tree;
+    check_table_tree(tb->root);
+    check_saved_stack(tb);
+    check_slot_pos(tb);
+}
+
+static TreeDbTerm *traverse_until(TreeDbTerm *t, int *current, int to)
+{
+    TreeDbTerm *tmp;
+    if (t == NULL) 
+	return NULL;
+    tmp = traverse_until(t->left, current, to);
+    if (tmp != NULL)
+	return tmp;
+    ++(*current);
+    if (*current == to)
+	return t;
+    return traverse_until(t->right, current, to);
+}
+
+static void check_slot_pos(DbTableTree *tb)
+{
+    int pos = 0;
+    TreeDbTerm *t;
+    if (tb->stack.slot == 0 || tb->stack.pos == 0)
+	return;
+    t = traverse_until(tb->root, &pos, tb->stack.slot);
+    if (t != tb->stack.array[tb->stack.pos - 1]) {
+	erts_fprintf(stderr, "Slot position does not correspont with stack, "
+		   "element position %d is really 0x%08X, when stack says "
+		   "it's 0x%08X\n", tb->stack.slot, t, 
+		   tb->stack.array[tb->stack.pos - 1]);
+	do_dump_tree2(ERTS_PRINT_STDERR, NULL, 1, tb->root, 0);
+    }
+}
+	
+
+static void check_saved_stack(DbTableTree *tb)
+{
+     TreeDbTerm *t = tb->root;
+     DbTreeStack* stack = &tb->static_stack;
+     int n = 0;
+     if (stack->pos == 0)
+	 return;
+     if (t != stack->array[0]) {
+	 erts_fprintf(stderr,"tb->stack[0] is 0x%08X, should be 0x%08X\n",
+		      stack->array[0], t);
+	 do_dump_tree2(ERTS_PRINT_STDERR, NULL, 1, tb->root, 0);
+	 return;
+     }
+     while (n < stack->pos) {
+	 if (t == NULL) {
+	     erts_fprintf(stderr, "NULL pointer in tree when stack not empty,"
+			" stack depth is %d\n", n);
+	     do_dump_tree2(ERTS_PRINT_STDERR, NULL, 1, tb->root, 0);
+	     return;
+	 }
+	 n++;
+	 if (n < stack->pos) {
+	     if (stack->array[n] == t->left)
+		 t = t->left;
+	     else if (stack->array[n] == t->right)
+		 t = t->right;
+	     else {
+		 erts_fprintf(stderr, "tb->stack[%d] == 0x%08X does not "
+			    "represent child pointer in tree!"
+			    "(left == 0x%08X, right == 0x%08X\n", 
+			    n, tb->stack[n], t->left, t->right);
+		 do_dump_tree2(ERTS_PRINT_STDERR, NULL, 1, tb->root, 0);
+		 return;
+	     }
+	 }
+     }
+}
+
+static int check_table_tree(TreeDbTerm *t)
+{
+    int lh, rh;
+    if (t == NULL)
+	return 0;
+    lh = check_table_tree(t->left);
+    rh = check_table_tree(t->right);
+    if ((rh - lh) != t->balance) {
+	erts_fprintf(stderr, "Invalid tree balance for this node:\n");
+	erts_fprintf(stderr,"balance = %d, left = 0x%08X, right = 0x%08X\n"
+		     "data = %T",
+		     t->balance, t->left, t->right,
+		     make_tuple(t->dbterm.tpl));
+	erts_fprintf(stderr,"\nDump:\n---------------------------------\n");
+	do_dump_tree2(ERTS_PRINT_STDERR, NULL, 1, t, 0);
+	erts_fprintf(stderr,"\n---------------------------------\n");
+    }
+    return ((rh > lh) ? rh : lh) + 1;
+}
+	
+#endif