erl_db.c

OTP是开放电信平台的简称

    }

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

    db_unlock(tb, LCK_READ);

    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);
    }

}

/* 
** The lookup BIF 
*/
BIF_RETTYPE ets_member_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);
    }

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

    db_unlock(tb, LCK_READ);

    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);
    }

}

/* 
** Get an element from a term
** get_element_3(Tab, Key, Index)
** return the element or a list of elements if bag
*/
BIF_RETTYPE ets_lookup_element_3(BIF_ALIST_3)
{
    DbTable* tb;
    Sint index;
    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);
    }

    if (is_not_small(BIF_ARG_3) || ((index = signed_val(BIF_ARG_3)) < 1)) {
	db_unlock(tb, LCK_READ);
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_get_element(BIF_P, tb, 
					   BIF_ARG_2, index, &ret);
    db_unlock(tb, LCK_READ);
    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);
    }
}

/* 
** BIF to erase a whole table and release all memory it holds 
*/
BIF_RETTYPE ets_db_delete_1(BIF_ALIST_1)
{
    DbTable* tb;

    if (is_CP(BIF_ARG_1)) {
#ifdef HARDDEBUG
	erts_fprintf(stderr,
		     "ets:delete(%T); TRAP Process: %T, initial: %T:%T/%bpu\n",
		     ((DbTable*)BIF_ARG_1)->common.id, BIF_P->id,
		     BIF_P->initial[0], BIF_P->initial[1], BIF_P->initial[2]);
#endif
	/*
	 * We have been called through a trap.
	 */
	return free_table_cont(BIF_P, (DbTable *) BIF_ARG_1, 0);
    }

#ifdef HARDDEBUG
    erts_fprintf(stderr,
		"ets:delete(%T); Process: %T, initial: %T:%T/%bpu\n",
		BIF_ARG_1, BIF_P->id,
		BIF_P->initial[0], BIF_P->initial[1], BIF_P->initial[2]);
#endif

    CHECK_TABLES();

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    
    /*
     * Clear all access bits to prevent any ets operation to access the
     * table while it is being deleted.
     */
    tb->common.status &= ~(DB_PROTECTED|DB_PUBLIC|DB_PRIVATE);
    
    if (tb->common.owner != BIF_P->id) {
	Eterm dummy;
	Eterm meta_tuple[3];

	/*
	 * The process is being deleted by a process other than its owner.
	 * To make sure that the table will be completely deleted if the
	 * current process will be killed (e.g. by an EXIT signal), we will
	 * now transfer the ownership to the current process.
	 */
	db_lock(meta_pid_to_tab, LCK_WRITE);
	db_erase_bag_exact2(meta_pid_to_tab, tb->common.owner,
			    make_small(tb->common.slot));

	BIF_P->flags |= F_USING_DB;
	tb->common.owner = BIF_P->id;

	db_put_hash(NULL,
		    meta_pid_to_tab,
		    TUPLE2(meta_tuple,BIF_P->id,make_small(tb->common.slot)),
		    &dummy);
	db_unlock(meta_pid_to_tab, LCK_WRITE);
    }

    free_fixations_locked(tb);
    return free_table_cont(BIF_P, tb, 1);
}

/* 
** BIF to erase a whole table and release all memory it holds 
*/
BIF_RETTYPE ets_delete_all_objects_1(BIF_ALIST_1)
{
    DbTable* tb;

    CHECK_TABLES();

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

    tb->common.meth->db_delete_all_objects(BIF_P, tb);

    db_unlock(tb, LCK_WRITE);

    BIF_RET(am_true);
}

/* 
** Erase an object with given key, or maybe several objects if we have a bag  
** Called as db_erase(Tab, Key), where Key is element 1 of the
** object(s) we want to erase                                  
*/
BIF_RETTYPE ets_delete_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

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

    cret = tb->common.meth->db_erase(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);
    }
}

/* 
** Erase a specific object, or maybe several objects if we have a bag  
*/
BIF_RETTYPE ets_delete_object_2(BIF_ALIST_2)
{
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if (is_not_tuple(BIF_ARG_2) || 
	(arityval(*tuple_val(BIF_ARG_2)) < tb->common.keypos)) {
	db_unlock(tb, LCK_WRITE);
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_erase_object(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);
    }
}

