erl_db.c

OTP是开放电信平台的简称

    }
    if (lst != NIL) {
	goto error;
    }
    BIF_RET2(ret,i);
}


/*
** External interface (NOT BIF's)
*/


/* Init the db */

void init_db(void)
{
    DbTable init_tb;
    int i;
    int max_ets;
    extern Eterm* em_apply_bif;
    Eterm *hp;

    erts_smp_spinlock_init(&db_tables_lock, "db_tables");

    erts_smp_atomic_init(&erts_tot_ets_memory_size, 0);
    last_slot = 0;
    db_initialize_util();
    if (( max_ets = (user_requested_db_max_tabs*5)/4 ) < DB_DEF_MAX_TABS)
	db_max_tabs = DB_DEF_MAX_TABS;
    else
	db_max_tabs = next_prime(max_ets);

    db_tables = erts_db_alloc_nt(ERTS_ALC_T_DB_TABLES,
				 sizeof(struct tab_entry)*db_max_tabs);

    no_tabs = 0;
    for (i=0; i<db_max_tabs; i++) {
	db_tables[i].id = DB_NOTUSED;
	db_tables[i].t = NULL;
    }

    db_initialize_hash();
    db_initialize_tree();

    /*TT*/
    /* Create meta table invertion. */
    erts_smp_atomic_init(&init_tb.common.memory_size, 0);
    meta_pid_to_tab = (DbTable*) erts_db_alloc(ERTS_ALC_T_DB_TABLE,
					       &init_tb,
					       sizeof(DbTable));
    erts_smp_atomic_init(&meta_pid_to_tab->common.memory_size,
			 erts_smp_atomic_read(&init_tb.common.memory_size));

    meta_pid_to_tab->common.id = NIL;
    meta_pid_to_tab->common.the_name = am_true;
    meta_pid_to_tab->common.status = (DB_NORMAL | DB_BAG | DB_LHASH | 
				      DB_PUBLIC);
#ifdef ERTS_SMP
    meta_pid_to_tab->common.type
	= meta_pid_to_tab->common.status & ERTS_ETS_TABLE_TYPES;
    /* Note, 'type' is *read only* from now on... */
#endif
    meta_pid_to_tab->common.keypos = 1;
    meta_pid_to_tab->common.owner  = NIL;
    meta_pid_to_tab->common.nitems = 0;
    meta_pid_to_tab->common.slot   = -1;
    meta_pid_to_tab->common.meth   = &db_hash;

    db_init_lock(meta_pid_to_tab, "meta_pid_to_tab");

    if (db_create_hash(NULL, meta_pid_to_tab) != DB_ERROR_NONE) {
	erl_exit(1,"Unable to create ets metadata tables.");
    }

    erts_smp_atomic_set(&init_tb.common.memory_size, 0);
    meta_pid_to_fixed_tab = (DbTable*) erts_db_alloc(ERTS_ALC_T_DB_TABLE,
						     &init_tb,
						     sizeof(DbTable));
    erts_smp_atomic_init(&meta_pid_to_fixed_tab->common.memory_size,
			 erts_smp_atomic_read(&init_tb.common.memory_size));

    meta_pid_to_fixed_tab->common.id = NIL;
    meta_pid_to_fixed_tab->common.the_name = am_true;
    meta_pid_to_fixed_tab->common.status = (DB_NORMAL | DB_BAG | DB_LHASH | 
					    DB_PUBLIC);
#ifdef ERTS_SMP
    meta_pid_to_fixed_tab->common.type
	= meta_pid_to_fixed_tab->common.status & ERTS_ETS_TABLE_TYPES;
    /* Note, 'type' is *read only* from now on... */
#endif
    meta_pid_to_fixed_tab->common.keypos = 1;
    meta_pid_to_fixed_tab->common.owner  = NIL;
    meta_pid_to_fixed_tab->common.nitems = 0;
    meta_pid_to_fixed_tab->common.slot   = -1;
    meta_pid_to_fixed_tab->common.meth   = &db_hash;

    db_init_lock(meta_pid_to_fixed_tab, "meta_pid_to_fixed_tab");

    if (db_create_hash(NULL, meta_pid_to_fixed_tab) != DB_ERROR_NONE) {
	erl_exit(1,"Unable to create ets metadata tables.");
    }

    /* Non visual BIF to trap to. */
    memset(&ets_select_delete_continue_exp, 0, sizeof(Export));
    ets_select_delete_continue_exp.address = 
	&ets_select_delete_continue_exp.code[3];
    ets_select_delete_continue_exp.code[0] = am_ets;
    ets_select_delete_continue_exp.code[1] = am_atom_put("delete_trap",11);
    ets_select_delete_continue_exp.code[2] = 1;
    ets_select_delete_continue_exp.code[3] =
	(Eterm) em_apply_bif;
    ets_select_delete_continue_exp.code[4] = 
	(Eterm) &ets_select_delete_1;
    /* Non visual BIF to trap to. */
    memset(&ets_select_count_continue_exp, 0, sizeof(Export));
    ets_select_count_continue_exp.address = 
	&ets_select_count_continue_exp.code[3];
    ets_select_count_continue_exp.code[0] = am_ets;
    ets_select_count_continue_exp.code[1] = am_atom_put("count_trap",11);
    ets_select_count_continue_exp.code[2] = 1;
    ets_select_count_continue_exp.code[3] =
	(Eterm) em_apply_bif;
    ets_select_count_continue_exp.code[4] = 
	(Eterm) &ets_select_count_1;
    /* Non visual BIF to trap to. */
    memset(&ets_select_continue_exp, 0, sizeof(Export));
    ets_select_continue_exp.address = 
	&ets_select_continue_exp.code[3];
    ets_select_continue_exp.code[0] = am_ets;
    ets_select_continue_exp.code[1] = am_atom_put("select_trap",11);
    ets_select_continue_exp.code[2] = 1;
    ets_select_continue_exp.code[3] =
	(Eterm) em_apply_bif;
    ets_select_continue_exp.code[4] = 
	(Eterm) &ets_select_trap_1;
    hp = ms_delete_all_buff;
    ms_delete_all = CONS(hp, am_true, NIL);
    hp += 2;
    ms_delete_all = TUPLE3(hp,am_Underscore,NIL,ms_delete_all);
    hp +=4;
    ms_delete_all = CONS(hp, ms_delete_all,NIL);
}

