Merge branch 'rickard/ets-tab-delete/OTP-8999' into dev

* rickard/ets-tab-delete/OTP-8999:
  Safe deallocation of ETS-table structures
  Fix rwlock resource leak when hitting system limit

Conflicts:
	erts/emulator/beam/erl_process.h
	erts/emulator/beam/erl_process.c

6 files changed, 202 insertions, 10 deletions

diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 408ffd12f7..b7b9c6a133 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -263,6 +263,8 @@ type	XPORTS_LIST	SHORT_LIVED	SYSTEM		extra_port_list
 type	PROC_LCK_WTR	LONG_LIVED	SYSTEM		proc_lock_waiter
 type	PROC_LCK_QS	LONG_LIVED	SYSTEM		proc_lock_queues
 type	RUNQ_BLNS	LONG_LIVED	SYSTEM		run_queue_balancing
+type	MISC_AUX_WORK_Q	LONG_LIVED	SYSTEM		misc_aux_work_q
+type	MISC_AUX_WORK	SHORT_LIVED	SYSTEM		misc_aux_work
 +endif
 
 #
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 65565cf3c7..8875780842 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -227,10 +227,10 @@ static ERTS_INLINE DbTable* db_ref(DbTable* tb, db_lock_kind_t kind)
     }
     return tb;
 }
-
-static ERTS_INLINE DbTable* db_unref(DbTable* tb, db_lock_kind_t kind)
+	
+static void
+free_dbtable(DbTable* tb)
 {
-    if (kind != LCK_READ && !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",
@@ -257,6 +257,36 @@ static ERTS_INLINE DbTable* db_unref(DbTable* tb, db_lock_kind_t kind)
 	ASSERT(is_immed(tb->common.heir_data));
 	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));		     
 	ERTS_ETS_MISC_MEM_ADD(-sizeof(DbTable));
+}
+
+#ifdef ERTS_SMP
+static void
+chk_free_dbtable(void *vtb)
+{
+    DbTable * tb = (DbTable *) vtb;
+    ERTS_THR_MEMORY_BARRIER;
+    if (erts_refc_dectest(&tb->common.ref, 0) == 0)
+	free_dbtable(tb);
+}
+#endif
+
+static ERTS_INLINE DbTable* db_unref(DbTable* tb, db_lock_kind_t kind)
+{
+    if (kind != LCK_READ && erts_refc_dectest(&tb->common.ref, 0) == 0) {
+#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);
+	}
+#else
+	free_dbtable(tb);
+#endif
 	return NULL;
     }
     return tb;
@@ -1436,8 +1466,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);
-	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
-	ERTS_ETS_MISC_MEM_ADD(-sizeof(DbTable));
+	db_unref(tb, LCK_NONE);
 	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
     }
 
diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c
index 7a6aaa6bbe..0185baee6b 100644
--- a/erts/emulator/beam/erl_lock_check.c
+++ b/erts/emulator/beam/erl_lock_check.c
@@ -177,6 +177,8 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"async_id",				NULL			},
     {	"pix_lock",				"address"		},
     {	"run_queues_lists",			NULL			},
+    {	"misc_aux_work_queue",			"index"			},
+    {	"misc_aux_work_pre_alloc_lock",		"address"		},
     {	"sched_stat",				NULL			},
     {	"run_queue_sleep_list",			"address"		},
 #endif
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 77ee1d6ac5..fbe4381aca 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -590,6 +590,122 @@ erts_sched_finish_poke(ErtsSchedulerSleepInfo *ssi, erts_aint32_t flags)
     }
 }
 