/*
** This is for trapping, cannot be called directly.
*/
static BIF_RETTYPE ets_select_delete_1(Process *p, Eterm a1)
{
    BIF_RETTYPE result;
    DbTable* tb;
    int cret;
    Eterm ret;
    Eterm *tptr;
    
    CHECK_TABLES();
#ifdef DEBUG
    /*
     * Make sure that the table exists.
     */
    if (!is_tuple(a1)) {
	/* "Cannot" happen, this is not a real BIF, its trapped
	   to under controlled conditions */
	erl_exit(1,"Internal error in ets:select_delete");
    }
#endif
    tptr = tuple_val(a1);
#ifdef DEBUG
    if (arityval(*tptr) < 1) {
	erl_exit(1,"Internal error in ets:select_delete");
    }
#endif
    
    if ((tb = db_get_table(p, tptr[1], DB_WRITE, LCK_WRITE)) == NULL) {
	/* This should be OK, the emulator handles errors in BIF's that aren't
	   exported nowdays... */
	BIF_ERROR(p,BADARG);
    }

    cret = tb->common.meth->db_select_delete_continue(p,tb,a1,&ret);

    db_unlock(tb, LCK_WRITE);
    /* SMP fixme table may be deleted !!! */

    switch (cret) {
    case DB_ERROR_NONE:
	if(IS_HASH_TABLE(tb->common.status) && !DID_TRAP(p,ret) &&
	   !ONLY_READER(p,tb)) {
	    ets_safe_fixtable_2(p, tb->common.id, am_false);
	} 
	ERTS_BIF_PREP_RET(result, ret);
	break;
    default:
	if(IS_HASH_TABLE(tb->common.status) && !ONLY_READER(p,tb)) {
	    ets_safe_fixtable_2(p, tb->common.id, am_false);
	} 
	ERTS_BIF_PREP_ERROR(result, p, BADARG);
	break;
    }

    erts_match_set_release_result(p);

    return result;
}
    

BIF_RETTYPE ets_select_delete_2(BIF_ALIST_2)
{
    BIF_RETTYPE result;
    DbTable* tb;
    int cret;
    Eterm ret;

    CHECK_TABLES();
    /*
     * Make sure that the table exists.
     */

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    if(eq(BIF_ARG_2, ms_delete_all)){
	int nitems = tb->common.nitems;
	tb->common.meth->db_delete_all_objects(BIF_P, tb);
	db_unlock(tb, LCK_WRITE);
	BIF_RET(erts_make_integer(nitems,BIF_P));
    }
    cret = tb->common.meth->db_select_delete(BIF_P, tb, BIF_ARG_2, &ret);

    switch (cret) {
    case DB_ERROR_NONE:
	if(IS_HASH_TABLE(tb->common.status) && DID_TRAP(BIF_P,ret) && 
	   !ONLY_READER(BIF_P,tb)) {
	    /* We will trap and as we're here this call wasn't a trap... */
	    fix_table_locked(BIF_P, tb);
	}
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_RET(result, ret);
	break;
    case DB_ERROR_SYSRES:
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_ERROR(result, BIF_P, SYSTEM_LIMIT);
	break;
    default:
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_ERROR(result, BIF_P, BADARG);
	break;
    }

    erts_match_set_release_result(BIF_P);

    return result;
}

/* 
** Return a list of tables on this node 
*/
BIF_RETTYPE ets_all_0(BIF_ALIST_0)
{
    DbTable* tb;
    Eterm previous;
    int i, j;
    Eterm* hp;
    int t_no_tabs;
    int t_db_max_tabs;

    erts_smp_spin_lock(&db_tables_lock);
    t_no_tabs = no_tabs;
    t_db_max_tabs = db_max_tabs;
    erts_smp_spin_unlock(&db_tables_lock);

    hp = HAlloc(BIF_P, 2*t_no_tabs);

    erts_smp_spin_lock(&db_tables_lock);
    previous = NIL;
    j = 0;
    for(i = 0; (i < t_db_max_tabs && j < t_no_tabs); i++) {
	if (!ISFREE(i)) {
	    j++;
	    tb = db_tables[i].t;
	    previous = CONS(hp, tb->common.id, previous);
	    hp += 2;
	}
    }
    ASSERT(j == t_no_tabs);
    erts_smp_spin_unlock(&db_tables_lock);
    BIF_RET(previous);
}


