erts: Suspend processes waiting for code_ix lock

This will prevent blocking entrire schedulers in the rare case when
several processes are racing to load/upgrade/delete/purge code.

5 files changed, 95 insertions, 55 deletions

diff --git a/erts/emulator/beam/beam_bif_load.c b/erts/emulator/beam/beam_bif_load.c
index 8a2169f9e6..1227af8cbb 100644
--- a/erts/emulator/beam/beam_bif_load.c
+++ b/erts/emulator/beam/beam_bif_load.c
@@ -38,27 +38,11 @@
 static void set_default_trace_pattern(Eterm module);
 static Eterm check_process_code(Process* rp, Module* modp);
 static void delete_code(Module* modp);
-static int purge_module(Process*, int module);
 static void decrement_refc(BeamInstr* code);
 static int is_native(BeamInstr* code);
 static int any_heap_ref_ptrs(Eterm* start, Eterm* end, char* mod_start, Uint mod_size);
 static int any_heap_refs(Eterm* start, Eterm* end, char* mod_start, Uint mod_size);
 
-BIF_RETTYPE purge_module_1(BIF_ALIST_1)
-{
-    int purge_res;
-
-    if (is_not_atom(BIF_ARG_1)) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
-
-    purge_res = purge_module(BIF_P, atom_val(BIF_ARG_1));
-
-    if (purge_res < 0) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
-    BIF_RET(am_true);
-}
 
 BIF_RETTYPE code_is_module_native_1(BIF_ALIST_1)
 {
@@ -178,11 +162,15 @@ finish_loading_1(BIF_ALIST_1)
 {
     int i;
     int n;
-    struct m* p;
+    struct m* p = NULL;
     Uint exceptions;
     Eterm res;
     int is_blocking = 0;
 
+    if (!erts_try_lock_code_ix(BIF_P)) {
+	ERTS_BIF_YIELD1(bif_export[BIF_finish_loading_1], BIF_P, BIF_ARG_1);
+    }
+
     /*
      * Validate the argument before we start loading; it must be a
      * proper list where each element is a magic binary containing
@@ -194,7 +182,8 @@ finish_loading_1(BIF_ALIST_1)
 
     n = list_length(BIF_ARG_1);
     if (n == -1) {
-	BIF_ERROR(BIF_P, BADARG);
+	ERTS_BIF_PREP_ERROR(res, BIF_P, BADARG);
+	goto done;
     }
     p = erts_alloc(ERTS_ALC_T_LOADER_TMP, n*sizeof(struct m));
 
@@ -209,15 +198,15 @@ finish_loading_1(BIF_ALIST_1)
 	ProcBin* pb;
 
 	if (!ERTS_TERM_IS_MAGIC_BINARY(term)) {
-	error:
-	    erts_free(ERTS_ALC_T_LOADER_TMP, p);
-	    BIF_ERROR(BIF_P, BADARG);
+	    ERTS_BIF_PREP_ERROR(res, BIF_P, BADARG);
+	    goto done;
 	}
 	pb = (ProcBin*) binary_val(term);
 	p[i].code = pb->val;
 	p[i].module = erts_module_for_prepared_code(p[i].code);
 	if (p[i].module == NIL) {
-	    goto error;
+	    ERTS_BIF_PREP_ERROR(res, BIF_P, BADARG);
+	    goto done;
 	}
 	BIF_ARG_1 = CDR(cons);
     }
@@ -230,8 +219,8 @@ finish_loading_1(BIF_ALIST_1)
      */
 
     if (n > 1) {
-	erts_free(ERTS_ALC_T_LOADER_TMP, p);
-	BIF_ERROR(BIF_P, SYSTEM_LIMIT);
+	ERTS_BIF_PREP_ERROR(res, BIF_P, SYSTEM_LIMIT);
+	goto done;
     }
 
     /*
@@ -243,7 +232,6 @@ finish_loading_1(BIF_ALIST_1)
      */
 
     res = am_ok;
-    erts_lock_code_ix();
     erts_start_staging_code_ix();
 
     for (i = 0; i < n; i++) {
@@ -314,7 +302,10 @@ finish_loading_1(BIF_ALIST_1)
 	erts_commit_staging_code_ix();
     }
 
-    erts_free(ERTS_ALC_T_LOADER_TMP, p);
+done:
+    if (p) {
+	erts_free(ERTS_ALC_T_LOADER_TMP, p);
+    }
     if (!is_blocking) {
 	erts_unlock_code_ix();
     } else {
@@ -418,10 +409,14 @@ BIF_RETTYPE delete_module_1(BIF_ALIST_1)
     int is_blocking = 0;
     Eterm res = NIL;
 
-    if (is_not_atom(BIF_ARG_1))
+    if (is_not_atom(BIF_ARG_1)) {
 	goto badarg;
+    }
+
+    if (!erts_try_lock_code_ix(BIF_P)) {
+	ERTS_BIF_YIELD1(bif_export[BIF_delete_module_1], BIF_P, BIF_ARG_1);
+    }
 
-    erts_lock_code_ix();
     do {
 	erts_start_staging_code_ix();
 	code_ix = erts_staging_code_ix();
@@ -856,18 +851,23 @@ any_heap_refs(Eterm* start, Eterm* end, char* mod_start, Uint mod_size)
 
 #undef in_area
 
-
-static int
-purge_module(Process* c_p, int module)
+BIF_RETTYPE purge_module_1(BIF_ALIST_1)
 {
     ErtsCodeIndex code_ix;
     BeamInstr* code;
     BeamInstr* end;
     Module* modp;
     int is_blocking = 0;
-    int ret;
+    Eterm ret;
+
+    if (is_not_atom(BIF_ARG_1)) {
+	BIF_ERROR(BIF_P, BADARG);
+    }
+
+    if (!erts_try_lock_code_ix(BIF_P)) {
+	ERTS_BIF_YIELD1(bif_export[BIF_purge_module_1], BIF_P, BIF_ARG_1);
+    }
 
-    erts_lock_code_ix();
 retry:
     code_ix = erts_active_code_ix();
 
@@ -875,8 +875,8 @@ retry:
      * Correct module?
      */
 
-    if ((modp = erts_get_module(make_atom(module), code_ix)) == NULL) {
-	ret = -2;
+    if ((modp = erts_get_module(BIF_ARG_1, code_ix)) == NULL) {
+	ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
     }
     else {
 	erts_rwlock_old_code(code_ix);
@@ -885,7 +885,7 @@ retry:
 	 * Any code to purge?
 	 */
 	if (modp->old.code == 0) {
-	    ret = -1;
+	    ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
 	}
 	else {
 	    /*
@@ -896,7 +896,7 @@ retry:
 		    /* ToDo: Do unload nif without blocking */
 		    erts_rwunlock_old_code(code_ix);
 		    erts_unlock_code_ix();
-		    erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_MAIN);
+		    erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
 		    erts_smp_thr_progress_block();
 		    is_blocking = 1;
 		    goto retry;
@@ -922,13 +922,13 @@ retry:
 	    modp->old.code_length = 0;
 	    modp->old.catches = BEAM_CATCHES_NIL;
 	    erts_remove_from_ranges(code);
-	    ret = 0;
+	    ERTS_BIF_PREP_RET(ret, am_true);
 	}
 	erts_rwunlock_old_code(code_ix);
     }
     if (is_blocking) {
 	erts_smp_thr_progress_unblock();
-	erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_MAIN);
+	erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
     }
     else {
 	erts_unlock_code_ix();
diff --git a/erts/emulator/beam/code_ix.c b/erts/emulator/beam/code_ix.c
index a8b6599e61..a1db003127 100644
--- a/erts/emulator/beam/code_ix.c
+++ b/erts/emulator/beam/code_ix.c
@@ -36,7 +36,13 @@
 erts_smp_atomic32_t the_active_code_index;
 erts_smp_atomic32_t the_staging_code_index;
 
-static erts_smp_mtx_t sverk_code_ix_lock; /*SVERK FIXME */
+static int the_code_ix_lock = 0;
+struct code_ix_queue_item {
+    Process *p;
+    struct code_ix_queue_item* next;
+};
+static struct code_ix_queue_item* the_code_ix_queue = NULL;
+static erts_smp_mtx_t the_code_ix_queue_lock;
 
 void erts_code_ix_init(void)
 {
@@ -46,7 +52,7 @@ void erts_code_ix_init(void)
      */
     erts_smp_atomic32_init_nob(&the_active_code_index, 0);
     erts_smp_atomic32_init_nob(&the_staging_code_index, 0);
-    erts_smp_mtx_init_x(&sverk_code_ix_lock, "sverk_code_ix_lock", NIL); /*SVERK FIXME */
+    erts_smp_mtx_init(&the_code_ix_queue_lock, "code_ix_queue");
     CIX_TRACE("init");
 }
 
@@ -88,23 +94,54 @@ void erts_abort_staging_code_ix(void)
 }
 
 
-/* Lock code_ix (enqueue and suspend until we get it)
-*/
-void erts_lock_code_ix(void)
+/* Try lock code_ix
+ * Calller _must_ yield if we return 0
+ */
+int erts_try_lock_code_ix(Process* c_p)
 {
-    erts_smp_mtx_lock(&sverk_code_ix_lock); /*SVERK FIXME */
+    int success;
+
+    erts_smp_mtx_lock(&the_code_ix_queue_lock);
+    success = !the_code_ix_lock;
+    if (success) {
+	the_code_ix_lock = 1;
+    }
+    else { /* Already locked */
+	struct code_ix_queue_item* qitem;
+	qitem = erts_alloc(ERTS_ALC_T_CODE_IX_LOCK_Q, sizeof(*qitem));
+	qitem->p = c_p;
+	erts_smp_proc_inc_refc(c_p);
+	qitem->next = the_code_ix_queue;
+	the_code_ix_queue = qitem;
+	erts_suspend(c_p, ERTS_PROC_LOCK_MAIN, NULL);
+    }
+   erts_smp_mtx_unlock(&the_code_ix_queue_lock);
+   return success;
 }
 
-/* Unlock code_ix (resume first waiter)
+/* Unlock code_ix (resume all waiters)
 */
-void erts_unlock_code_ix(void)
+void erts_unlock_code_ix()
 {
-    erts_smp_mtx_unlock(&sverk_code_ix_lock); /*SVERK FIXME */
+    erts_smp_mtx_lock(&the_code_ix_queue_lock);
+    while (the_code_ix_queue != NULL) { /* unleash the entire herd */
+	struct code_ix_queue_item* qitem = the_code_ix_queue;
+	erts_smp_proc_lock(qitem->p, ERTS_PROC_LOCK_STATUS);
+	if (!ERTS_PROC_IS_EXITING(qitem->p)) {
+	    erts_resume(qitem->p, ERTS_PROC_LOCK_STATUS);
+	}
+	erts_smp_proc_unlock(qitem->p, ERTS_PROC_LOCK_STATUS);
+	the_code_ix_queue = qitem->next;
+	erts_smp_proc_dec_refc(qitem->p);
+	erts_free(ERTS_ALC_T_CODE_IX_LOCK_Q, qitem);
+    }
+    the_code_ix_lock = 0;
+    erts_smp_mtx_unlock(&the_code_ix_queue_lock);
 }
 
 #ifdef ERTS_ENABLE_LOCK_CHECK
 int erts_is_code_ix_locked(void)
 {
-    return erts_smp_lc_mtx_is_locked(&sverk_code_ix_lock);
+    return the_code_ix_lock;
 }
 #endif
diff --git a/erts/emulator/beam/code_ix.h b/erts/emulator/beam/code_ix.h
index 99fd4ec348..66543fa2a8 100644
--- a/erts/emulator/beam/code_ix.h
+++ b/erts/emulator/beam/code_ix.h
@@ -55,10 +55,13 @@
 #  endif
 #  include "sys.h"
 #endif
+struct process;
+
 
 #define ERTS_NUM_CODE_IX 3
 typedef unsigned ErtsCodeIndex;
 
+
 /* Called once at emulator initialization.
  */
 void erts_code_ix_init(void);
@@ -78,14 +81,13 @@ ErtsCodeIndex erts_active_code_ix(void);
 ERTS_GLB_INLINE
 ErtsCodeIndex erts_staging_code_ix(void);
 
-/* Lock code_ix.
- * Gives (exclusive) access to the staging area and write access to active code index.
- * ToDo: Waiting process should be queued and return to be suspended.
+/* Try lock code_ix that is needed for (exlusive) access of the staging area.
+ * Main process lock (only) must be held.
+ * Caller is suspended and *must* yield if 0 is returned. 
  */
-void erts_lock_code_ix(void);
+int erts_try_lock_code_ix(struct process*);
 
 /* Unlock code_ix
- * ToDo: Dequeue and resume waiting processes.
  */
 void erts_unlock_code_ix(void);
 
diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types
index 1b697d604c..de7d5599bb 100644
--- a/erts/emulator/beam/erl_alloc.types
+++ b/erts/emulator/beam/erl_alloc.types
@@ -264,6 +264,7 @@ type	ZLIB		STANDARD	SYSTEM		zlib
 type	CPU_GRPS_MAP	LONG_LIVED	SYSTEM		cpu_groups_map
 type	AUX_WORK_TMO	LONG_LIVED	SYSTEM		aux_work_timeouts
 type	MISC_AUX_WORK_Q	LONG_LIVED	SYSTEM		misc_aux_work_q
+type	CODE_IX_LOCK_Q  SHORT_LIVED	SYSTEM		code_ix_lock_q
 
 +if threads_no_smp
 # Need thread safe allocs, but std_alloc and fix_alloc are not;
diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c
index 772d3a03b2..8c7e1f36ee 100644
--- a/erts/emulator/beam/erl_lock_check.c
+++ b/erts/emulator/beam/erl_lock_check.c
@@ -84,7 +84,6 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"reg_tab",				NULL			},
     {	"migration_info_update",		NULL			},
     {	"proc_main",				"pid"			},
-    {   "sverk_code_ix_lock",   NULL }, /*SVERK FIXME */
     {   "old_code",                             "address"               },
 #ifdef HIPE
     {	"hipe_mfait_lock",			NULL			},
@@ -95,6 +94,7 @@ static erts_lc_lock_order_t erts_lock_order[] = {
     {	"proc_msgq",				"pid"			},
     {	"dist_entry",				"address"		},
     {	"dist_entry_links",			"address"		},
+    {   "code_ix_queue",                        NULL                    },
     {	"proc_status",				"pid"			},
     {	"proc_tab",				NULL			},
     {   "ports_snapshot",                       NULL                    },