erl_db.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$
 */
/*
 * This file contains the bif interface functions and
 * the handling of the hash "meta table" ie the table of 
 * db 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"


erts_smp_atomic_t erts_tot_ets_memory_size;

/*
** Utility macros
*/

/* Get a key from any table structure and a tagged object */
#define TERM_GETKEY(tb, obj) db_getkey((tb)->common.keypos, (obj)) 

/* Utility macros for determining need of auto fixtable */

#define ONLY_WRITER(P,T) (((T)->common.status & DB_PRIVATE) || \
(((T)->common.status & DB_PROTECTED) && (T)->common.owner == (P)->id))
#define ONLY_READER(P,T) (((T)->common.status & DB_PRIVATE) && \
(T)->common.owner == (P)->id)
#define DID_TRAP(P,Ret) (!is_value(Ret) && ((P)->freason == TRAP))
#define SOLE_LOCKER(P,Fixations) ((Fixations) != NULL && \
(Fixations)->next == NULL && (Fixations)->pid == (P)->id && \
(Fixations)->counter == 1)

/* 
** The id in a tab_entry slot is
** DB_NOTUSED if it's never been used(*)
** DB_USED if it's been freed(*)
** An occupied slot has an (atom|small) id equal to the table's id
** This is so that we shall be able to terminate a search when we
** reach a point in the table that is impossible to reach if the id
** is there, we have to consider that tables can be removed thogh, so if
** we come to a removed slot, we must continue the search
** (*) A slot that has been used can actually be marked as DB_NOTUSED, 
** that happens if the next slot is also DB_NOTUSED when freeing.
** Then the slot we're freeing need not be marked as 
** DB_USED as you needn't continue search in that case.
**
**
*/

#define ISFREE(i)	((db_tables[i].id == DB_USED) || ISNOTUSED(i))
#define ISNOTUSED(i)	(db_tables[i].id == DB_NOTUSED)

/* 
** Globals 
*/

/* SMP fast spin lock for manipulating the db_tables entries */

static erts_smp_spinlock_t db_tables_lock;


/* This is a hashlist of all tables we have */

static struct tab_entry {
    DbTable *t;
    Uint id;              /* Automatically initialized */
    Eterm name;           /* An atom */
} *db_tables;  /* Local variable db_tables */

typedef enum { LCK_READ=1, LCK_WRITE=2 } db_lock_kind_t;

extern DbTableMethod db_hash;
extern DbTableMethod db_tree;

int user_requested_db_max_tabs;
int erts_ets_realloc_always_moves;
static int db_max_tabs;
static int last_slot;
static int no_tabs;		/* Number of active tables */
static DbTable *meta_pid_to_tab; /* Pid mapped to owned tables */
static DbTable *meta_pid_to_fixed_tab; /* Pid mapped to fixed tables */
static Eterm ms_delete_all;
static Eterm ms_delete_all_buff[8]; /* To compare with for deletion 
				       of all objects */

/* 
** Forward decls, static functions 
*/

static void fix_table_locked(Process* p, DbTable* tb);
static void unfix_table_locked(Process* p,  DbTable* tb);
static void free_fixations_locked(DbTable *tb);

static Eterm free_table_cont(Process *p, DbTable *tb, int first);
static void print_table(int to, void *to_arg, int show,  DbTable* tb);
static int next_prime(int n);
static BIF_RETTYPE ets_select_delete_1(Process *p, Eterm a1);
static BIF_RETTYPE ets_select_count_1(Process *p, Eterm a1);
static BIF_RETTYPE ets_select_trap_1(Process *p, Eterm a1);

/* 
 * Exported global
 */
Export ets_select_delete_continue_exp;
Export ets_select_count_continue_exp;
Export ets_select_continue_exp;

static ERTS_INLINE DbTable* db_ref(DbTable* tb)
{
    if (tb != NULL)
	erts_refc_inc(&tb->common.ref, 2);
    return tb;
}

static ERTS_INLINE DbTable* db_unref(DbTable* tb)
{
    if (!erts_refc_dectest(&tb->common.ref, 0)) {
#ifdef HARDDEBUG
	if (erts_smp_atomic_read(&tb->common.memory_size) != sizeof(DbTable)) {
	    erts_fprintf(stderr, "ets: db_unref memory remain=%ld fix=%x\n",
			 erts_smp_atomic_read(&tb->common.memory_size)-sizeof(DbTable), 
			 tb->common.fixations);
	}
	erts_fprintf(stderr, "ets: db_unref(%T) deleted!!!\r\n", 
		     tb->common.id);

	erts_fprintf(stderr, "ets: db_unref: meta_pid_to_tab common.memory_size = %ld\n",
		     erts_smp_atomic_read(&meta_pid_to_tab->common.memory_size));
	print_table(ERTS_PRINT_STDOUT, NULL, 1, meta_pid_to_tab);


	erts_fprintf(stderr, "ets: db_unref: meta_pid_to_fixed_tab common.memory_size = %ld\n",
		     erts_smp_atomic_read(&meta_pid_to_fixed_tab->common.memory_size));
	print_table(ERTS_PRINT_STDOUT, NULL, 1, meta_pid_to_fixed_tab);
	
#endif
#ifdef ERTS_SMP
	erts_smp_rwmtx_destroy(&tb->common.rwlock);
#endif
	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
	return NULL;
    }
    return tb;
}

