erts: Implement tracer modules

Add the possibility to use modules as trace data receivers. The functions
in the module have to be nifs as otherwise complex trace probes will be
very hard to handle (complex means trace probes for ports for example).

This commit changes the way that the ptab->tracer field works from always
being an immediate, to now be NIL if no tracer is present or else be
the tuple {TracerModule, TracerState} where TracerModule is an atom that
is later used to lookup the appropriate tracer callbacks to call and
TracerState is just passed to the tracer callback. The default process and
port tracers have been rewritten to use the new API.

This commit also changes the order which trace messages are delivered to the
potential tracer process. Any enif_send done in a tracer module may be delayed
indefinitely because of lock order issues. If a message is delayed any other
trace message send from that process is also delayed so that order is preserved
for each traced entity. This means that for some trace events (i.e. send/receive)
the events may come in an unintuitive order (receive before send) to the
trace receiver. Timestamps are taken when the trace message is generated so
trace messages from differented processes may arrive with the timestamp
out of order.

Both the erlang:trace and seq_trace:set_system_tracer accept the new tracer
module tracers and also the backwards compatible arguments.

OTP-10267

1 files changed, 204 insertions, 48 deletions

diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index 61dba4ef82..f35c08450c 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -130,17 +130,7 @@ void erts_pre_nif(ErlNifEnv* env, Process* p, struct erl_module_nif* mod_nif)
     env->fpe_was_unmasked = erts_block_fpe();
     env->tmp_obj_list = NULL;
     env->exception_thrown = 0;
-}
-
-static void pre_nif_noproc(ErlNifEnv* env, struct erl_module_nif* mod_nif)
-{
-    env->mod_nif = mod_nif;
-    env->proc = NULL;
-    env->hp = NULL;
-    env->hp_end = NULL;
-    env->heap_frag = NULL;
-    env->fpe_was_unmasked = erts_block_fpe();
-    env->tmp_obj_list = NULL;
+    env->tracee = NULL;
 }
 
 /* Temporary object header, auto-deallocated when NIF returns
@@ -180,13 +170,6 @@ void erts_post_nif(ErlNifEnv* env)
     free_tmp_objs(env);
 }
 
-static void post_nif_noproc(ErlNifEnv* env)
-{
-    erts_unblock_fpe(env->fpe_was_unmasked);
-    free_tmp_objs(env);
-}
-
-
 /* Flush out our cached heap pointers to allow an ordinary HAlloc
 */
 static void flush_env(ErlNifEnv* env)
@@ -247,18 +230,20 @@ struct enif_msg_environment_t
     Process phony_proc;
 };
 
-ErlNifEnv* enif_alloc_env(void)
+static ERTS_INLINE void
+setup_nif_env(struct enif_msg_environment_t* msg_env,
+              struct erl_module_nif* mod)
 {
-    struct enif_msg_environment_t* msg_env =
-	erts_alloc_fnf(ERTS_ALC_T_NIF, sizeof(struct enif_msg_environment_t));
     Eterm* phony_heap = (Eterm*) msg_env; /* dummy non-NULL ptr */
-	
-    msg_env->env.hp = phony_heap; 
+
+    msg_env->env.hp = phony_heap;
     msg_env->env.hp_end = phony_heap;
     msg_env->env.heap_frag = NULL;
-    msg_env->env.mod_nif = NULL;
+    msg_env->env.mod_nif = mod;
     msg_env->env.tmp_obj_list = NULL;
+    msg_env->env.fpe_was_unmasked = erts_block_fpe();
     msg_env->env.proc = &msg_env->phony_proc;
+    msg_env->env.exception_thrown = 0;
     memset(&msg_env->phony_proc, 0, sizeof(Process));
     HEAP_START(&msg_env->phony_proc) = phony_heap;
     HEAP_TOP(&msg_env->phony_proc) = phony_heap;
@@ -270,6 +255,13 @@ ErlNifEnv* enif_alloc_env(void)
     msg_env->phony_proc.space_verified = 0;
     msg_env->phony_proc.space_verified_from = NULL;
 #endif
+}
+
+ErlNifEnv* enif_alloc_env(void)
+{
+    struct enif_msg_environment_t* msg_env =
+	erts_alloc_fnf(ERTS_ALC_T_NIF, sizeof(struct enif_msg_environment_t));
+    setup_nif_env(msg_env, NULL);
     return &msg_env->env;
 }
 void enif_free_env(ErlNifEnv* env)