/* Called when  a process which has created any tables dies */
/* So that we can remove the tables ceated by the process   */

#define ARRAY_CHUNK 100
void db_proc_dead(Eterm pid)
{
    Eterm arr[ARRAY_CHUNK];
    int arr_siz;
    int ret;
    Eterm dummy;
    DbFixation *fix,**pp;
    int i;

#ifdef HARDDEBUG
    erts_fprintf(stderr, "db_proc_dead(); Process: %T\n", pid);
#endif

    for (;;) {
	arr_siz = ARRAY_CHUNK;
	db_lock(meta_pid_to_tab, LCK_READ);
	if ((ret = db_get_element_array(meta_pid_to_tab,
					pid, 2, arr, &arr_siz)) == 
	    DB_ERROR_BADKEY) {
	    db_unlock(meta_pid_to_tab, LCK_READ);
	    /* done */
	    break;
	} else if (ret != DB_ERROR_NONE) {
	    erl_exit(1,"Inconsistent ets metadata");
	}
	db_unlock(meta_pid_to_tab, LCK_READ);

	/* first delete mark all tables */
	for (i = 0; i < arr_siz; ++i) {
	    DbTable* tb;
	    Sint ix = unsigned_val(arr[i]); /* slot */

	    erts_smp_spin_lock(&db_tables_lock);
	    if ((tb = db_ref(db_tables[ix].t)) != NULL) {
		meta_mark_free(ix);
		db_tables[ix].t = NULL;
		no_tabs--;
	    }
	    erts_smp_spin_unlock(&db_tables_lock);
	    if (tb != NULL) {
#ifdef HARDDEBUG
		erts_fprintf(stderr, "db_proc_dead(); Table: %T,  Process: %T\n", tb->common.id, pid);
#endif

		db_lock_take_over_ref(tb, LCK_WRITE);
		/* Clear all access bits. */
		tb->common.status &= ~(DB_PROTECTED|DB_PUBLIC|DB_PRIVATE);
		free_fixations_locked(tb);
		tb->common.meth->db_free_table(tb);
		db_unlock(tb, LCK_WRITE);
		db_unref(tb); /* this one MAY delete the table */
	    }
	    if (arr_siz == ARRAY_CHUNK) {
		/* Need to erase each explicitly */
		db_lock(meta_pid_to_tab, LCK_WRITE);
		db_erase_bag_exact2(meta_pid_to_tab,pid,arr[i]);
		db_unlock(meta_pid_to_tab, LCK_WRITE);
	    }
	}
	if (arr_siz < ARRAY_CHUNK) {
	    db_lock(meta_pid_to_tab, LCK_WRITE);
	    db_erase_hash(NULL,meta_pid_to_tab,pid,&dummy);
	    db_unlock(meta_pid_to_tab, LCK_WRITE);
	}
    }

    /* And now for the fixations... */
    for (;;) {
	arr_siz = ARRAY_CHUNK;
	db_lock(meta_pid_to_fixed_tab, LCK_READ);
	ret = db_get_element_array(meta_pid_to_fixed_tab, 
				   pid, 2, arr, &arr_siz);
	db_unlock(meta_pid_to_fixed_tab, LCK_READ);

	if (ret == DB_ERROR_BADKEY) {
	    /* done */
	    return;
	} else if (ret != DB_ERROR_NONE) {
	    erl_exit(1,"Inconsistent ets metadata");
	}

	for (i = 0; i < arr_siz; ++i) {
	    Sint ix = unsigned_val(arr[i]); /* slot */
	    DbTable* tb;

	    erts_smp_spin_lock(&db_tables_lock);
	    tb = db_ref(db_tables[ix].t);
	    erts_smp_spin_unlock(&db_tables_lock);
	    if (tb == NULL) {
		/* Was owner */
		continue;
	    }
	    db_lock_take_over_ref(tb, LCK_WRITE);

	    for (pp = &(tb->common.fixations); *pp != NULL;
		 pp = &((*pp)->next)) {
		if ((*pp)->pid == pid) {
		    fix = *pp;
		    *pp = (*pp)->next;
		    if (arr_siz == ARRAY_CHUNK) {
			db_lock(meta_pid_to_fixed_tab, LCK_WRITE);
			db_erase_bag_exact2(meta_pid_to_fixed_tab,
					    pid,
					    make_small(tb->common.slot));
			db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
		    }
		    erts_db_free(ERTS_ALC_T_DB_FIXATION,
				 tb, (void *) fix, sizeof(DbFixation));
		    break;
		}
	    }
	    if (tb->common.fixations == NULL) {
		if (IS_HASH_TABLE(tb->common.status)) {
		    db_unfix_table_hash(&(tb->hash));
		}
		tb->common.status &= ~DB_FIXED;
	    }
	    db_unlock(tb, LCK_WRITE);
	}
	if (arr_siz < ARRAY_CHUNK) {
	    db_lock(meta_pid_to_fixed_tab, LCK_WRITE);
	    db_erase_hash(NULL,meta_pid_to_fixed_tab,pid,&dummy);
	    db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
	} 
    }
}