/*
** db_slot(Db, Slot) -> [Items].
*/
BIF_RETTYPE ets_slot_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);
    }
    /* The slot number is checked in table specific code. */
    cret = tb->common.meth->db_slot(BIF_P, tb, BIF_ARG_2, &ret);
    db_unlock(tb, LCK_READ);
    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);
    }
}

/* 
** The match BIF,  called as ets:match(Table, Pattern), ets:match(Continuation) or ets:match(Table,Pattern,ChunkSize).
*/

BIF_RETTYPE ets_match_1(BIF_ALIST_1)
{
    return ets_select_1(BIF_P, BIF_ARG_1);
}

BIF_RETTYPE ets_match_2(BIF_ALIST_2)
{
    Eterm ms;
    Eterm buff[8];
    Eterm *hp = buff;
    /*hp = HAlloc(BIF_P, 8);*/
    ms = CONS(hp, am_DollarDollar, NIL);
    hp += 2;
    ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
    hp += 4;
    ms = CONS(hp, ms, NIL);
    return ets_select_2(BIF_P, BIF_ARG_1, ms);
}

BIF_RETTYPE ets_match_3(BIF_ALIST_3)
{
    Eterm ms;
    Eterm buff[8];
    Eterm *hp = buff;
    /*hp = HAlloc(BIF_P, 8);*/
    ms = CONS(hp, am_DollarDollar, NIL);
    hp += 2;
    ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
    hp += 4;
    ms = CONS(hp, ms, NIL);
    return ets_select_3(BIF_P, BIF_ARG_1, ms, BIF_ARG_3);
}


BIF_RETTYPE ets_select_3(BIF_ALIST_3)
{
    BIF_RETTYPE result;
    DbTable* tb;
    int cret;
    Eterm ret;
    Sint chunk_size;

    CHECK_TABLES();
    /*
     * Make sure that the table exists.
     */

    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_WRITE)) == NULL) {
	BIF_ERROR(BIF_P, BADARG);
    }
    
    /* Chunk size strictly greater than 0 */
    if (is_not_small(BIF_ARG_3) || (chunk_size = signed_val(BIF_ARG_3)) <= 0) {
	db_unlock(tb, LCK_WRITE);
	BIF_ERROR(BIF_P, BADARG);
    }

    cret = tb->common.meth->db_select_chunk(BIF_P, tb,
					    BIF_ARG_2, chunk_size, 
					    0 /* not reversed */, &ret);
    switch (cret) {
    case DB_ERROR_NONE:
	if(IS_HASH_TABLE(tb->common.status) && DID_TRAP(BIF_P,ret) &&
	   !ONLY_WRITER(BIF_P,tb)) {
	    /* We will trap and as we're here this call wasn't a trap... */
	    fix_table_locked(BIF_P, tb);
	} /* Otherwise keep it as is */
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_RET(result, ret);
	break;
    case DB_ERROR_SYSRES:
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_ERROR(result, BIF_P, SYSTEM_LIMIT);
	break;
    default:
	db_unlock(tb, LCK_WRITE);
	ERTS_BIF_PREP_ERROR(result, BIF_P, BADARG);
	break;
    }

    erts_match_set_release_result(BIF_P);

    return result;
}


/* We get here instead of in the real BIF when trapping */
static BIF_RETTYPE ets_select_trap_1(Process *p, Eterm a1)
{
    BIF_RETTYPE result;
    DbTable* tb;
    int cret;
    Eterm ret;
    Eterm *tptr;

    CHECK_TABLES();

    tptr = tuple_val(a1);
    ASSERT(arityval(*tptr) >= 1)

    if ((tb = db_get_table(p, tptr[1], DB_READ, LCK_WRITE)) == NULL) {
	BIF_ERROR(p, BADARG);
    }

    cret = tb->common.meth->db_select_continue(p, tb,a1,&ret);

    switch (cret) {
    case DB_ERROR_NONE:
	if(IS_HASH_TABLE(tb->common.status) && !DID_TRAP(p,ret) &&