erl_db_hash.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 unordered ETS tables.
** The tables are implemented as linear dynamic hash tables.
*/

/*
#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 "export.h"

#include "erl_db_hash.h"


/* 
 * The following symbols can be manipulated to "tune" the linear hash array 
 */
#define BASIC_SIZE 64               /* #words for empty array       */
#define CHAIN_LEN 6                 /* Medium bucket chain len      */


#define SZEXP   8
#define SEGSZ   (1 << SZEXP)
#define SZMASK  ((1 << SZEXP)-1)


#define SEG_LEN         256   /* When growing init segs */
#define SEG_INCREAMENT  128   /* Number of segments to grow */

#define BUCKET(tb, i) (tb)->seg[(i) >> SZEXP][(i) & SZMASK]

/*
 * When deleting a table, the number of records to delete.
 * Approximate number, because we must delete entire buckets.
 */
#define DELETE_RECORD_LIMIT 10000

/* ix is a NAME parameter :-) */
#define HASH(tb, hval, ix) \
  do { \
     if ((ix = ((hval) & (tb)->szm)) < (tb)->p) \
        ix = (hval) & (((tb)->szm << 1) | 1); \
  } while(0)

#define MAX_HASH 0xEFFFFFFFUL
#define INVALID_HASH 0xFFFFFFFFUL

/* optimised version of make_hash (normal case? atomic key) */
#define MAKE_HASH(term) \
    ((is_atom(term) ? (atom_tab(atom_val(term))->slot.bucket.hvalue) : \
      make_hash2(term)) % MAX_HASH)

/* 
 * tplp is an untagged pointer to a tuple we know is large enough 
 * and dth is a pointer to a DbTableHash.   
 */
#define GETKEY(dth, tplp)   (*((tplp) +  (dth)->common.keypos))

/*
 * Some special binary flags
 */
#define BIN_FLAG_ALL_OBJECTS         BIN_FLAG_USR1

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

/*
 * Local types 
 */
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 key_given;
    HashDbTerm **dlists[10];     /* default buffer for list of "pre found"
				  * buckets */
    HashDbTerm ***lists;         /* List if poters to list pointers for the 
				  * buckets to search if keys are given, 
				  * = dlists initially */
    unsigned num_lists;         /* Number of elements in "lists",
				 * = 0 initially */
    Binary *mp;                 /* The compiled match program */
};



/*
** Forward decl's (static functions)
*/
static HashDbTerm** alloc_seg(DbTableHash *tb);
static int realloc_counter(DbTableCommon *tb, HashDbTerm** bp, Uint sz, 
			   Eterm new_counter, int counterpos);
static HashDbTerm* next(DbTableHash *tb, Uint *iptr, HashDbTerm *list);
static HashDbTerm* search_list(DbTableHash* tb, Eterm key, 
			       HashValue hval, HashDbTerm *list);
static void shrink(DbTableHash* tb);
static void grow(DbTableHash* tb);
static void free_term(DbTableHash *tb, HashDbTerm* p);
static Eterm put_term_list(Process* p, HashDbTerm* ptr1, HashDbTerm* ptr2);
static HashDbTerm* get_term(DbTableHash* tb, HashDbTerm* old, 
			    Eterm obj, HashValue hval);
static int analyze_pattern(DbTableHash *tb, Eterm pattern, 
			   struct mp_info *mpi);
/*
 *  Method interface functions
 */
static int db_first_hash(Process *p, 
			 DbTable *tbl, 
			 Eterm *ret);

static int db_next_hash(Process *p, 
			DbTable *tbl, 
			Eterm key,
			Eterm *ret);

static int db_update_counter_hash(Process *p, 
				  DbTable *tbl, 
				  Eterm key,
				  Eterm incr,
				  int warp,
				  int counterpos,
				  Eterm *ret);


static int db_member_hash(Process *p, DbTable *tbl, 
			  Eterm key, Eterm *ret);

static int db_get_element_hash(Process *p, DbTable *tbl, 
			       Eterm key, int ndex, Eterm *ret);

static int db_erase_object_hash(Process *p, DbTable *tbl, 
				Eterm object,Eterm *ret);