static int next_prime(int n)
{
    int i;
    if (n % 2 == 0) /* No even numbers... */
	++n;
    for (;;) {
	for (i = 3; (i*i) <= n; i+=2) {
	    if (n % i == 0)
		break;
	}
	if ((i*i) > n)
	    return n;
	n += 2;
    }
}


/*  SMP note: table must be WRITE locked */
static void fix_table_locked(Process* p, DbTable* tb)
{
    DbFixation *fix;
    Eterm dummy;
    Eterm meta_tuple[3];

    if (!(tb->common.status & DB_FIXED)) { 
	tb->common.status |= DB_FIXED;
	get_now(&(tb->common.megasec),
		&(tb->common.sec), 
		&(tb->common.microsec));
    }
    for (fix = tb->common.fixations; fix != NULL; fix = fix->next) {
	if (fix->pid == p->id) {
	    ++(fix->counter);
	    break;
	}
    }
    if (fix == NULL) {
	fix = (DbFixation *) erts_db_alloc(ERTS_ALC_T_DB_FIXATION,
					   tb, sizeof(DbFixation));
	fix->pid = p->id;
	fix->counter = 1;
	fix->next = tb->common.fixations;
	tb->common.fixations = fix;
	/* SMP: I guess we need some kind of lock here ? */
	p->flags |= F_USING_DB;        
	db_lock(meta_pid_to_fixed_tab, LCK_WRITE);
	if (db_put_hash(NULL,meta_pid_to_fixed_tab,
			TUPLE2(meta_tuple, 
			       p->id, 
			       make_small(tb->common.slot)),
			&dummy)
	    != DB_ERROR_NONE) {
	    erl_exit(1,"Could not insert ets metadata"
		     " in safe_fixtable.");
	}	
	db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
    }
}

