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

1 files changed, 137 insertions, 0 deletions

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