Stop using reference counter when write accessing ETS-tables

1 files changed, 59 insertions, 77 deletions

diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index e662b17592..6f19acdc3b 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -219,43 +219,34 @@ Export ets_select_continue_exp;
  * Static traps
  */
 static Export ets_delete_continue_exp;
-
-static ERTS_INLINE DbTable* db_ref(DbTable* tb, db_lock_kind_t kind)
-{
-    if (tb != NULL && kind != LCK_READ) {
-	erts_refc_inc(&tb->common.ref, 2);
-    }
-    return tb;
-}
 	
 static void
 free_dbtable(DbTable* tb)
 {
 #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), 
+	    erts_fprintf(stderr, "ets: free_dbtable 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", 
+	erts_fprintf(stderr, "ets: free_dbtable(%T) deleted!!!\r\n",
 		     tb->common.id);
 
-	erts_fprintf(stderr, "ets: db_unref: meta_pid_to_tab common.memory_size = %ld\n",
+	erts_fprintf(stderr, "ets: free_dbtable: 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_fprintf(stderr, "ets: free_dbtable: 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);
 	erts_smp_mtx_destroy(&tb->common.fixlock);
 #endif
 	ASSERT(is_immed(tb->common.heir_data));
-	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));		     
+	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
 	ERTS_ETS_MISC_MEM_ADD(-sizeof(DbTable));
 }
 
@@ -270,26 +261,26 @@ chk_free_dbtable(void *vtb)
 }
 #endif
 
-static ERTS_INLINE DbTable* db_unref(DbTable* tb, db_lock_kind_t kind)
+static void schedule_free_dbtable(DbTable* tb)
 {
-    if (kind != LCK_READ && erts_refc_dectest(&tb->common.ref, 0) == 0) {
+    /*
+     * NON-SMP case: Caller is *not* allowed to access the *tb
+     *               structure after this function has returned!          
+     * SMP case:     Caller is allowed to access the *tb structure
+     *               until the bif has returned (we typically
+     *               need to unlock the table lock after this
+     *               function has returned).
+     */
 #ifdef ERTS_SMP
-	int scheds = erts_get_max_no_executing_schedulers();
-	if (scheds == 1)
-	    free_dbtable(tb);
-	else {
-	    int refs = scheds - 1;
-	    ASSERT(scheds > 1);
-	    ERTS_THR_MEMORY_BARRIER;
-	    erts_refc_add(&tb->common.ref, refs, refs);
-	    erts_smp_schedule_misc_aux_work(1, scheds, chk_free_dbtable, tb);
-	}
+    int scheds = erts_get_max_no_executing_schedulers();
+    ASSERT(scheds >= 1);
+    ASSERT(erts_refc_read(&tb->common.ref, 0) == 0);
+    erts_refc_init(&tb->common.ref, scheds);
+    ERTS_THR_MEMORY_BARRIER;
+    erts_smp_schedule_misc_aux_work(0, scheds, chk_free_dbtable, tb);
 #else
-	free_dbtable(tb);
+    free_dbtable(tb);
 #endif
-	return NULL;
-    }
-    return tb;
 }
 
 static ERTS_INLINE void db_init_lock(DbTable* tb, int use_frequent_read_lock,
@@ -300,8 +291,6 @@ static ERTS_INLINE void db_init_lock(DbTable* tb, int use_frequent_read_lock,
     if (use_frequent_read_lock)
 	rwmtx_opt.type = ERTS_SMP_RWMTX_TYPE_FREQUENT_READ;
 #endif
-    erts_refc_init(&tb->common.ref, 1);
-    erts_refc_init(&tb->common.fixref, 0);
 #ifdef ERTS_SMP
     erts_smp_rwmtx_init_opt_x(&tb->common.rwlock, &rwmtx_opt,
 			      rwname, tb->common.the_name);
@@ -310,7 +299,7 @@ static ERTS_INLINE void db_init_lock(DbTable* tb, int use_frequent_read_lock,
 #endif
 }
 
-static ERTS_INLINE void db_lock_take_over_ref(DbTable* tb, db_lock_kind_t kind)
+static ERTS_INLINE void db_lock(DbTable* tb, db_lock_kind_t kind)
 {
 #ifdef ERTS_SMP
     ASSERT(tb != meta_pid_to_tab && tb != meta_pid_to_fixed_tab);
@@ -338,16 +327,13 @@ static ERTS_INLINE void db_lock_take_over_ref(DbTable* tb, db_lock_kind_t kind)
 #endif
 }
 
-static ERTS_INLINE void db_lock(DbTable* tb, db_lock_kind_t kind)
-{
-    (void) db_ref(tb, kind);
-#ifdef ERTS_SMP
-    db_lock_take_over_ref(tb, kind);
-#endif
-}
-
 static ERTS_INLINE void db_unlock(DbTable* tb, db_lock_kind_t kind)
 {
+    /*
+     * In NON-SMP case tb may refer to an already deallocated
+     * DbTable structure. That is, ONLY the SMP case is allowed
+     * to follow the tb pointer!
+     */
 #ifdef ERTS_SMP
     ASSERT(tb != meta_pid_to_tab && tb != meta_pid_to_fixed_tab);
 
@@ -374,7 +360,6 @@ static ERTS_INLINE void db_unlock(DbTable* tb, db_lock_kind_t kind)
 	}
     }
 #endif
-    (void) db_unref(tb, kind); /* May delete table... */
 }
 
 