static int db_slot_hash(Process *p, DbTable *tbl, 
			Eterm slot_term, Eterm *ret);

static int db_select_chunk_hash(Process *p, DbTable *tbl, 
				Eterm pattern, Sint chunk_size,
				int reverse, Eterm *ret);
static int db_select_hash(Process *p, DbTable *tbl, 
			  Eterm pattern, int reverse, Eterm *ret);
static int db_select_count_hash(Process *p, DbTable *tbl, 
				Eterm pattern, Eterm *ret);
static int db_select_delete_hash(Process *p, DbTable *tbl, 
				 Eterm pattern, Eterm *ret);

static int db_select_continue_hash(Process *p, DbTable *tbl, 
				   Eterm continuation, Eterm *ret);

static int db_select_count_continue_hash(Process *p, DbTable *tbl, 
					 Eterm continuation, Eterm *ret);

static int db_select_delete_continue_hash(Process *p, DbTable *tbl,
					  Eterm continuation, Eterm *ret);
static void db_print_hash(int to,
			  void *to_arg,
			  int show,
			  DbTable *tbl);
static int db_free_table_hash(DbTable *tbl);

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


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

static int db_delete_all_objects_hash(Process* p, DbTable* tbl);
#ifdef HARDDEBUG
static void db_check_table_hash(DbTableHash *tb);
#endif


/*
** Static variables
*/

/*
** External interface 
*/
DbTableMethod db_hash =
{
    db_create_hash,
    db_first_hash,
    db_next_hash,
    db_first_hash,   /* last == first  */
    db_next_hash,    /* prev == next   */
    db_put_hash,
    db_get_hash,
    db_get_element_hash,
    db_member_hash,
    db_erase_hash,
    db_erase_object_hash,
    db_slot_hash,
    db_update_counter_hash,
    db_select_chunk_hash,
    db_select_hash,
    db_select_delete_hash,
    db_select_continue_hash, /* hmm continue_hash? */
    db_select_delete_continue_hash,
    db_select_count_hash,
    db_select_count_continue_hash,
    db_delete_all_objects_hash,
    db_free_table_hash,
    db_free_table_continue_hash,
    db_print_hash,
    db_foreach_offheap_hash,
#ifdef HARDDEBUG
    db_check_table_hash,
#else
    NULL,
#endif

};

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

void db_unfix_table_hash(DbTableHash *tb)
{
    while (tb->fixdel != NULL) {
	FixedDeletion *fx = tb->fixdel;
	int ix = fx->slot;
	HashDbTerm **bp = &BUCKET(tb, ix);
	HashDbTerm *b = *bp;

	tb->fixdel = fx->next;
	erts_db_free(ERTS_ALC_T_DB_FIX_DEL,
		     (DbTable *) tb,
		     (void *) fx,
		     sizeof(FixedDeletion));

	while (b != NULL) {
	    if (b->hvalue == INVALID_HASH) {
		*bp = b->next;
		free_term(tb, b);
		b = *bp;
	    } else {
		bp = &b->next;
		b = b->next;
	    }
	}
    }
    tb->common.kept_items = 0;
}

int db_create_hash(Process *p, DbTable *tbl)
{
    DbTableHash *tb = &tbl->hash;
    tb->szm = SZMASK;
    tb->nslots = SEGSZ;
    tb->nactive = SEGSZ;
    tb->p = 0;
    tb->nsegs = 1;
    tb->seg = (HashDbTerm***) erts_db_alloc(ERTS_ALC_T_DB_SEG_TAB,
					    (DbTable *) tb,
					    sizeof(HashDbTerm**));
    tb->seg[0] = alloc_seg(tb);
    tb->fixdel = NULL;

    return DB_ERROR_NONE;
}

static int db_first_hash(Process *p, DbTable *tbl, Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    int i = 0;
    while (i < tb->nactive) {
	HashDbTerm* list = BUCKET(tb, i);
	if (list != 0) {
	    Eterm key = GETKEY(tb, list->dbterm.tpl);

	    COPY_OBJECT(key, p, ret);
	    return DB_ERROR_NONE;
	}
	i++;
    }
    *ret = am_EOT;
    return DB_ERROR_NONE;
}