+typedef struct erts_misc_aux_work_t_ erts_misc_aux_work_t;
+struct erts_misc_aux_work_t_ {
+    erts_misc_aux_work_t *next;
+    void (*func)(void *);
+    void *arg;
+};
+
+typedef struct {
+    erts_smp_mtx_t mtx;
+    erts_misc_aux_work_t *first;
+    erts_misc_aux_work_t *last;
+} erts_misc_aux_work_q_t;
+
+typedef union {
+    erts_misc_aux_work_q_t data;
+    char align[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(erts_misc_aux_work_q_t))];
+} erts_algnd_misc_aux_work_q_t;
+
+static erts_algnd_misc_aux_work_q_t *misc_aux_work_queues;
+
+ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(misc_aux_work,
+				 erts_misc_aux_work_t,
+				 200,
+				 ERTS_ALC_T_MISC_AUX_WORK)
+
+static void
+init_misc_aux_work(void)
+{
+    int ix;
+
+    init_misc_aux_work_alloc();
+
+    misc_aux_work_queues = erts_alloc(ERTS_ALC_T_MISC_AUX_WORK_Q,
+				      (sizeof(erts_algnd_misc_aux_work_q_t)
+				       *(erts_no_schedulers+1)));
+    if ((((UWord) misc_aux_work_queues) & ERTS_CACHE_LINE_MASK) != 0)
+	misc_aux_work_queues = ((erts_algnd_misc_aux_work_q_t *)
+				   ((((UWord) misc_aux_work_queues)
+				     & ~ERTS_CACHE_LINE_MASK)
+				    + ERTS_CACHE_LINE_SIZE));
+
+    for (ix = 0; ix < erts_no_schedulers; ix++) {
+	erts_smp_mtx_init_x(&misc_aux_work_queues[ix].data.mtx,
+			    "misc_aux_work_queue",
+			    make_small(ix + 1));
+	misc_aux_work_queues[ix].data.first = NULL;
+	misc_aux_work_queues[ix].data.last = NULL;
+    }
+}
+
+static void
+handle_misc_aux_work(ErtsSchedulerData *esdp)
+{
+    int ix = (int) esdp->no - 1;
+    erts_misc_aux_work_t *mawp;
+
+    erts_smp_mtx_lock(&misc_aux_work_queues[ix].data.mtx);
+    mawp = misc_aux_work_queues[ix].data.first;
+    misc_aux_work_queues[ix].data.first = NULL;
+    misc_aux_work_queues[ix].data.last = NULL;
+    erts_smp_mtx_unlock(&misc_aux_work_queues[ix].data.mtx);
+
+    while (mawp) {
+	erts_misc_aux_work_t *free_mawp;
+	mawp->func(mawp->arg);
+	free_mawp = mawp;
+	mawp = mawp->next;
+	misc_aux_work_free(free_mawp);
+    }
+}
+
+void
+erts_smp_schedule_misc_aux_work(int ignore_self,
+				int max_sched,
+				void (*func)(void *),
+				void *arg)
+{
+    int ix, ignore_ix = -1;
+
+    if (ignore_self) {
+	ErtsSchedulerData *esdp = erts_get_scheduler_data();
+	if (esdp)
+	    ignore_ix = (int) esdp->no - 1;
+    }
+
+    ASSERT(0 <= max_sched && max_sched <= erts_no_schedulers);
+
+    for (ix = 0; ix < max_sched; ix++) {
+	long aux_work;
+	erts_misc_aux_work_t *mawp;
+	ErtsSchedulerSleepInfo *ssi;
+	if (ix == ignore_ix)
+	    continue;
+
+	mawp = misc_aux_work_alloc();
+
+	mawp->func = func;
+	mawp->arg = arg;
+	mawp->next = NULL;
+
+	erts_smp_mtx_lock(&misc_aux_work_queues[ix].data.mtx);
+	if (!misc_aux_work_queues[ix].data.last)
+	    misc_aux_work_queues[ix].data.first = mawp;
+	else
+	    misc_aux_work_queues[ix].data.last->next = mawp;
+	misc_aux_work_queues[ix].data.last = mawp;
+	erts_smp_mtx_unlock(&misc_aux_work_queues[ix].data.mtx);
+
+	ssi = ERTS_SCHED_SLEEP_INFO_IX(ix);
+	aux_work = erts_smp_atomic32_bor(&ssi->aux_work,
+					 ERTS_SSI_AUX_WORK_MISC);
+	if ((aux_work & ERTS_SSI_AUX_WORK_MISC) == 0)
+	    erts_sched_poke(ssi);
+   }
+}
+
 #ifdef ERTS_SMP_SCHEDULERS_NEED_TO_CHECK_CHILDREN
 void
 erts_smp_notify_check_children_needed(void)