@@ -302,24 +294,114 @@ void enif_clear_env(ErlNifEnv* env)
     menv->env.hp = menv->env.hp_end = HEAP_TOP(p);
     
     ASSERT(!is_offheap(&MSO(p)));
+    erts_unblock_fpe(env->fpe_was_unmasked);
     free_tmp_objs(env);
 }
-int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
-	      ErlNifEnv* msg_env, ERL_NIF_TERM msg)
+
+#ifdef ERTS_SMP
+#ifdef DEBUG
+static int enif_send_delay = 0;
+#define ERTS_FORCE_ENIF_SEND_DELAY() (enif_send_delay++ % 2 == 0)
+#else
+#ifdef ERTS_PROC_LOCK_OWN_IMPL
+#define ERTS_FORCE_ENIF_SEND_DELAY() 0
+#else
+/*
+ * We always schedule messages if we do not use our own
+ * process lock implementation, as if we try to do a trylock on
+ * a lock that might already be locked by the same thread.
+ * And what happens then with different mutex implementations
+ * is not always guaranteed.
+ */
+#define ERTS_FORCE_ENIF_SEND_DELAY() 1
+#endif
+#endif
+
+int erts_flush_trace_messages(Process *c_p, ErtsProcLocks c_p_locks)
+{
+    ErlTraceMessageQueue *msgq, **last_msgq;
+    int reds = 0;
+
+    erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_TRACE);
+
+    msgq = c_p->trace_msg_q;
+
+    if (!msgq)
+        goto error;
+
+    do {
+        Process* rp;
+        ErtsProcLocks rp_locks;
+        ErtsMessage *first, **last;
+        Uint len;
+
+        first = msgq->first;
+        last = msgq->last;
+        len = msgq->len;
+        msgq->first = NULL;
+        msgq->last = &msgq->first;
+        msgq->len = 0;
+        erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_TRACE);
+
+        ASSERT(len != 0);
+
+        rp = erts_proc_lookup(msgq->receiver);
+        if (rp) {
+            rp_locks = 0;
+            if (rp->common.id == c_p->common.id)
+                rp_locks = c_p_locks;
+            erts_queue_messages(rp, &rp_locks, first, last, len);
+            if (rp->common.id == c_p->common.id)
+                rp_locks &= ~c_p_locks;
+            if (rp_locks)
+                erts_smp_proc_unlock(rp, rp_locks);
+            reds += len;
+        } else {
+            erts_cleanup_messages(first);
+        }
+        reds += 1;
+        erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_TRACE);
+        msgq = msgq->next;
+    } while (msgq);
+
+    last_msgq = &c_p->trace_msg_q;
+
+    while (*last_msgq) {
+        msgq = *last_msgq;
+        if (msgq->len == 0) {
+            *last_msgq = msgq->next;
+            erts_free(ERTS_ALC_T_TRACE_MSG_QUEUE, msgq);
+        } else {
+            last_msgq = &msgq->next;
+        }
+    }
+
+error:
+    erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_TRACE);
+
+    return reds;
+}
+
+#endif
+
+int
+enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
+          ErlNifEnv* msg_env, ERL_NIF_TERM msg)
 {
     struct enif_msg_environment_t* menv = (struct enif_msg_environment_t*)msg_env;
-    ErtsProcLocks rp_locks = 0;
+    ErtsProcLocks rp_locks = 0, lc_locks = 0, c_p_locks = ERTS_PROC_LOCK_MAIN;
     Process* rp;
     Process* c_p;
     ErtsMessage *mp;
     Eterm receiver = to_pid->pid;
     int flush_me = 0;
-    int scheduler = erts_get_scheduler_id() != 0;
+    ErtsSchedulerData *esdp = erts_get_scheduler_data();
+    int scheduler = esdp ? esdp->no : 0;
 
     if (env != NULL) {
 	c_p = env->proc;
 	if (receiver == c_p->common.id) {
-	    rp_locks = ERTS_PROC_LOCK_MAIN;
+	    rp_locks = c_p_locks;
 	    flush_me = 1;
 	}
     }
@@ -329,7 +411,7 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
 #else
 	erts_exit(ERTS_ABORT_EXIT,"enif_send: env==NULL on non-SMP VM");
 #endif
-    }    
+    }
 
     rp = (scheduler
 	  ? erts_proc_lookup(receiver)
@@ -353,7 +435,6 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
             menv->env.heap_frag = NULL;
             MBUF(&menv->phony_proc) = NULL;
         }