static int db_next_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    HashValue hval;
    Uint ix;
    HashDbTerm* b1;

    hval = MAKE_HASH(key);
    HASH(tb, hval, ix);
    b1 = BUCKET(tb, ix);

    while(b1 != 0) {
	if (((b1->hvalue == hval) || b1->hvalue == INVALID_HASH) 
	    && EQ(key, GETKEY(tb, b1->dbterm.tpl))) {
	    HashDbTerm* b2 = next(tb, &ix, b1);
	    if ((tb->common.status & DB_BAG) || 
		(tb->common.status & DB_DUPLICATE_BAG)) {
		while (b2 != 0) {
		    Eterm key2 = GETKEY(tb, b2->dbterm.tpl);
		    if (EQ(key, key2)) {
			b2 = next(tb, &ix, b2);
			continue;
		    }
		    break;
		}
	    }
	    if (b2 == 0) {
		*ret = am_EOT;
		return DB_ERROR_NONE;
	    }
	    else {
		COPY_OBJECT(GETKEY(tb, b2->dbterm.tpl), p, ret);
		return DB_ERROR_NONE;
	    }
	}
	b1 = b1->next;
    }
    return DB_ERROR_BADKEY;
}    

static int db_update_counter_hash(Process *p, DbTable *tbl,
				  Eterm key,
				  Eterm incr, 
				  int warp,
				  int counterpos,
				  Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    HashDbTerm* b;
    HashDbTerm** bp;
    int ix;
    HashValue hval;

    hval = MAKE_HASH(key);
    HASH(tb, hval, ix);
    bp = &BUCKET(tb, ix);
    b = *bp;

    while (b != 0) {
	if ((b->hvalue == hval) && EQ(key,GETKEY(tb, b->dbterm.tpl)))
	    break;
	bp = &b->next;
	b = *bp;
   }

    if (b == 0) 
	return DB_ERROR_BADKEY;

    if (counterpos <= 0)
	counterpos = tb->common.keypos + 1;

    return db_do_update_counter(p, (DbTableCommon *) tb,
				(void *) bp, b->dbterm.tpl,
				counterpos, 
				(int (*)(DbTableCommon *,
					 void *,
					 Uint,
					 Eterm,
					 int))
				&realloc_counter, incr, warp, ret);
}

int db_put_hash(Process *proc, DbTable *tbl, Eterm obj, Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    HashValue hval;
    int ix;
    Eterm key;
    HashDbTerm** bp;
    HashDbTerm* b;
    HashDbTerm* q;

    key = GETKEY(tb, tuple_val(obj));
    hval = MAKE_HASH(key);
    HASH(tb, hval, ix);
    bp = &BUCKET(tb, ix);
    b = *bp;

    while(b != 0) {
	if (((b->hvalue == hval) || b->hvalue == INVALID_HASH)
	    && EQ(key, GETKEY(tb, b->dbterm.tpl))) {
	    if (tb->common.status & DB_SET) {
		HashDbTerm* bnext = b->next;
		if (b->hvalue == INVALID_HASH) {
		    tb->common.nitems++;
		}
		q = get_term(tb, b, obj, hval);
		q->next = bnext;
		q->hvalue = hval; /* In case of INVALID_HASH */
		*bp = q;
		*ret = am_true;
		return DB_ERROR_NONE;
	    }
	    else if (tb->common.status & DB_BAG) {
		HashDbTerm** tp = bp;
                HashDbTerm* p = b;
		
                if (eq(make_tuple(b->dbterm.tpl), obj)) {
		    if (b->hvalue == INVALID_HASH) {
			tb->common.nitems++;
		    }
		    b->hvalue = hval;
		    *ret = am_true;
		    return DB_ERROR_NONE;
                }
                bp = &b->next;
                b = b->next;
                while ((b != 0) && 
		       ((b->hvalue == hval) || b->hvalue == INVALID_HASH) && 
                       EQ(key, GETKEY(tb, b->dbterm.tpl))) {
                    if (eq(make_tuple(b->dbterm.tpl), obj)) {
			if (b->hvalue == INVALID_HASH) {
			    tb->common.nitems++;
			}
			b->hvalue = hval;
			*ret = am_true;
			return DB_ERROR_NONE;
                    }
                    bp = &b->next;
                    b = b->next;
                }

                q = get_term(tb, NULL, obj, hval);
                q->next = p;
                *tp = q;
		goto Lupdate;
	    }
	    else {  /* if (tb->status & DB_DUPLICATE_BAG) */
		q = get_term(tb, NULL, obj, hval);
		q->next = b;
		*bp = q;
		goto Lupdate;
	    }
	}
	bp = &b->next;
	b = b->next;
    }

    q = get_term(tb, NULL, obj, hval);
    q->next = b;
    *bp = q;

 Lupdate:
    tb->common.nitems++;

    if ( ((tb->common.nitems / tb->nactive) > CHAIN_LEN) && 
	((tb->common.status & DB_FIXED) == 0))
	grow(tb);
    CHECK_TABLES();
    *ret = am_true;
    return DB_ERROR_NONE;
}

