erl_db.c

OTP是开放电信平台的简称

    }
    if (!(tb->common.status & (DB_SET | DB_ORDERED_SET))) { /*TT*/
	goto badarg;
    }
    if (is_tuple(BIF_ARG_3)) { /* key position specified */
	Eterm *tpl = tuple_val(BIF_ARG_3);
	switch (arityval(*tpl)) {
	case 4: /* threshold specified */
	    if (!(is_small(tpl[3]) || is_big(tpl[3])) ||
		!(is_small(tpl[4]) || is_big(tpl[4]))) {
		goto badarg;
	    }
	    threshold = tpl[3];
	    warp_to = tpl[4];
	    /* Fall through */
	case 2:
	    if (!is_small(tpl[1]) ||
		!(is_small(tpl[2]) || is_big(tpl[2]))) {
		goto badarg;
	    }
	    position = signed_val(tpl[1]);
	    increment = tpl[2];
	    if (position == tb->common.keypos) {
		goto badarg;
	    }
	    break;
	default:
	    goto badarg;
	}
    }
	
    cret = tb->common.meth->db_update_counter(BIF_P,tb,
					      BIF_ARG_2, increment, 0, 
					      position, &ret);

    if (cret == DB_ERROR_NONE &&
	threshold != NIL) { /* Maybe warp it */
	if ((cmp(increment,make_small(0)) < 0) ? /* negative increment? */
	    (cmp(ret,threshold) < 0) :  /* if negative, check if below */
	    (cmp(ret,threshold) > 0)) { /* else check if above threshold */

	    cret = tb->common.meth->db_update_counter(BIF_P,tb,
						      BIF_ARG_2, warp_to, 1, 
						      position, &ret);
	}
    }

    db_unlock(tb, LCK_WRITE);

    switch (cret) {
    case DB_ERROR_NONE:
	/* Check for threshold */
	BIF_RET(ret);
    case DB_ERROR_SYSRES:
	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
    default:
	BIF_ERROR(BIF_P, BADARG);
	break;
    }
 badarg:
    db_unlock(tb, LCK_WRITE);
    BIF_ERROR(BIF_P, BADARG);
}

/* 
** The put BIF 
*/
BIF_RETTYPE ets_insert_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret = DB_ERROR_NONE;
    Eterm ret = am_true;
    Eterm lst;
    DbTableMethod* meth;

    CHECK_TABLES();

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (BIF_ARG_2 == NIL) {
	db_unlock(tb, LCK_WRITE);
	BIF_RET(am_true);
    }
    meth = tb->common.meth;
    if (is_list(BIF_ARG_2)) {
	for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
	    if (is_not_tuple(CAR(list_val(lst))) || 
		(arityval(*tuple_val(CAR(list_val(lst)))) < tb->common.keypos)) {
		goto badarg;
	    }
	}
	if (lst != NIL) {
	    goto badarg;
	}
	for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
	    cret = meth->db_put(BIF_P, tb, CAR(list_val(lst)), &ret);
	    if (cret != DB_ERROR_NONE)
		break;
	}
    } else {
	if (is_not_tuple(BIF_ARG_2) || 
	    (arityval(*tuple_val(BIF_ARG_2)) < tb->common.keypos)) {
	    goto badarg;
	}
	cret = meth->db_put(BIF_P, tb, BIF_ARG_2, &ret);
    }

    db_unlock(tb, LCK_WRITE);
    
    switch (cret) {
    case DB_ERROR_NONE:
	BIF_RET(ret);
    case DB_ERROR_SYSRES:
	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
    default:
	BIF_ERROR(BIF_P, BADARG);
    }
 badarg:
    db_unlock(tb, LCK_WRITE);
    BIF_ERROR(BIF_P, BADARG);    
}


/* 
** The put-if-not-already-there BIF... 
*/
BIF_RETTYPE ets_insert_new_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret = DB_ERROR_NONE;
    Eterm ret = am_true;
    Eterm lst;
    Eterm lookup_ret;
    DbTableMethod* meth;

    CHECK_TABLES();

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (BIF_ARG_2 == NIL) {
	db_unlock(tb, LCK_WRITE);
	BIF_RET(am_true);
    }
    meth = tb->common.meth;
    if (is_list(BIF_ARG_2)) {
	for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
	    if (is_not_tuple(CAR(list_val(lst))) || 
		(arityval(*tuple_val(CAR(list_val(lst)))) < tb->common.keypos)) {
		goto badarg;
	    }
	}
	if (lst != NIL) {
	    goto badarg;
	}

	for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
	    cret = meth->db_member(BIF_P, tb,
				   TERM_GETKEY(tb,CAR(list_val(lst))),
				   &lookup_ret);
	    if ((cret != DB_ERROR_NONE) || (lookup_ret != am_false)) {
		ret = am_false;
		goto done;
	    }
	}

	for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
	    cret = meth->db_put(BIF_P, tb,CAR(list_val(lst)),&ret);
	    if (cret != DB_ERROR_NONE)
		break;
	}
    } else {
	if (is_not_tuple(BIF_ARG_2) || 
	    (arityval(*tuple_val(BIF_ARG_2)) < tb->common.keypos)) {
	    goto badarg;
	}
	cret = meth->db_member(BIF_P, tb,TERM_GETKEY(tb,BIF_ARG_2),
			       &lookup_ret);
	if ((cret != DB_ERROR_NONE) || (lookup_ret != am_false)) 
	    ret = am_false;
	else {
	    cret = meth->db_put(BIF_P,tb,BIF_ARG_2, &ret);
	}
    }