static ERTS_INLINE void db_init_lock(DbTable* tb, char *name)
{
    erts_refc_init(&tb->common.ref, 1);
#ifdef ERTS_SMP
    erts_smp_rwmtx_init(&tb->common.rwlock, name);
#endif
}

static ERTS_INLINE void db_lock_take_over_ref(DbTable* tb, db_lock_kind_t kind)
{
#ifdef ERTS_SMP
    if (kind == LCK_WRITE)
	erts_smp_rwmtx_rwlock(&tb->common.rwlock);
    else
	erts_smp_rwmtx_rlock(&tb->common.rwlock);
#endif
}

static ERTS_INLINE void db_lock(DbTable* tb, db_lock_kind_t kind)
{
    (void) db_ref(tb);
#ifdef ERTS_SMP
    db_lock_take_over_ref(tb, kind);
#endif
}

static ERTS_INLINE void db_unlock(DbTable* tb, db_lock_kind_t kind)
{
#ifdef ERTS_SMP
    if (kind == LCK_WRITE)
	erts_smp_rwmtx_rwunlock(&tb->common.rwlock);
    else
	erts_smp_rwmtx_runlock(&tb->common.rwlock);
#endif
    (void) db_unref(tb); /* May delete table... */
}
    
static ERTS_INLINE
DbTable* db_get_table_aux(Process *p,
			  Eterm id,
			  int what,
			  db_lock_kind_t kind,
			  db_lock_kind_t (*get_kind)(DbTable *))
{
    Sint i, j;

    if (is_small(id))
	j = unsigned_val(id);
    else if (is_atom(id))
	j = atom_val(id);
    else
	return NULL;

    erts_smp_spin_lock(&db_tables_lock);

    i = j = j % db_max_tabs;
    while (1) {
	if (db_tables[i].id == id) {
	    DbTable* tb;
	    /* SMP: inc to prevent race, between unlock of db_tables_mtx
	     * and the locking outside the db_tables_mtx 
	     */
	    tb = db_ref(db_tables[i].t);
	    erts_smp_spin_unlock(&db_tables_lock);
#ifdef ERTS_SMP
	    if (get_kind)
		kind = (*get_kind)(tb);
#endif
	    db_lock_take_over_ref(tb, kind);
	    if (((tb->common.status & what) != 0) ||
		(p->id == tb->common.owner)) {
		return tb;
	    }
	    db_unlock(tb, kind);
	    return NULL;
	}
	if (ISNOTUSED(i++))
	    break;
	if (i == db_max_tabs) 
	    i = 0; 
	if (i == j)
	    break;
    }
    erts_smp_spin_unlock(&db_tables_lock);
    return NULL;
}

static DbTable* db_get_table(Process *p,
			     Eterm id,
			     int what,
			     db_lock_kind_t kind)
{
    return db_get_table_aux(p, id, what, kind, NULL);
}

static DbTable* db_get_table2(Process *p,
			      Eterm id,
			      int what,
			      db_lock_kind_t (*get_kind)(DbTable *))
{
    return db_get_table_aux(p, id, what, LCK_WRITE, get_kind);
}

/*
** Internal functions.(require db_tables_lock locked)
*/
static void meta_mark_free(int idx)
{
    int i;
    if (ISNOTUSED((idx + 1) % db_max_tabs)) {
	db_tables[idx].id = DB_NOTUSED;
	for (i = ((idx > 0) ? idx : db_max_tabs) - 1; 
	     i != idx && db_tables[i].id == DB_USED; 
	     i = ((i > 0) ? i : db_max_tabs) - 1) {
	    db_tables[i].id = DB_NOTUSED;
	}
    } else {
	db_tables[idx].id = DB_USED;
    }
}



/*
** BIF's
*/