@@ -615,6 +731,12 @@ blockable_aux_work(ErtsSchedulerData *esdp,
 		   erts_aint32_t aux_work)
 {
     if (aux_work & ERTS_SSI_BLOCKABLE_AUX_WORK_MASK) {
+	if (aux_work & ERTS_SSI_AUX_WORK_MISC) {
+	    aux_work = erts_smp_atomic32_band(&ssi->aux_work,
+					      ~ERTS_SSI_AUX_WORK_MISC);
+	    aux_work &= ~ERTS_SSI_AUX_WORK_MISC;
+	    handle_misc_aux_work(esdp);
+	}
 #ifdef ERTS_SMP_SCHEDULERS_NEED_TO_CHECK_CHILDREN
 	if (aux_work & ERTS_SSI_AUX_WORK_CHECK_CHILDREN) {
 	    aux_work = erts_smp_atomic32_band(&ssi->aux_work,
@@ -2612,6 +2734,8 @@ erts_init_scheduling(int mrq, int no_schedulers, int no_schedulers_online)
 
     erts_smp_atomic32_init(&doing_sys_schedule, 0);
 
+    init_misc_aux_work();
+
 #else /* !ERTS_SMP */
     {
 	ErtsSchedulerData *esdp;
@@ -2739,6 +2863,19 @@ resume_process(Process *p)
     p->rstatus = P_FREE;    
 }
 
+int
+erts_get_max_no_executing_schedulers(void)
+{
+#ifdef ERTS_SMP
+    if (erts_smp_atomic32_read(&schdlr_sspnd.changing))
+	return (int) erts_no_schedulers;
+    ERTS_THR_MEMORY_BARRIER;
+    return (int) erts_smp_atomic32_read(&schdlr_sspnd.active);
+#else
+    return 1;
+#endif
+}
+
 #ifdef ERTS_SMP
 
 static void
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 8e32121a68..d927415f37 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -236,16 +236,14 @@ typedef enum {
   | ERTS_SSI_FLG_WAITING				\
   | ERTS_SSI_FLG_SUSPENDED)
 
-
-#if !defined(ERTS_SCHED_NEED_BLOCKABLE_AUX_WORK) \
-    && defined(ERTS_SMP_SCHEDULERS_NEED_TO_CHECK_CHILDREN)
 #define ERTS_SCHED_NEED_BLOCKABLE_AUX_WORK
-#endif
 
 #define ERTS_SSI_AUX_WORK_CHECK_CHILDREN	(((erts_aint32_t) 1) << 0)
+#define ERTS_SSI_AUX_WORK_MISC			(((erts_aint32_t) 1) << 1)
 
 #define ERTS_SSI_BLOCKABLE_AUX_WORK_MASK \
-  (ERTS_SSI_AUX_WORK_CHECK_CHILDREN)
+  (ERTS_SSI_AUX_WORK_CHECK_CHILDREN \
+   | ERTS_SSI_AUX_WORK_MISC)
 #define ERTS_SSI_NONBLOCKABLE_AUX_WORK_MASK \
   (0)
 
@@ -1034,6 +1032,7 @@ int erts_sched_set_wakeup_limit(char *str);
 #ifdef DEBUG
 void erts_dbg_multi_scheduling_return_trap(Process *, Eterm);
 #endif
+int erts_get_max_no_executing_schedulers(void);
 #ifdef ERTS_SMP
 ErtsSchedSuspendResult
 erts_schedulers_state(Uint *, Uint *, Uint *, int);
@@ -1048,6 +1047,11 @@ int erts_is_multi_scheduling_blocked(void);
 Eterm erts_multi_scheduling_blockers(Process *);
 void erts_start_schedulers(void);
 void erts_smp_notify_check_children_needed(void);
+void
+erts_smp_schedule_misc_aux_work(int ignore_self,
+				int max_sched,
+				void (*func)(void *),
+				void *arg);
 #endif
 void erts_sched_notify_check_cpu_bind(void);
 Uint erts_active_schedulers(void);
diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl
index 620848003c..4e789790f6 100644
--- a/lib/stdlib/test/ets_SUITE.erl
+++ b/lib/stdlib/test/ets_SUITE.erl
@@ -5325,7 +5325,25 @@ my_tab_to_list(_Ts,'$end_of_table', Acc) -> lists:reverse(Acc);
 my_tab_to_list(Ts,Key, Acc) ->
     my_tab_to_list(Ts,ets:next(Ts,Key),[ets:lookup(Ts, Key)| Acc]).
 
+wait_for_all_schedulers_online_to_execute() ->
+    PMs = lists:map(fun (Sched) ->
+			    spawn_opt(fun () -> ok end,
+				      [monitor, {scheduler, Sched}])
+		    end,
+		    lists:seq(1,erlang:system_info(schedulers_online))),
+    lists:foreach(fun ({P, M}) ->
+			  receive
+			      {'DOWN', M, process, P, _} -> ok
+			  end
+		  end,
+		  PMs),
+    ok.
+
 etsmem() ->
+    %% Wait until it is guaranteed that all already scheduled
+    %% deallocations of DbTable structures have completed.
+    wait_for_all_schedulers_online_to_execute(),
+
     AllTabs = lists:map(fun(T) -> {T,ets:info(T,name),ets:info(T,size),
 				   ets:info(T,memory),ets:info(T,type)} 
 			end, ets:all()),