done:
    db_unlock(tb, LCK_WRITE);
    switch (cret) {
    case DB_ERROR_NONE:
	BIF_RET(ret);
    case DB_ERROR_SYSRES:
	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
    default:
	BIF_ERROR(BIF_P, BADARG);
    }
 badarg:
    db_unlock(tb, LCK_WRITE);
    BIF_ERROR(BIF_P, BADARG);    
}

/*
** Rename a (possibly) named table
*/

BIF_RETTYPE ets_rename_2(BIF_ALIST_2)
{
    DbTable* tb;
    int oldslot, newslot;
    Eterm dummy;
    Eterm meta_tuple[3];
    DbFixation *fix;
    Eterm ret;

#ifdef HARDDEBUG
    erts_fprintf(stderr,
		"ets:rename(%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

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

    if (is_not_atom(BIF_ARG_2)) {
	goto badarg;
    }

    if (is_not_atom(tb->common.id)) { /* Not a named table */
	tb->common.the_name = BIF_ARG_2;
	goto done;
    }

    oldslot = tb->common.slot;

    /* Ok, a named table, find a new slot for it */
    erts_smp_spin_lock(&db_tables_lock);
    newslot = atom_val(BIF_ARG_2) % db_max_tabs;
    while (1) {
	if (ISFREE(newslot))
	    break;
	if (db_tables[newslot].id == BIF_ARG_2) {
	    erts_smp_spin_unlock(&db_tables_lock);
	    goto badarg;
	}
	if (++newslot == db_max_tabs)
	    newslot=0; 
    }
    db_tables[newslot].id = BIF_ARG_2;
    db_tables[newslot].t = tb;
    tb->common.id = tb->common.the_name = BIF_ARG_2;
    tb->common.slot = newslot;
    meta_mark_free(oldslot);
    db_tables[oldslot].t = NULL;

    erts_smp_spin_unlock(&db_tables_lock);

    db_lock(meta_pid_to_tab, LCK_WRITE);
    if (db_put_hash(NULL,meta_pid_to_tab,
		    TUPLE2(meta_tuple, 
			   tb->common.owner, 
			   make_small(newslot)),
		    &dummy)
	!= DB_ERROR_NONE) {
	erl_exit(1,"Could not insert ets metadata in rename.");
    }

    db_erase_bag_exact2(meta_pid_to_tab,
			tb->common.owner,make_small(oldslot));
    db_unlock(meta_pid_to_tab, LCK_WRITE);

    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(oldslot));
	db_unlock(meta_pid_to_fixed_tab, LCK_WRITE);
	if (db_put_hash(NULL, meta_pid_to_fixed_tab,
			TUPLE2(meta_tuple, 
			       fix->pid, 
			       make_small(newslot)),
			&dummy)
	    != DB_ERROR_NONE) {
	    erl_exit(1,"Could not insert ets metadata"
		     " in rename.");
	}		
    }
 done:
    ret = tb->common.id;
    db_unlock(tb, LCK_WRITE);
    BIF_RET(ret);
 badarg:
    db_unlock(tb, LCK_WRITE);
    BIF_ERROR(BIF_P, BADARG);    
}


/* 
** The create table BIF     
** Args: (Name, Properties) 
*/