/*  SMP note: table must be WRITE locked */
static void unfix_table_locked(Process* p,  DbTable* tb)
{
    DbFixation **pp;
    DbFixation *fix;
    
    for (pp = &(tb->common.fixations); *pp != NULL; pp = &((*pp)->next)) {
	if ((*pp)->pid == p->id) {
	    --((*pp)->counter);
	    ASSERT((*pp)->counter >= 0);
	    if ((*pp)->counter == 0) {
		fix = *pp;
		*pp = (*pp)->next;
		db_lock(meta_pid_to_fixed_tab, LCK_WRITE);
		db_erase_bag_exact2(meta_pid_to_fixed_tab,
				    p->id,
				    make_small(tb->common.slot));
		db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
		erts_db_free(ERTS_ALC_T_DB_FIXATION,
			     tb, (void *) fix, sizeof(DbFixation));
	    }
	    break;
	}
    }
    if (tb->common.fixations == NULL) {
	if (IS_HASH_TABLE(tb->common.status)) {
	    db_unfix_table_hash(&(tb->hash));
	}
	tb->common.status &= ~DB_FIXED;
    }
}

/* Assume that tb is locked (write) */
static void free_fixations_locked(DbTable *tb)
{
    DbFixation *fix;
    DbFixation *next_fix;

    fix = tb->common.fixations;
    while (fix != NULL) {
	next_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);
	erts_db_free(ERTS_ALC_T_DB_FIXATION,
		     tb, (void *) fix, sizeof(DbFixation));
	fix = next_fix;
    }
    tb->common.fixations = NULL;
}