@@ -401,6 +386,13 @@ DbTable* db_get_table_aux(Process *p,
     DbTable *tb = NULL;
     erts_smp_rwmtx_t *mtl = NULL;
 
+    /*
+     * IMPORTANT: Only scheduler threads are allowed
+     *            to access tables. Memory management
+     *            depend on it.
+     */
+    ASSERT(erts_get_scheduler_data());
+
     if (is_small(id)) {
 	Uint slot = unsigned_val(id) & meta_main_tab_slot_mask;
 	if (!meta_already_locked) {
@@ -414,12 +406,8 @@ DbTable* db_get_table_aux(Process *p,
 			       || erts_lc_rwmtx_is_rwlocked(test_mtl));
 	}
 #endif
-	if (slot < db_max_tabs && IS_SLOT_ALIVE(slot)) {
-	    /* SMP: inc to prevent race, between unlock of meta_main_tab_lock
-	     * and the table locking outside the meta_main_tab_lock
-	     */
-	    tb = db_ref(meta_main_tab[slot].u.tb, kind);
-	}
+	if (slot < db_max_tabs && IS_SLOT_ALIVE(slot))
+	    tb = meta_main_tab[slot].u.tb;
     }
     else if (is_atom(id)) {
 	struct meta_name_tab_entry* bucket = meta_name_tab_bucket(id,&mtl);
@@ -433,16 +421,15 @@ DbTable* db_get_table_aux(Process *p,
 
 	if (bucket->pu.tb != NULL) {
 	    if (is_atom(bucket->u.name_atom)) { /* single */
-		if (bucket->u.name_atom == id) {
-		    tb = db_ref(bucket->pu.tb, kind);
-		}
+		if (bucket->u.name_atom == id)
+		    tb = bucket->pu.tb;
 	    }
 	    else { /* multi */
 		Uint cnt = unsigned_val(bucket->u.mcnt);
 		Uint i;
 		for (i=0; i<cnt; i++) {
 		    if (bucket->pu.mvec[i].u.name_atom == id) {
-			tb = db_ref(bucket->pu.mvec[i].pu.tb, kind);
+			tb = bucket->pu.mvec[i].pu.tb;
 			break;
 		    }
 		}
@@ -450,7 +437,7 @@ DbTable* db_get_table_aux(Process *p,
 	}
     }
     if (tb) {
-	db_lock_take_over_ref(tb, kind);
+	db_lock(tb, kind);
 	if (tb->common.id != id
 	    || ((tb->common.status & what) == 0 && p->id != tb->common.owner)) {
 	    db_unlock(tb, kind);
@@ -624,11 +611,11 @@ done:
 */
 static ERTS_INLINE void local_fix_table(DbTable* tb)
 {
-    erts_refc_inc(&tb->common.fixref, 1);
+    erts_refc_inc(&tb->common.ref, 1);
 }	    
 static ERTS_INLINE void local_unfix_table(DbTable* tb)
 {	
-    if (erts_refc_dectest(&tb->common.fixref, 0) == 0) {
+    if (erts_refc_dectest(&tb->common.ref, 0) == 0) {
 	ASSERT(IS_HASH_TABLE(tb->common.status));
 	db_unfix_table_hash(&(tb->hash));
     }
@@ -1444,6 +1431,7 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
     tb->common.type = status & ERTS_ETS_TABLE_TYPES;
     /* Note, 'type' is *read only* from now on... */
 #endif
+    erts_refc_init(&tb->common.ref, 0);
     db_init_lock(tb, status & (DB_FINE_LOCKED|DB_FREQ_READ),
 		 "db_tab", "db_tab_fix");
     tb->common.keypos = keypos;
@@ -1466,7 +1454,7 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
 				      "** Too many db tables **\n");
 	free_heir_data(tb);
 	tb->common.meth->db_free_table(tb);
-	db_unref(tb, LCK_NONE);
+	free_dbtable(tb);
 	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
     }
 
@@ -1500,9 +1488,10 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
 	free_slot(slot);
 	erts_smp_rwmtx_rwunlock(mmtl);
 
-	db_lock_take_over_ref(tb,LCK_WRITE);
+	db_lock(tb,LCK_WRITE);
 	free_heir_data(tb);
 	tb->common.meth->db_free_table(tb);
+	schedule_free_dbtable(tb);
 	db_unlock(tb,LCK_WRITE);
 	BIF_ERROR(BIF_P, BADARG);
     }
@@ -2874,8 +2863,7 @@ void init_db(void)
     meta_pid_to_tab->common.meth   = &db_hash;
     meta_pid_to_tab->common.compress = 0;
 
-    erts_refc_init(&meta_pid_to_tab->common.ref, 1);
-    erts_refc_init(&meta_pid_to_tab->common.fixref, 0);
+    erts_refc_init(&meta_pid_to_tab->common.ref, 0);
     /* Neither rwlock or fixlock used
     db_init_lock(meta_pid_to_tab, "meta_pid_to_tab", "meta_pid_to_tab_FIX");*/
 
@@ -2907,8 +2895,7 @@ void init_db(void)
     meta_pid_to_fixed_tab->common.meth   = &db_hash;
     meta_pid_to_fixed_tab->common.compress = 0;
 
-    erts_refc_init(&meta_pid_to_fixed_tab->common.ref, 1);
-    erts_refc_init(&meta_pid_to_fixed_tab->common.fixref, 0);
+    erts_refc_init(&meta_pid_to_fixed_tab->common.ref, 0);
     /* Neither rwlock or fixlock used
     db_init_lock(meta_pid_to_fixed_tab, "meta_pid_to_fixed_tab", "meta_pid_to_fixed_tab_FIX");*/
 
@@ -3066,12 +3053,10 @@ retry:
 				to_pid, to_locks,
 				ERTS_P2P_FLG_TRY_LOCK);
     if (to_proc == ERTS_PROC_LOCK_BUSY) {
-	db_ref(tb, LCK_NONE); /* while unlocked */
 	db_unlock(tb,LCK_WRITE);    
 	to_proc = erts_pid2proc(p, ERTS_PROC_LOCK_MAIN,
 				to_pid, to_locks);    
 	db_lock(tb,LCK_WRITE);
-	tb = db_unref(tb, LCK_NONE);
 	ASSERT(tb != NULL);
     
 	if (tb->common.owner != p->id) {
@@ -3182,13 +3167,13 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
 		erts_smp_rwmtx_t *mmtl = get_meta_main_tab_lock(ix);
 		erts_smp_rwmtx_rlock(mmtl);
 		if (!IS_SLOT_FREE(ix)) {
-		    tb = db_ref(GET_ANY_SLOT_TAB(ix), LCK_WRITE);
+		    tb = GET_ANY_SLOT_TAB(ix);
 		    ASSERT(tb);
 		}
 		erts_smp_rwmtx_runlock(mmtl);
 		if (tb) {
 		    int do_yield;
-		    db_lock_take_over_ref(tb, LCK_WRITE);
+		    db_lock(tb, LCK_WRITE);
 		    /* Ownership may have changed since
 		       we looked up the table. */
 		    if (tb->common.owner != pid) {
@@ -3270,7 +3255,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
 		erts_smp_rwmtx_t *mmtl = get_meta_main_tab_lock(ix);
 		erts_smp_rwmtx_rlock(mmtl);
 		if (IS_SLOT_ALIVE(ix)) {
-		    tb = db_ref(meta_main_tab[ix].u.tb, LCK_WRITE_REC);
+		    tb = meta_main_tab[ix].u.tb;
 		    ASSERT(tb);
 		}
 		erts_smp_rwmtx_runlock(mmtl);
@@ -3278,7 +3263,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
 		    int reds;
 		    DbFixation** pp;
 
-		    db_lock_take_over_ref(tb, LCK_WRITE_REC);
+		    db_lock(tb, LCK_WRITE_REC);
 		    #ifdef ERTS_SMP
 		    erts_smp_mtx_lock(&tb->common.fixlock);
 		    #endif
@@ -3289,7 +3274,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
 			if ((*pp)->pid == pid) {
 			    DbFixation* fix = *pp;
 			    long diff = -(long)fix->counter;
-			    erts_refc_add(&tb->common.fixref,diff,0);
+			    erts_refc_add(&tb->common.ref,diff,0);
 			    *pp = fix->next;
 			    erts_db_free(ERTS_ALC_T_DB_FIXATION,
 					 tb, fix, sizeof(DbFixation));
@@ -3364,7 +3349,7 @@ static void fix_table_locked(Process* p, DbTable* tb)
 #ifdef ERTS_SMP
     erts_smp_mtx_lock(&tb->common.fixlock);
 #endif
-    erts_refc_inc(&tb->common.fixref,1);
+    erts_refc_inc(&tb->common.ref,1);
     fix = tb->common.fixations;
     if (fix == NULL) { 
 	get_now(&(tb->common.megasec),
@@ -3418,7 +3403,7 @@ static void unfix_table_locked(Process* p,  DbTable* tb,
     for (pp = &tb->common.fixations; *pp != NULL; pp = &(*pp)->next) {
 	if ((*pp)->pid == p->id) {
 	    DbFixation* fix = *pp;
-	    erts_refc_dec(&tb->common.fixref,0);
+	    erts_refc_dec(&tb->common.ref,0);
 	    --(fix->counter);
 	    ASSERT(fix->counter >= 0);
 	    if (fix->counter > 0) {
@@ -3448,7 +3433,6 @@ unlocked:
 #ifdef ERTS_SMP
 	if (*kind_p == LCK_READ && tb->common.is_thread_safe) {
 	    /* Must have write lock while purging pseudo-deleted (OTP-8166) */
-	    db_ref(tb, LCK_WRITE); /* LCK_WRITE need it, but not LCK_READ */
 	    erts_smp_rwmtx_runlock(&tb->common.rwlock);
 	    erts_smp_rwmtx_rwlock(&tb->common.rwlock);
 	    *kind_p = LCK_WRITE;
@@ -3467,6 +3451,8 @@ static void free_fixations_locked(DbTable *tb)
 
     fix = tb->common.fixations;
     while (fix != NULL) {
+	long diff = -(long)fix->counter;
+	erts_refc_add(&tb->common.ref,diff,0);
 	next_fix = fix->next;
 	db_meta_lock(meta_pid_to_fixed_tab, LCK_WRITE_REC);
 	db_erase_bag_exact2(meta_pid_to_fixed_tab,
@@ -3590,10 +3576,6 @@ static int free_table_cont(Process *p,
 	mmtl = get_meta_main_tab_lock(tb->common.slot);
 #ifdef ERTS_SMP
 	if (erts_smp_rwmtx_tryrwlock(mmtl) == EBUSY) {
-	    /*
-	     * We keep our increased refc over this op in order to
-	     * prevent the table from disapearing.
-	     */
 	    erts_smp_rwmtx_rwunlock(&tb->common.rwlock);
 	    erts_smp_rwmtx_rwlock(mmtl);
 	    erts_smp_rwmtx_rwlock(&tb->common.rwlock);
@@ -3608,7 +3590,7 @@ static int free_table_cont(Process *p,
 				make_small(tb->common.slot));
 	    db_meta_unlock(meta_pid_to_tab, LCK_WRITE_REC);
 	}
-	db_unref(tb, LCK_NONE);
+	schedule_free_dbtable(tb);
 	BUMP_REDS(p, 100);
 	return 0;
     }