int db_get_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    HashValue hval;
    int ix;
    HashDbTerm* b1;

    hval = MAKE_HASH(key);
    HASH(tb, hval, ix);
    b1 = BUCKET(tb, ix);

    while(b1 != 0) {
	if ((b1->hvalue == hval) && EQ(key, GETKEY(tb, b1->dbterm.tpl))) {
	    HashDbTerm* b2 = b1->next;
	    Eterm copy;

	    if ((tb->common.status & DB_BAG) || 
		(tb->common.status & DB_DUPLICATE_BAG)) {
		while((b2 != 0) && ((b2->hvalue == hval) || 
				    (b2->hvalue == INVALID_HASH)) &&
		      EQ(key, GETKEY(tb, b2->dbterm.tpl)))
		    b2 = b2->next;
	    }
	    copy = put_term_list(p, b1, b2);
	    CHECK_TABLES();
	    *ret = copy;
	    return DB_ERROR_NONE;
	}
	b1 = b1->next;
    }
    *ret = NIL;
    return DB_ERROR_NONE;
}

int db_get_element_array(DbTable *tbl, 
			 Eterm key,
			 int ndex, 
			 Eterm *ret,
			 int *num_ret)
{
    DbTableHash *tb = &tbl->hash;
    HashValue hval;
    int ix;
    HashDbTerm* b1;
    int num = 0;
    
    hval = MAKE_HASH(key);
    HASH(tb, hval, ix);
    b1 = BUCKET(tb, ix);

    while(b1 != 0) {
	if ((b1->hvalue == hval) && EQ(key, GETKEY(tb, b1->dbterm.tpl))) {
	    if ((tb->common.status & DB_BAG) || 
		(tb->common.status & DB_DUPLICATE_BAG)) {
		HashDbTerm* b;
		HashDbTerm* b2 = b1->next;

		while((b2 != 0) && ((b2->hvalue == hval) || 
				    (b2->hvalue == INVALID_HASH)) &&
		      EQ(key, GETKEY(tb, b2->dbterm.tpl))) {
		    if (ndex > arityval(b2->dbterm.tpl[0]))
			return DB_ERROR_BADITEM;
		    b2 = b2->next;
		}

		b = b1;
		while(b != b2) {
		    if (num < *num_ret) {
			ret[num++] = b->dbterm.tpl[ndex];
		    } else {
			return DB_ERROR_NONE;
		    }
		    b = b->next;
		}
		*num_ret = num;
		return DB_ERROR_NONE;
	    }
	    else {
		ASSERT(*num_ret > 0);
		ret[0] = b1->dbterm.tpl[ndex];
		*num_ret = 1;
		return DB_ERROR_NONE;
	    }
	}
	b1 = b1->next;
    }
    return DB_ERROR_BADKEY;
}
    
    
static int db_member_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
{
    DbTableHash *tb = &tbl->hash;
    HashValue hval;
    int ix;
    HashDbTerm* b1;