erts: Fix seq_trace to not clear token for system messages

A lot of erts internal messages used behind APIs to create
non-blocking calls, e.g. port_command, would cause the seq_trace
token to be cleared from the caller when it should not.
This commit fixes that and adds asserts that makes sure
that all messages sent have to correct token set.

Fixes: ERL-602

11 files changed, 87 insertions, 43 deletions

diff --git a/erts/emulator/beam/beam_bif_load.c b/erts/emulator/beam/beam_bif_load.c
index a0dbd9ec7b..d221e6aea6 100644
--- a/erts/emulator/beam/beam_bif_load.c
+++ b/erts/emulator/beam/beam_bif_load.c
@@ -1753,6 +1753,7 @@ BIF_RETTYPE erts_internal_purge_module_2(BIF_ALIST_2)
 
 	if (literals) {
 	    ErtsLiteralAreaRef *ref;
+            ErtsMessage *mp;
 	    ref = erts_alloc(ERTS_ALC_T_LITERAL_REF,
 			     sizeof(ErtsLiteralAreaRef));
 	    ref->literal_area = literals;
@@ -1767,10 +1768,12 @@ BIF_RETTYPE erts_internal_purge_module_2(BIF_ALIST_2)
 		release_literal_areas.last = ref;
 	    }
 	    erts_mtx_unlock(&release_literal_areas.mtx);
+            mp = erts_alloc_message(0, NULL);
+            ERL_MESSAGE_TOKEN(mp) = am_undefined;
 	    erts_queue_proc_message(BIF_P,
                                erts_literal_area_collector,
 			       0,
-			       erts_alloc_message(0, NULL),
+			       mp,
 			       am_copy_literals);
 	}
 
diff --git a/erts/emulator/beam/beam_debug.c b/erts/emulator/beam/beam_debug.c
index 10d20c8f3f..9633de2021 100644
--- a/erts/emulator/beam/beam_debug.c
+++ b/erts/emulator/beam/beam_debug.c
@@ -1191,6 +1191,7 @@ dirty_send_message(Process *c_p, Eterm to, Eterm tag)
     mp = erts_alloc_message_heap(rp, &rp_locks, 3, &hp, &ohp);
 
     msg = TUPLE2(hp, tag, c_p->common.id);
+    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     erts_queue_proc_message(c_p, rp, rp_locks, mp, msg);
 
     if (rp == real_c_p)
diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index 97e1ee1286..f18af8bcd7 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -2063,7 +2063,7 @@ do_send(Process *p, Eterm to, Eterm msg, Eterm *refp, ErtsSendContext *ctx)
 	if (p == rp)
 	    rp_locks |= ERTS_PROC_LOCK_MAIN;
 	/* send to local process */
-	erts_send_message(p, rp, &rp_locks, msg, 0);
+	erts_send_message(p, rp, &rp_locks, msg);
 	erts_proc_unlock(rp,
 			     p == rp
 			     ? (rp_locks & ~ERTS_PROC_LOCK_MAIN)
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 61090e2c47..146c00b07d 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -3672,6 +3672,7 @@ int erts_auto_connect(DistEntry* dep, Process *proc, ErtsProcLocks proc_locks)
         dhandle = erts_build_dhandle(&hp, ohp, dep);
         msg = TUPLE4(hp, am_auto_connect, dep->sysname, make_small(conn_id),
                      dhandle);
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
         erts_queue_proc_message(proc, net_kernel, nk_locks, mp, msg);
         erts_proc_unlock(net_kernel, nk_locks);
     }
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 36d83d93f4..c009a3bde8 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -3647,6 +3647,7 @@ send_ets_transfer_message(Process *c_p, Process *proc,
         hd_copy = copy_struct(heir_data, hd_sz, &hp, ohp);
     sender = c_p->common.id;
     msg = TUPLE4(hp, am_ETS_TRANSFER, tid, sender, hd_copy);
+    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     erts_queue_proc_message(c_p, proc, *locks, mp, msg);
 }
 
diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c
index 507cc989d2..a3274d7443 100644
--- a/erts/emulator/beam/erl_message.c
+++ b/erts/emulator/beam/erl_message.c
@@ -409,6 +409,11 @@ ErtsMessage* prepend_pending_sig_maybe(Process* sender, Process* receiver,
  *
  * @brief Send one message from *NOT* a local process.
  *
+ * seq_trace does not work with this type of messages
+ * to it is set to am_undefined which means that the
+ * receiving process will not remove the seq_trace token
+ * when it gets this message.
+ *
  */
 void
 erts_queue_message(Process* receiver, ErtsProcLocks receiver_locks,
@@ -417,11 +422,19 @@ erts_queue_message(Process* receiver, ErtsProcLocks receiver_locks,
     ASSERT(is_not_internal_pid(from));
     ERL_MESSAGE_TERM(mp) = msg;
     ERL_MESSAGE_FROM(mp) = from;
+    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     queue_messages(receiver, receiver_locks, mp, &mp->next, 1);
 }
 
 /**
  * @brief Send one message from a local process.
+ *
+ * It is up to the caller of this function to set the
+ * correct seq_trace. The general rule of thumb is that
+ * it should be set to am_undefined if the message
+ * cannot be traced using seq_trace, if it can be
+ * traced it should be set to the trace token. It should
+ * very rarely be explicitly set to NIL!
  */
 void
 erts_queue_proc_message(Process* sender,
@@ -584,8 +597,7 @@ void
 erts_send_message(Process* sender,
 		  Process* receiver,
 		  ErtsProcLocks *receiver_locks,
-		  Eterm message,
-		  unsigned flags)
+		  Eterm message)
 {
     Uint msize;
     ErtsMessage* mp;
@@ -619,7 +631,7 @@ erts_send_message(Process* sender,
 
     receiver_state = erts_atomic32_read_nob(&receiver->state);
 
-    if (SEQ_TRACE_TOKEN(sender) != NIL && !(flags & ERTS_SND_FLG_NO_SEQ_TRACE)) {
+    if (SEQ_TRACE_TOKEN(sender) != NIL) {
         Eterm* hp;
 	Eterm stoken = SEQ_TRACE_TOKEN(sender);
 	Uint seq_trace_size = 0;
diff --git a/erts/emulator/beam/erl_message.h b/erts/emulator/beam/erl_message.h
index d120111634..b2550814fd 100644
--- a/erts/emulator/beam/erl_message.h
+++ b/erts/emulator/beam/erl_message.h
@@ -405,8 +405,6 @@ typedef struct erl_trace_message_queue__ {
 #define SAVE_MESSAGE(p) \
      (p)->sig_qs.save = &(*(p)->sig_qs.save)->next
 
-#define ERTS_SND_FLG_NO_SEQ_TRACE		(((unsigned) 1) << 0)
-
 #define ERTS_HEAP_FRAG_SIZE(DATA_WORDS) \
    (sizeof(ErlHeapFragment) - sizeof(Eterm) + (DATA_WORDS)*sizeof(Eterm))
 
@@ -429,7 +427,7 @@ typedef struct erl_trace_message_queue__ {
     do {                                                \
         (MP)->next = NULL;                              \
         ERL_MESSAGE_TERM(MP) = THE_NON_VALUE;           \
-        ERL_MESSAGE_TOKEN(MP) = NIL;                    \
+        ERL_MESSAGE_TOKEN(MP) = THE_NON_VALUE;          \
         ERL_MESSAGE_FROM(MP) = NIL;                     \
         ERL_MESSAGE_DT_UTAG_INIT(MP);                   \
         MP->data.attached = NULL;                       \
@@ -446,7 +444,7 @@ void erts_queue_proc_message(Process* from,Process* to, ErtsProcLocks,ErtsMessag
 void erts_queue_proc_messages(Process* from, Process* to, ErtsProcLocks,
                               ErtsMessage*, ErtsMessage**, Uint);
 void erts_deliver_exit_message(Eterm, Process*, ErtsProcLocks *, Eterm, Eterm);
-void erts_send_message(Process*, Process*, ErtsProcLocks*, Eterm, unsigned);
+void erts_send_message(Process*, Process*, ErtsProcLocks*, Eterm);
 void erts_link_mbuf_to_proc(Process *proc, ErlHeapFragment *bp);
 
 Uint erts_msg_attached_data_size_aux(ErtsMessage *msg);
diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index 0fbf0eb03a..26498eda7b 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -744,8 +744,58 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
 	rp_locks = ERTS_PROC_LOCK_MAIN;
 
     if (menv) {
+        Eterm token = c_p ? SEQ_TRACE_TOKEN(c_p) : am_undefined;
+        if (token != NIL && token != am_undefined) {
+            /* This code is copied from erts_send_message */
+            Eterm stoken = SEQ_TRACE_TOKEN(c_p);
+#ifdef USE_VM_PROBES
+            DTRACE_CHARBUF(sender_name, 64);
+            DTRACE_CHARBUF(receiver_name, 64);
+            Sint tok_label = 0;
+            Sint tok_lastcnt = 0;
+            Sint tok_serial = 0;
+            Eterm utag = NIL;
+            *sender_name = *receiver_name = '\0';
+            if (DTRACE_ENABLED(message_send)) {
+                erts_snprintf(sender_name, sizeof(DTRACE_CHARBUF_NAME(sender_name)),
+                              "%T", c_p->common.id);
+                erts_snprintf(receiver_name, sizeof(DTRACE_CHARBUF_NAME(receiver_name)),
+                              "%T", rp->common.id);
+            }
+#endif
+            if (have_seqtrace(stoken)) {
+                seq_trace_update_send(c_p);
+                seq_trace_output(stoken, msg, SEQ_TRACE_SEND,
+                                 rp->common.id, c_p);
+            }
+#ifdef USE_VM_PROBES
+            if (!(DT_UTAG_FLAGS(c_p) & DT_UTAG_SPREADING)) {
+                stoken = NIL;
+            }
+#endif
+            token = enif_make_copy(msg_env, stoken);
+
+#ifdef USE_VM_PROBES
+            if (DT_UTAG_FLAGS(c_p) & DT_UTAG_SPREADING) {
+                if (is_immed(DT_UTAG(c_p)))
+                    utag = DT_UTAG(c_p);
+                else
+                    utag = enif_make_copy(msg_env, DT_UTAG(c_p));
+            }
+            if (DTRACE_ENABLED(message_send)) {
+                if (have_seqtrace(stoken)) {
+                    tok_label = SEQ_TRACE_T_DTRACE_LABEL(stoken);
+                    tok_lastcnt = signed_val(SEQ_TRACE_T_LASTCNT(stoken));
+                    tok_serial = signed_val(SEQ_TRACE_T_SERIAL(stoken));
+                }
+                DTRACE6(message_send, sender_name, receiver_name,
+                        size_object(msg), tok_label, tok_lastcnt, tok_serial);
+            }
+#endif
+        }
         flush_env(msg_env);
         mp = erts_alloc_message(0, NULL);
+        ERL_MESSAGE_TOKEN(mp) = token;
         mp->data.heap_frag = menv->env.heap_frag;
         ASSERT(mp->data.heap_frag == MBUF(&menv->phony_proc));
         if (mp->data.heap_frag != NULL) {
@@ -783,6 +833,7 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
 		ohp = &bp->off_heap;
 	    }
 	}
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
         msg = copy_struct_litopt(msg, sz, &hp, ohp, &litarea);
     }
 
@@ -826,6 +877,7 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
 
             ERL_MESSAGE_TERM(mp) = msg;
             ERL_MESSAGE_FROM(mp) = from;
+            ERL_MESSAGE_TOKEN(mp) = am_undefined;
 
             if (!msgq) {
                 msgq = erts_alloc(ERTS_ALC_T_TRACE_MSG_QUEUE,
diff --git a/erts/emulator/beam/erl_proc_sig_queue.c b/erts/emulator/beam/erl_proc_sig_queue.c
index d6d22677e7..f90501cbf7 100644
--- a/erts/emulator/beam/erl_proc_sig_queue.c
+++ b/erts/emulator/beam/erl_proc_sig_queue.c
@@ -52,8 +52,8 @@
 
 #define ERTS_SIG_Q_OP_MAX 13
 
-#define ERTS_SIG_Q_OP_EXIT                      0
-#define ERTS_SIG_Q_OP_EXIT_LINKED               1
+#define ERTS_SIG_Q_OP_EXIT                      0  /* Exit signal due to bif call */
+#define ERTS_SIG_Q_OP_EXIT_LINKED               1  /* Exit signal due to link break*/
 #define ERTS_SIG_Q_OP_MONITOR_DOWN              2
 #define ERTS_SIG_Q_OP_MONITOR                   3
 #define ERTS_SIG_Q_OP_DEMONITOR                 4
@@ -1167,10 +1167,7 @@ erts_proc_sig_send_persistent_monitor_msg(Uint16 type, Eterm key,
     ERL_MESSAGE_TERM(mp) = ERTS_PROC_SIG_MAKE_TAG(ERTS_SIG_Q_OP_PERSISTENT_MON_MSG,
                                                   type, 0);
     ERL_MESSAGE_FROM(mp) = from;
-    ERL_MESSAGE_TOKEN(mp) = NIL;
-#ifdef USE_VM_PROBES
-    ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
+    ERL_MESSAGE_TOKEN(mp) = am_undefined;
 
     if (!proc_queue_signal(NULL, to, (ErtsSignal *) mp,
                            ERTS_SIG_Q_OP_PERSISTENT_MON_MSG)) {
@@ -1564,11 +1561,6 @@ erts_proc_sig_send_is_alive_request(Process *c_p, Eterm to, Eterm ref)
     ERL_MESSAGE_TERM(mp) = ERTS_PROC_SIG_MAKE_TAG(ERTS_SIG_Q_OP_IS_ALIVE,
                                                   ERTS_SIG_Q_TYPE_UNDEFINED,
                                                   0);
-    ERL_MESSAGE_TOKEN(mp) = NIL;
-    ERL_MESSAGE_FROM(mp) = am_system;
-#ifdef USE_VM_PROBES
-    ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
 
     if (proc_queue_signal(c_p, to, (ErtsSignal *) mp, ERTS_SIG_Q_OP_IS_ALIVE))
         (void) maybe_elevate_sig_handling_prio(c_p, to);
@@ -1672,11 +1664,6 @@ erts_proc_sig_send_sync_suspend(Process *c_p, Eterm to, Eterm tag, Eterm reply)
     ERL_MESSAGE_TERM(mp) = ERTS_PROC_SIG_MAKE_TAG(ERTS_SIG_Q_OP_SYNC_SUSPEND,
                                                   ERTS_SIG_Q_TYPE_UNDEFINED,
                                                   0);
-    ERL_MESSAGE_TOKEN(mp) = NIL;
-    ERL_MESSAGE_FROM(mp) = am_system;
-#ifdef USE_VM_PROBES
-    ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
 
     if (proc_queue_signal(c_p, to, (ErtsSignal *) mp, ERTS_SIG_Q_OP_SYNC_SUSPEND))
         (void) maybe_elevate_sig_handling_prio(c_p, to);
@@ -1790,7 +1777,7 @@ handle_rpc(Process *c_p, ErtsProcSigRPC *rpc, int cnt, int limit, int *yieldp)
         msg = TUPLE2(hp, ref, res);
 
         mp->hfrag.next = bp;
-
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
         erts_queue_proc_message(c_p, rp, 0, mp, msg);
     }
 
@@ -2155,10 +2142,7 @@ handle_exit_signal(Process *c_p, ErtsSigRecvTracing *tracing,
                 pid = STORE_NC(&hp, ohp, from);
 
                 ERL_MESSAGE_TERM(mp) = TUPLE3(hp, am_EXIT, pid, reason);
-                ERL_MESSAGE_TOKEN(mp) = NIL;
-#ifdef USE_VM_PROBES
-                ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
+                ERL_MESSAGE_TOKEN(mp) = am_undefined;
                 if (is_immed(pid))
                     ERL_MESSAGE_FROM(mp) = pid;
                 else {
@@ -2346,10 +2330,7 @@ convert_to_down_message(Process *c_p,
                                   type, from, reason);
     hp += 6;
 
-    ERL_MESSAGE_TOKEN(mp) = NIL;
-#ifdef USE_VM_PROBES
-    ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
+    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     /* Replace original signal with the exit message... */
     convert_to_msg(c_p, sig, mp, next_nm_sig);
 
@@ -2397,10 +2378,7 @@ convert_to_nodedown_messages(Process *c_p,
 
             ERL_MESSAGE_TERM(mp) = TUPLE2(hp, am_nodedown, node);
             ERL_MESSAGE_FROM(mp) = am_system;
-            ERL_MESSAGE_TOKEN(mp) = NIL;
-#ifdef USE_VM_PROBES
-            ERL_MESSAGE_DT_UTAG(mp) = NIL;
-#endif
+            ERL_MESSAGE_TOKEN(mp) = am_undefined;
             mp->next = nd_first;
             nd_first = mp;
             if (!nd_last)
@@ -2830,6 +2808,7 @@ handle_process_info(Process *c_p, ErtsSigRecvTracing *tracing,
         if (is_alive)
             erts_factory_trim_and_close(&hfact, &msg, 1);
 
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
         erts_queue_proc_message(c_p, rp, locks, mp, msg);
 
         if (!is_alive && locks)
@@ -2931,6 +2910,7 @@ sync_suspend_reply(Process *c_p, ErtsMessage *mp, erts_aint32_t state)
                 tp[2] = ssusp->async ? am_not_suspended : am_internal_error;
             }
         }
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
         erts_queue_proc_message(c_p, rp, 0, mp, ssusp->message);
     }
 }
diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c
index 8784eb5a63..9386f79b56 100644
--- a/erts/emulator/beam/erl_process.c
+++ b/erts/emulator/beam/erl_process.c
@@ -9874,6 +9874,7 @@ notify_sys_task_executed(Process *c_p, ErtsProcSysTask *st,
 	ASSERT(hp_start + hsz == hp);
 #endif
 
+        ERL_MESSAGE_TOKEN(mp) = am_undefined;
 	erts_queue_proc_message(c_p, rp, rp_locks, mp, msg);
 
 	if (c_p == rp)
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 2446b3c074..133ab485d9 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -3287,7 +3287,6 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to,
     if (trace_send)
         trace_port_send(prt, to, tuple, 1);
 
-    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id);
     if (rp_locks)
 	erts_proc_unlock(rp, rp_locks);
@@ -3459,7 +3458,6 @@ deliver_vec_message(Port* prt,			/* Port */
     if (IS_TRACED_FL(prt, F_TRACE_SEND))
         trace_port_send(prt, to, tuple, 1);
 
-    ERL_MESSAGE_TOKEN(mp) = am_undefined;
     erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id);
     erts_proc_unlock(rp, rp_locks);
     if (!scheduler)
@@ -5382,7 +5380,6 @@ void driver_report_exit(ErlDrvPort ix, int status)
     if (IS_TRACED_FL(prt, F_TRACE_SEND))
         trace_port_send(prt, pid, tuple, 1);
 
-   ERL_MESSAGE_TOKEN(mp) = am_undefined;
    erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id);
 
    erts_proc_unlock(rp, rp_locks);
@@ -5988,8 +5985,6 @@ driver_deliver_term(Port *prt, Eterm to, ErlDrvTermData* data, int len)
 	    from = prt->common.id;
 	}
 
-	/* send message */
-        ERL_MESSAGE_TOKEN(factory.message) = am_undefined;
 	erts_queue_message(rp, rp_locks, factory.message, mess, from);
     }
     else if (res == -2) {