Merge pull request #9590 from rickard-green/rickard/prio-msg-fix/OTP-19198

Priority message fixes

Author: rickard-green

erts/emulator/beam/erl_proc_sig_queue.c

@@ -303,11 +303,11 @@ static void proc_sig_queue_unlock_buffer(ErtsSignalInQueueBuffer* slot);
 static void handle_message_enqueued_tracing(Process *c_p,
                                             ErtsSigRecvTracing *tracing,
                                             ErtsMessage *msg);
-static void
+static int
 insert_prepared_prio_msg(Process *c_p, ErtsSigRecvTracing *tracing,
                          ErtsMessage *sig, Eterm message, Eterm token,
                          ErtsMessage ***next_nm_sig);
-static void
+static int
 insert_prepared_prio_msg_attached(Process *c_p, ErtsSigRecvTracing *tracing,
                                   ErtsMessage *sig, void *attached,
                                   Eterm message, Eterm token,
@@ -4703,9 +4703,9 @@ handle_exit_signal(Process *c_p, ErtsSigRecvTracing *tracing,
                            : ((ErtsSeqTokenExitSignalData *) xsigd)->token);
             if (prio) {
                 remove_nm_sig(c_p, sig, next_nm_sig);
-                insert_prepared_prio_msg(c_p, tracing, sig,
-                                         xsigd->message, token,
-                                         next_nm_sig);
+                cnt += insert_prepared_prio_msg(c_p, tracing, sig,
+                                                xsigd->message, token,
+                                                next_nm_sig);
             }
             else {
                 convert_prepared_sig_to_msg(c_p, tracing, sig,
@@ -4778,17 +4778,19 @@ convert_prepared_down_message(Process *c_p, ErtsSigRecvTracing *tracing,
                               int prio, ErtsMessage *sig, Eterm msg,
                               ErtsMessage ***next_nm_sig)
 {
+    int cnt = 0;
     if (prio) {
         remove_nm_sig(c_p, sig, next_nm_sig);
-        insert_prepared_prio_msg(c_p, tracing, sig, msg, am_undefined,
-                                 next_nm_sig);
+        cnt += insert_prepared_prio_msg(c_p, tracing, sig, msg, am_undefined,
+                                        next_nm_sig);
     }
     else {
         convert_prepared_sig_to_msg(c_p, tracing, sig, msg, am_undefined,
                                     next_nm_sig);
+        cnt++;
     }
     erts_proc_notify_new_message(c_p, ERTS_PROC_LOCK_MAIN);
-    return 1;
+    return cnt;
 }
 
 static int
@@ -4998,11 +5000,10 @@ convert_to_down_message(Process *c_p,
         *priop = !!((*omon)->flags & ERTS_ML_FLG_PRIO_ML);
         if (*priop) {
             remove_nm_sig(c_p, sig, next_nm_sig);
-            insert_prepared_prio_msg_attached(c_p, tracing, mp,
-                                              mp->data.attached, message,
-                                              am_undefined,
-                                              next_nm_sig);
-            cnt += 4;
+            cnt += insert_prepared_prio_msg_attached(c_p, tracing, mp,
+                                                     mp->data.attached, message,
+                                                     am_undefined,
+                                                     next_nm_sig);
             goto notify_new_message;
         }
 
@@ -6061,9 +6062,9 @@ handle_altact_msg(Process *c_p, ErtsSigRecvTracing *tracing,
                                                  data_attached, next_nm_sig);
         }
         else {
-            insert_prepared_prio_msg_attached(c_p, tracing, sig,
-                                              data_attached, msg, token,
-                                              next_nm_sig);
+            cnt += insert_prepared_prio_msg_attached(c_p, tracing, sig,
+                                                     data_attached, msg, token,
+                                                     next_nm_sig);
         }
         cnt++;
         break;
@@ -6128,8 +6129,8 @@ handle_altact_msg(Process *c_p, ErtsSigRecvTracing *tracing,
                 /* drop faulty encoded external message... */
                 return cnt;
             }
-            insert_prepared_prio_msg(c_p, tracing, mp, ERL_MESSAGE_TERM(mp),
-                                     token, next_nm_sig);
+            cnt += insert_prepared_prio_msg(c_p, tracing, mp, ERL_MESSAGE_TERM(mp),
+                                            token, next_nm_sig);
         }
         break;
     }
@@ -9171,13 +9172,13 @@ prio_queue_check_recv_marks(Eterm obj, Eterm *ref,
     }
 }
 