static BIF_RETTYPE free_table_cont(Process* p, DbTable *tb, int first)
{
    Eterm result;

    if (!first) {
#ifdef HARDDEBUG
	erts_fprintf(stderr,"ets: free_table_cont %T (continue)\r\n",
		     tb->common.id);
#endif
	db_lock(tb, LCK_WRITE);
    }

    result = tb->common.meth->db_free_table_continue(tb, first);

    if (result == 0) {
#ifdef HARDDEBUG
	erts_fprintf(stderr,"ets: free_table_cont %T (continue begin)\r\n",
		     tb->common.id);
#endif
	db_unlock(tb, LCK_WRITE);
	/* More work to be done. Let other processes work and call us again. */
	/* SMP FIXME: can we really pass tb here??? */
	BUMP_ALL_REDS(p);
	BIF_TRAP1(bif_export[BIF_ets_db_delete_1], p, (Eterm) tb);
    } else {
	int nitems;
#ifdef HARDDEBUG
	erts_fprintf(stderr,"ets: free_table_cont %T (continue end)\r\n",
		     tb->common.id);
#endif
	/* Completely done - we will not get called again. */
	erts_smp_spin_lock(&db_tables_lock);
	meta_mark_free(tb->common.slot);
	db_tables[tb->common.slot].t = NULL;
	no_tabs--;
	erts_smp_spin_unlock(&db_tables_lock);

	db_lock(meta_pid_to_tab, LCK_WRITE);
	db_erase_bag_exact2(meta_pid_to_tab,tb->common.owner,
			    make_small(tb->common.slot));
	db_unlock(meta_pid_to_tab, LCK_WRITE);
	nitems = tb->common.nitems;
	db_unlock(tb, LCK_WRITE);
	db_unref(tb);
	BUMP_REDS(p, nitems/15);
	return am_true;
    }
}

static void print_table(int to, void *to_arg, int show,  DbTable* tb)
{
    erts_print(to, to_arg, "Table: %T\n", tb->common.id);
    erts_print(to, to_arg, "Name: %T\n", tb->common.the_name);

    tb->common.meth->db_print(to, to_arg, show, tb);

    erts_print(to, to_arg, "Objects: %d\n", tb->common.nitems);
    erts_print(to, to_arg, "Words: %bpu\n",
	       (Uint) ((erts_smp_atomic_read(&tb->common.memory_size)
			+ sizeof(Uint)
			- 1)
		       / sizeof(Uint)));
}

void db_info(int to, void *to_arg, int show)    /* Called by break handler */
{
    int i;
    for (i=0; i < db_max_tabs; i++) 
	if (!ISFREE(i)) {
	    erts_print(to, to_arg, "=ets:%T\n", db_tables[i].t->common.owner);
	    erts_print(to, to_arg, "Slot: %d\n", i);
	    print_table(to, to_arg, show, db_tables[i].t);
	}
#ifdef DEBUG
    erts_print(to, to_arg, "=internal_ets: Process to table index\n");
    print_table(to, to_arg, show, meta_pid_to_tab);
    erts_print(to, to_arg, "=internal_ets: Process to fixation index\n");
    print_table(to, to_arg, show, meta_pid_to_fixed_tab);
#endif
}

Uint
erts_ets_memory_size(void)
{
    return (Uint) erts_smp_atomic_read(&erts_tot_ets_memory_size);
}

/* SMP Note: May only be used when system is locked */
void
erts_db_foreach_table(void (*func)(DbTable *, void *), void *arg)
{
    int i, j;
    j = 0;
    for(i = 0; (i < db_max_tabs && j < no_tabs); i++) {
	if (!ISFREE(i)) {
	    j++;
	    (*func)(db_tables[i].t, arg);
	}
    }
    ASSERT(j == no_tabs);
}

/* SMP Note: May only be used when system is locked */
void
erts_db_foreach_offheap(DbTable *tb,
			void (*func)(ErlOffHeap *, void *),
			void *arg)
{
    tb->common.meth->db_foreach_offheap(tb, func, arg);
}

#ifdef HARDDEBUG   /* Here comes some debug functions */

void db_check_tables(void)
{
#ifdef ERTS_SMP
    return;
#else
    int i;

    for (i = 0; i < db_max_tabs; i++) {
	if (!ISFREE(i)) {
	    DbTable* tb = db_tables[i].t; 
	    tb->common.meth->db_check_table(tb);
	}
    }
#endif
}

#endif /* HARDDEBUG */