-        ASSERT(!is_offheap(&MSO(&menv->phony_proc)));
     } else {
         Uint sz = size_object(msg);
         Eterm *hp;
@@ -362,14 +443,83 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
         ASSERT(hp == mp->hfrag.mem+mp->hfrag.used_size);
     }
 
-    if (flush_me) {	
-	flush_env(env); /* Needed for ERTS_HOLE_CHECK */ 
+    ERL_MESSAGE_TERM(mp) = msg;
+
+    if (flush_me) {
+	flush_env(env); /* Needed for ERTS_HOLE_CHECK */
+    }
+
+    if (!env || !env->tracee) {
+        erts_queue_message(rp, &rp_locks, mp, msg);
+    }
+#ifdef ERTS_SMP
+    else {
+        /* This clause is taken when the nif is called in the context
+           of a traced process. We do not know which locks we have
+           so we have to do a try lock and if that fails we en queue
+           the message in a special trace message output queue of the
+           tracee */
+        ErlTraceMessageQueue *msgq;
+        Process *t_p = env->tracee;
+
+
+        erts_smp_proc_lock(t_p, ERTS_PROC_LOCK_TRACE);
+
+        msgq = t_p->trace_msg_q;
+
+        while (msgq != NULL) {
+            if (msgq->receiver == receiver) {
+                break;
+            }
+            msgq = msgq->next;
+        }
+
+#ifdef ERTS_ENABLE_LOCK_CHECK
+        lc_locks = erts_proc_lc_my_proc_locks(rp);
+        rp_locks |= lc_locks;
+        if (receiver == c_p->common.id)
+            c_p_locks |= lc_locks;
+#endif
+        if (ERTS_FORCE_ENIF_SEND_DELAY() || msgq ||
+            rp_locks & ERTS_PROC_LOCK_MSGQ ||
+            erts_smp_proc_trylock(rp, ERTS_PROC_LOCK_MSGQ) == EBUSY) {
+
+            if (!msgq) {
+
+                msgq = erts_alloc(ERTS_ALC_T_TRACE_MSG_QUEUE,
+                                  sizeof(ErlTraceMessageQueue));
+                msgq->receiver = receiver;
+                msgq->first = mp;
+                msgq->last = &mp->next;
+                msgq->len = 1;
+
+                /* Insert in linked list */
+                msgq->next = t_p->trace_msg_q;
+                t_p->trace_msg_q = msgq;
+
+                erts_smp_proc_unlock(t_p, ERTS_PROC_LOCK_TRACE);
+
+                erts_schedule_flush_trace_messages(t_p->common.id);
+
+            } else {
+                msgq->len++;
+                *msgq->last = mp;
+                msgq->last = &mp->next;
+                erts_smp_proc_unlock(t_p, ERTS_PROC_LOCK_TRACE);
+            }
+        } else {
+            erts_smp_proc_unlock(t_p, ERTS_PROC_LOCK_TRACE);
+            rp_locks &= ~ERTS_PROC_LOCK_TRACE;
+            rp_locks |= ERTS_PROC_LOCK_MSGQ;
+            erts_queue_message(rp, &rp_locks, mp, msg);
+        }
     }
-    erts_queue_message(rp, &rp_locks, mp, msg, am_undefined);
+#endif
+
     if (c_p == rp)
 	rp_locks &= ~ERTS_PROC_LOCK_MAIN;
-    if (rp_locks)
-	erts_smp_proc_unlock(rp, rp_locks);
+    if (rp_locks & ~lc_locks)
+	erts_smp_proc_unlock(rp, rp_locks & ~lc_locks);
     if (!scheduler)
 	erts_proc_dec_refc(rp);
     if (flush_me) {
@@ -1485,10 +1635,10 @@ static void close_lib(struct erl_module_nif* lib)
     ASSERT(erts_refc_read(&lib->rt_dtor_cnt,0) == 0);
 
     if (lib->entry != NULL && lib->entry->unload != NULL) {
-	ErlNifEnv env;
-	pre_nif_noproc(&env, lib);
-	lib->entry->unload(&env, lib->priv_data);
-	post_nif_noproc(&env);
+	struct enif_msg_environment_t msg_env;
+        setup_nif_env(&msg_env, lib);
+	lib->entry->unload(&msg_env.env, lib->priv_data);
+        enif_clear_env(&msg_env.env);
     }
     if (!erts_is_static_nif(lib->handle))
       erts_sys_ddll_close(lib->handle);
@@ -1634,10 +1784,10 @@ static void nif_resource_dtor(Binary* bin)
     ASSERT(ERTS_MAGIC_BIN_DESTRUCTOR(bin) == &nif_resource_dtor);
 
     if (type->dtor != NULL) {
-	ErlNifEnv env;
-	pre_nif_noproc(&env, type->owner);
-	type->dtor(&env,resource->data);
-	post_nif_noproc(&env);
+        struct enif_msg_environment_t msg_env;
+        setup_nif_env(&msg_env, type->owner);
+	type->dtor(&msg_env.env, resource->data);
+        enif_clear_env(&msg_env.env);
     }
     if (erts_refc_dectest(&type->refc, 0) == 0) {
 	ASSERT(type->next == NULL);
@@ -2916,6 +3066,9 @@ erts_unload_nif(struct erl_module_nif* lib)
     ASSERT(erts_smp_thr_progress_is_blocking());
     ASSERT(lib != NULL);
     ASSERT(lib->mod != NULL);
+
+    erts_tracer_nif_clear();
+
     for (rt = resource_type_list.next;
 	 rt != &resource_type_list;
 	 rt = next) {
@@ -2958,6 +3111,7 @@ void erl_nif_init()
     resource_type_list.owner = NULL;
     resource_type_list.module = THE_NON_VALUE;
     resource_type_list.name = THE_NON_VALUE;
+
 }
 
 int erts_nif_get_funcs(struct erl_module_nif* mod,
@@ -2967,7 +3121,8 @@ int erts_nif_get_funcs(struct erl_module_nif* mod,
     return mod->entry->num_of_funcs;
 }
 
-Eterm erts_nif_call_function(Process *p, struct erl_module_nif* mod,
+Eterm erts_nif_call_function(Process *p, Process *tracee,
+                             struct erl_module_nif* mod,
                              ErlNifFunc *fun, int argc, Eterm *argv)
 {
     Eterm nif_result;
@@ -2979,13 +3134,14 @@ Eterm erts_nif_call_function(Process *p, struct erl_module_nif* mod,
             break;
     ASSERT(i < mod->entry->num_of_funcs);
     if (p)
-        ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(p) & ERTS_PROC_LOCK_MAIN);
+        ERTS_SMP_LC_ASSERT(erts_proc_lc_my_proc_locks(p) & ERTS_PROC_LOCK_MAIN
+                           || erts_smp_thr_progress_is_blocking());
 #endif
     if (p) {
         struct enif_environment_t env;
         ASSERT(is_internal_pid(p->common.id));
         erts_pre_nif(&env, p, mod);
-        env.may_delay_enif_send = 1;
+        env.tracee = tracee;
         nif_result = (*fun->fptr)(&env, argc, argv);
         if (env.exception_thrown)
             nif_result = THE_NON_VALUE;
@@ -2995,7 +3151,7 @@ Eterm erts_nif_call_function(Process *p, struct erl_module_nif* mod,
            so we create a phony one. */
         struct enif_msg_environment_t msg_env;
         setup_nif_env(&msg_env, mod);
-        msg_env.env.may_delay_enif_send = 1;
+        msg_env.env.tracee = tracee;
         nif_result = (*fun->fptr)(&msg_env.env, argc, argv);
         if (msg_env.env.exception_thrown)
             nif_result = THE_NON_VALUE;