-static void
+static int
 insert_prepared_prio_msg_attached(Process *c_p, ErtsSigRecvTracing *tracing,
                                   ErtsMessage *sig, void *data_attached,
                                   Eterm message, Eterm token,
                                   ErtsMessage ***next_nm_sig)
 {
-    int i, empty_prio_q = !(c_p->sig_qs.flags & FS_PRIO_MQ_END_MARK);
+    int i, cnt = 1, empty_prio_q = !(c_p->sig_qs.flags & FS_PRIO_MQ_END_MARK);
     ErtsRecvMarkerBlock *blk = c_p->sig_qs.recv_mrk_blk;
     ErtsPrioQInfo *pq_info = get_prio_queue_info(c_p);
     ErtsRecvMarker *pq_end = &pq_info->marker[ERTS_PRIO_Q_MARK_END];
@@ -9225,6 +9226,7 @@ insert_prepared_prio_msg_attached(Process *c_p, ErtsSigRecvTracing *tracing,
 
                 /* Note that this ref has been seen in the prio queue... */
                 mark->in_prioq = !0;
+                cnt += 4;
             }
         }
     }
@@ -9333,33 +9335,46 @@ insert_prepared_prio_msg_attached(Process *c_p, ErtsSigRecvTracing *tracing,
         }
     }
 
-    /* Append the actual prio message to the prio queue... */
     sig->data.attached = data_attached;
 
     ERL_MESSAGE_TERM(sig) = message;
     ERL_MESSAGE_TOKEN(sig) = token;
 
+    if (tracing->messages.active) {
+        if (ERTS_SIG_IS_EXTERNAL_MSG(sig)) {
+            cnt += 50; /* Decode is expensive... */
+            if (!erts_proc_sig_decode_dist(c_p, ERTS_PROC_LOCK_MAIN,
+                                           sig, 0)) {
+                /* Bad dist message; drop it... */
+                sig->next = NULL;
+                erts_cleanup_messages(sig);
+                return cnt;
+            }
+        }
+        handle_message_enqueued_tracing(c_p, tracing, sig);
+    }
+
+    /* Append the actual prio message to the prio queue... */
+
     *pq_end->prev_next = sig;
     pq_end->prev_next = &sig->next;
     sig->next = (ErtsMessage *) pq_end;
 
     c_p->sig_qs.mq_len++;
     erts_chk_sys_mon_long_msgq_on(c_p);
 
-    if (tracing->messages.active)
-        handle_message_enqueued_tracing(c_p, tracing, sig);
-
+    return cnt;
 }
 
-static void
+static int
 insert_prepared_prio_msg(Process *c_p, ErtsSigRecvTracing *tracing,
                          ErtsMessage *sig, Eterm message, Eterm token,
                          ErtsMessage ***next_nm_sig)
 {
-    insert_prepared_prio_msg_attached(c_p, tracing, sig,
-                                      ERTS_MSG_COMBINED_HFRAG,
-                                      message, token,
-                                      next_nm_sig);
+    return insert_prepared_prio_msg_attached(c_p, tracing, sig,
+                                             ERTS_MSG_COMBINED_HFRAG,
+                                             message, token,
+                                             next_nm_sig);
 }
 
 static void

erts/emulator/test/signal_SUITE.erl

@@ -2169,23 +2169,25 @@ priority_messages_old_nodes_test(Config, Node) ->
     exit(Alias, prio_exit_1, [priority]),
     wait_until(fun () -> lists:member(Node, nodes()) end),
     exit(Alias, prio_exit_2, [priority]),
+    Pid ! msg_2,
+    wait_until(fun () -> msg_received(Pid, msg_2) end),
 
     node_disconnect(Node),
     exit(Pid, prio_exit_3, [priority]),
     wait_until(fun () -> msg_received(Pid, {'EXIT', Self, prio_exit_3}) end),
     exit(Pid, prio_exit_4, [priority]),
     wait_until(fun () -> msg_received(Pid, {'EXIT', Self, prio_exit_4}) end),