/*
** Disables/enables rehashing for a table (if it is a hash table).
*/
BIF_RETTYPE ets_fixtable_2(BIF_ALIST_2)
{
    DbTable* tb;
    Eterm arg;

    /* This doesn't affect trees, but who cares... */

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }

    arg = BIF_ARG_2;

    if (BIF_ARG_2 == am_true) {
	fix_table_locked(BIF_P, tb);
    }
    else if (BIF_ARG_2 == am_false) {
	DbFixation *fix;

	for (fix = tb->common.fixations; fix != NULL; fix = fix->next) {
	    db_lock(meta_pid_to_fixed_tab, LCK_WRITE);
	    db_erase_bag_exact2(meta_pid_to_fixed_tab,
				fix->pid,
				make_small(tb->common.slot));
	    db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
	}
	while (tb->common.fixations != NULL) {
	    fix = tb->common.fixations;
	    tb->common.fixations = fix->next;
	    erts_db_free(ERTS_ALC_T_DB_FIXATION,
			 tb, (void *) fix, sizeof(DbFixation));
	}
	if (IS_HASH_TABLE(tb->common.status)) {
	    db_unfix_table_hash(&(tb->hash));
	}	
	tb->common.status &= ~DB_FIXED;
    }
    else {
	db_unlock(tb, LCK_WRITE);
	BIF_ERROR(BIF_P, BADARG);
    }
    db_unlock(tb, LCK_WRITE);
    BIF_RET(am_true);
}

BIF_RETTYPE ets_safe_fixtable_2(BIF_ALIST_2)
{
    DbTable *tb;

#ifdef HARDDEBUG
    erts_fprintf(stderr,
		"ets:safe_fixtable(%T,%T); Process: %T, initial: %T:%T/%bpu\n",
		BIF_ARG_1, BIF_ARG_2, BIF_P->id,
		BIF_P->initial[0], BIF_P->initial[1], BIF_P->initial[2]);
#endif
    /* SMP fixme (should be a write lock) */
    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }

    if (BIF_ARG_2 == am_true) {
	fix_table_locked(BIF_P, tb);
    }
    else if (BIF_ARG_2 == am_false) {
	if (tb->common.status & DB_FIXED) {
	    unfix_table_locked(BIF_P, tb);
	}
    }
    else {
	db_unlock(tb, LCK_WRITE);
	BIF_ERROR(BIF_P, BADARG);
    }
    db_unlock(tb, LCK_WRITE);
    BIF_RET(am_true);
}

#ifdef ERTS_SMP
#define STEP_LOCK_TYPE(T) \
  (IS_TREE_TABLE((T)->common.type) ? LCK_WRITE : LCK_READ)
#else
#define STEP_LOCK_TYPE(T) LCK_WRITE
#endif

static db_lock_kind_t
step_lock_type(DbTable *tb)
{
    return STEP_LOCK_TYPE(tb);
}

/* 
** Returns the first Key in a table 
*/
BIF_RETTYPE ets_first_1(BIF_ALIST_1)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    tb = db_get_table2(BIF_P, BIF_ARG_1, DB_READ, step_lock_type);

    if (!tb) {
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_first(BIF_P, tb, &ret);

    db_unlock(tb, STEP_LOCK_TYPE(tb));

    if (cret != DB_ERROR_NONE) {
	BIF_ERROR(BIF_P, BADARG);
    }
    BIF_RET(ret);
}

/* 
** The next BIF, given a key, return the "next" key 
*/
BIF_RETTYPE ets_next_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    tb = db_get_table2(BIF_P, BIF_ARG_1, DB_READ, step_lock_type);

    if (!tb) {
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_next(BIF_P, tb, BIF_ARG_2, &ret);

    db_unlock(tb, STEP_LOCK_TYPE(tb));

    if (cret != DB_ERROR_NONE) {
	BIF_ERROR(BIF_P, BADARG);
    }
    BIF_RET(ret);
}

/* 
** Returns the last Key in a table 
*/
BIF_RETTYPE ets_last_1(BIF_ALIST_1)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    tb = db_get_table2(BIF_P, BIF_ARG_1, DB_READ, step_lock_type);

    if (!tb) {
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_last(BIF_P, tb, &ret);

    db_unlock(tb, STEP_LOCK_TYPE(tb));

    if (cret != DB_ERROR_NONE) {
	BIF_ERROR(BIF_P, BADARG);
    }
    BIF_RET(ret);
}

/* 
** The prev BIF, given a key, return the "previous" key 
*/
BIF_RETTYPE ets_prev_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    tb = db_get_table2(BIF_P, BIF_ARG_1, DB_READ, step_lock_type);

    if (!tb) {
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_prev(BIF_P,tb,BIF_ARG_2,&ret);

    db_unlock(tb, STEP_LOCK_TYPE(tb));

    if (cret != DB_ERROR_NONE) {
	BIF_ERROR(BIF_P, BADARG);
    }
    BIF_RET(ret);
}

/* 
** update_counter(Tab, Key, Increment) 
** Returns new value (integer)
*/

BIF_RETTYPE ets_update_counter_3(BIF_ALIST_3)
{
    DbTable* tb;
    Eterm ret;
    int cret;
    Eterm increment = BIF_ARG_3;
    Sint position = 0;
    Eterm threshold = NIL;
    Eterm warp_to = NIL;

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);