BIF_RETTYPE ets_new_2(BIF_ALIST_2)
{
    DbTable* tb = NULL;
    int slot;
    Eterm list;
    Eterm val;
    Eterm ret;
    Uint32 status;
    Sint keypos;
    int is_named;
    int cret;
    Eterm dummy;
    Eterm meta_tuple[3];
    DbTableMethod* meth;


    if (is_not_atom(BIF_ARG_1)) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (is_not_nil(BIF_ARG_2) && is_not_list(BIF_ARG_2)) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (no_tabs >  (db_max_tabs * 4) / 5) {
	erts_send_error_to_logger_str(BIF_P->group_leader,
				      "** Too many db tables **\n");
	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
    }

    status = DB_NORMAL | DB_SET | DB_LHASH | DB_PROTECTED;
    keypos = 1;
    is_named = 0;

    list = BIF_ARG_2;
    while(is_list(list)) {
	val = CAR(list_val(list));
	if (val == am_bag) {
	    status |= DB_BAG;
	    status &= ~(DB_SET | DB_DUPLICATE_BAG | DB_ORDERED_SET);
	}
	else if (val == am_duplicate_bag) {
	    status |= DB_DUPLICATE_BAG;
	    status &= ~(DB_SET | DB_BAG | DB_ORDERED_SET);
	}
	else if (val == am_ordered_set) {
	    status |= DB_ORDERED_SET;
	    status &= ~(DB_SET | DB_BAG | DB_DUPLICATE_BAG);
	}
	/*TT*/
	else if (is_tuple(val)) {
	    Eterm *tp = tuple_val(val);

	    if ((arityval(tp[0]) == 2) && (tp[1] == am_keypos) &&
		is_small(tp[2]) && (signed_val(tp[2]) > 0)) {
		keypos = signed_val(tp[2]);
	    }
	    else {
		BIF_ERROR(BIF_P, BADARG);
	    }
	}
	else if (val == am_public) {
	    status |= DB_PUBLIC;
	    status &= ~DB_PROTECTED;
	}
	else if (val == am_private) {
	    status |= DB_PRIVATE;
	    status &= ~DB_PROTECTED;
	}
	else if (val == am_named_table) {
	    is_named = 1;
	}
	else if (val == am_set || val == am_protected)
	    ;
	else {
	    BIF_ERROR(BIF_P, BADARG);
	}
	list = CDR(list_val(list));
    }
    if (is_not_nil(list)) /* it must be a well formed list */
	BIF_ERROR(BIF_P, BADARG);

    /* we create table outside the global table lock 
     * and take the unusal cost of destroy table if it
     * fails to find a slot 
     */
    {
        DbTable init_tb;

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

    if (IS_HASH_TABLE(status))
	meth = &db_hash;
    else if (IS_TREE_TABLE(status))
	meth = &db_tree;
    else
	meth = NULL;

    tb->common.meth = meth;

    db_init_lock(tb, "db_tab");

    tb->common.the_name = BIF_ARG_1;
    tb->common.status = status;
#ifdef ERTS_SMP
    tb->common.type = status & ERTS_ETS_TABLE_TYPES;
    /* Note, 'type' is *read only* from now on... */
#endif
    tb->common.keypos = keypos;
    tb->common.owner = BIF_P->id;

    tb->common.nitems = 0;
    tb->common.kept_items = 0;

    tb->common.fixations = NULL;

    if (meth == NULL) 
	cret = DB_ERROR_UNSPEC;
    else
	cret = meth->db_create(BIF_P, tb);
    if (cret != DB_ERROR_NONE) {
	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
	BIF_ERROR(BIF_P, BADARG);
    }

    erts_smp_spin_lock(&db_tables_lock);

    /* allocate the slot for the table */
    if (++last_slot == db_max_tabs) 
	last_slot = 0;
    if (is_named) {
	slot = atom_val(BIF_ARG_1) % db_max_tabs;
	while (1) {
	    if (ISFREE(slot)) {
		ret = BIF_ARG_1;
		break;
	    }
	    if (db_tables[slot].id == BIF_ARG_1) {
		goto badarg;
	    }
	    if (++slot == db_max_tabs) {
		slot=0; 
	    }
	}
    }
    else {  /* Allocate number slot */
	slot = last_slot;
	while(1) {
	    if (ISFREE(slot)) {
		ret = make_small(slot);
		break;
	    }
	    if (++slot == db_max_tabs) {
		slot=0; 
	    }
	}
    }
    tb->common.id = ret;
    tb->common.slot = slot;           /* store slot for erase */

    db_tables[slot].id = ret;  /* Insert the table */
    db_tables[slot].t = tb;
    
    BIF_P->flags |= F_USING_DB; /* So we can remove tb if p dies */

    no_tabs++;

    erts_smp_spin_unlock(&db_tables_lock);

#ifdef HARDDEBUG
    erts_fprintf(stderr,
		"ets:new(%T,%T)=%T; Process: %T, initial: %T:%T/%bpu\n",
		 BIF_ARG_1, BIF_ARG_2, ret, BIF_P->id,
		 BIF_P->initial[0], BIF_P->initial[1], BIF_P->initial[2]);
	erts_fprintf(stderr, "ets: new: meta_pid_to_tab common.memory_size = %ld\n",
		     erts_smp_atomic_read(&meta_pid_to_tab->common.memory_size));
	erts_fprintf(stderr, "ets: new: meta_pid_to_fixed_tab common.memory_size = %ld\n",
		     erts_smp_atomic_read(&meta_pid_to_fixed_tab->common.memory_size));
#endif

    db_lock(meta_pid_to_tab, LCK_WRITE);
    if (db_put_hash(NULL, meta_pid_to_tab,
		    TUPLE2(meta_tuple, BIF_P->id, 
			   make_small(slot)),&dummy) != DB_ERROR_NONE) {
	erl_exit(1,"Could not update ets metadata.");
    }
    db_unlock(meta_pid_to_tab, LCK_WRITE);


    BIF_RET(ret);

/* bad args white hold lock */
 badarg:
    erts_smp_spin_unlock(&db_tables_lock);
    if (tb != NULL) {
	tb->common.meth->db_free_table(tb);
	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
    }
    BIF_ERROR(BIF_P, BADARG);
}

/* 
** The lookup BIF 
*/
BIF_RETTYPE ets_lookup_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);