-                         case PrioMsgSupport of
+    case PrioMsgSupport of
         true ->
             {messages, [prio_msg_1, prio_msg_2, {'EXIT', Self, prio_exit_1},
                         {'EXIT', Self, prio_exit_2}, msg_1, prio_msg_3,
-                        prio_msg_4, prio_msg_5, prio_msg_6, {'EXIT', Self, prio_exit_3},
-                        {'EXIT', Self, prio_exit_4}]}
+                        prio_msg_4, prio_msg_5, prio_msg_6, msg_2,
+                        {'EXIT', Self, prio_exit_3}, {'EXIT', Self, prio_exit_4}]}
                 = proc_info(Pid, messages);
         false ->
             {messages, [msg_1, prio_msg_1, prio_msg_2, prio_msg_3, prio_msg_4,
-                        prio_msg_5, prio_msg_6, {'EXIT', Self, prio_exit_3},
-                        {'EXIT', Self, prio_exit_4}]}
+                        prio_msg_5, prio_msg_6, msg_2,
+                        {'EXIT', Self, prio_exit_3}, {'EXIT', Self, prio_exit_4}]}
                 = proc_info(Pid, messages)
     end,
 

erts/emulator/test/trace_SUITE.erl

@@ -30,6 +30,7 @@
          init_per_testcase/2, end_per_testcase/2,
          link_receive_call_correlation/0,
          receive_trace/1, receive_trace_non_fetching_receiver/1,
+         receive_trace_priority_messages/1,
          link_receive_call_correlation/1, self_send/1,
 	 timeout_trace/1, send_trace/1,
 	 procs_trace/1, dist_procs_trace/1, procs_new_trace/1,
@@ -69,6 +70,7 @@ groups() ->
 
 testcases() ->
     [cpu_timestamp, receive_trace, receive_trace_non_fetching_receiver,
+     receive_trace_priority_messages,
      link_receive_call_correlation, self_send, timeout_trace,
      send_trace, procs_trace, dist_procs_trace, suspend,
      suspend_exit, suspender_exit,
@@ -310,6 +312,68 @@ receive_trace_non_fetching_receiver_test(Node) ->
     _ = process_flag(trap_exit, TrapExit),
     ok.
 
+receive_trace_priority_messages(Config) when is_list(Config) ->
+    receive_trace_priority_messages_test(node()),
+    {ok, Peer, Node} = ?CT_PEER(),
+    receive_trace_priority_messages_test(Node),
+    peer:stop(Peer),
+    ok.
+
+receive_trace_priority_messages_test(Node) ->
+    Tester = self(),
+
+    PrioReceiver = spawn_link(fun () ->
+                                      Tester ! {self(), alias([priority])},
+                                      receive after infinity -> ok end
+                              end),
+    PriorityAlias = receive
+                        {PrioReceiver, PrioAlias} ->
+                            PrioAlias
+                    end,
+
+    1 = erlang_trace(PrioReceiver, true, ['receive']),
+
+    Msgs = [{msg, 1}, {prio_msg, 1}, {msg, 2}, {prio_msg, 2}, {msg, 3},
+            {msg, 4}, {prio_msg, 3}, {prio_msg, 4}, {msg, 5}, {prio_msg, 5},
+            {msg, 6}, {msg, 7}, {msg, 8}, {prio_msg, 6}, {prio_msg, 7}],
+
+    ok = erpc:call(Node,
+                   fun () ->
+                           lists:foreach(
+                             fun (Msg) ->
+                                     Tmo = rand:uniform(101) - 1,
+                                     receive after Tmo -> ok end,
+                                     case Msg of
+                                         {msg, _} ->
+                                             PrioReceiver ! Msg;
+                                         {prio_msg, _} ->
+                                             erlang:send(PriorityAlias, Msg,
+                                                         [priority])
+                                     end
+                             end, Msgs),
+                           ok
+                   end),
+
+    lists:foreach(fun (Msg) ->
+                          {trace, PrioReceiver, 'receive', Msg}
+                              = receive_first_trace()
+                  end, Msgs),
+
+    MsgQ = lists:sort(fun ({X, XA}, {X, XB}) ->
+                              XA =< XB;
+                          ({prio_msg, _}, {msg, _}) ->
+                              true;
+                          ({msg, _}, {prio_msg, _}) ->
+                              false
+                      end, Msgs),
+
+    {messages, MsgQ} = process_info(PrioReceiver, messages),
+
+    unlink(PrioReceiver),
+    exit(PrioReceiver, kill),
+    false = is_process_alive(PrioReceiver),
+    ok.
+
 %% Tests that receive of a message always happens before a call with
 %% that message and that links/unlinks are ordered together with the
 %% 'receive'.