Fix broken receive optimization

When a ref is created before performing a receive that will only
receive message containing that ref, there is a compiler optimization
to avoid scanning messages that can't possible contain the newly
created ref.

Magnus Lång pointed out that the implementation of the optimization
is flawed. Exceptions or recursive calls could cause the receive
operation to scan the receive queue from a position beyond the expected
message (that is, the message containing the ref would never be
matched out). See the receive_opt_exception/1 and receive_opt_recursion/1
test cases in receive_SUITE.

It turns out that we can simplify the implementation of the
optimization while fixing the bug (suggested by Magnus Lång). We
actually don't need the c_p->msg.mark field. It is enough to have
c_p->msg.saved_pos; if it is non-zero, it is a valid position in the
message qeueue.  All we need to do is to ensure that we clear
c_p->msg.saved_pos when a receive is exited normally or abnormally.

We can clear c_p->msg.saved_pos in JOIN_MESSAGE(), since it is called
both when leaving a receive because a message matched and because there
was a timeout and the 'after' clause was executed. In addition, we
need to clear c_p->msg.saved_pos when an exception is caught.

https://bugs.erlang.org/browse/ERL-511

5 files changed, 85 insertions, 22 deletions

diff --git a/erts/emulator/beam/beam_emu.c b/erts/emulator/beam/beam_emu.c
index bc95ccec52..ef9abcde08 100644
--- a/erts/emulator/beam/beam_emu.c
+++ b/erts/emulator/beam/beam_emu.c
@@ -1454,6 +1454,7 @@ handle_error(Process* c_p, BeamInstr* pc, Eterm* reg, ErtsCodeMFA *bif_mfa)
 	reg[3] = c_p->ftrace;
         if ((new_pc = next_catch(c_p, reg))) {
 	    c_p->cp = 0;	/* To avoid keeping stale references. */
+            c_p->msg.saved_last = 0;  /* No longer safe to use this position */
 	    return new_pc;
 	}
 	if (c_p->catches > 0) erts_exit(ERTS_ERROR_EXIT, "Catch not found");
diff --git a/erts/emulator/beam/erl_message.h b/erts/emulator/beam/erl_message.h
index 9c8cf84e43..a14f4f51d8 100644
--- a/erts/emulator/beam/erl_message.h
+++ b/erts/emulator/beam/erl_message.h
@@ -167,10 +167,9 @@ typedef struct {
     Sint len;            /* queue length */
 
     /*
-     * The following two fields are used by the recv_mark/1 and
+     * The following field is used by the recv_mark/1 and
      * recv_set/1 instructions.
      */
-    BeamInstr* mark;		/* address to rec_loop/2 instruction */
     ErtsMessage** saved_last;	/* saved last pointer */
 } ErlMessageQueue;
 
@@ -236,12 +235,17 @@ typedef struct erl_trace_message_queue__ {
      (p)->msg.len--; \
      if (__mp == NULL) \
          (p)->msg.last = (p)->msg.save; \
-     (p)->msg.mark = 0; \
 } while(0)
 
-/* Reset message save point (after receive match) */
-#define JOIN_MESSAGE(p) \
-     (p)->msg.save = &(p)->msg.first
+/*
+ * Reset message save point (after receive match).
+ * Also invalidate the saved position since it may no
+ * longer be safe to use.
+ */
+#define JOIN_MESSAGE(p) do {                    \
+    (p)->msg.save = &(p)->msg.first;            \
+    (p)->msg.saved_last = 0;                    \
+} while(0)
 
 /* Save current message */
 #define SAVE_MESSAGE(p) \
diff --git a/erts/emulator/beam/msg_instrs.tab b/erts/emulator/beam/msg_instrs.tab
index 8055a8616f..d6d4d2fb49 100644
--- a/erts/emulator/beam/msg_instrs.tab
+++ b/erts/emulator/beam/msg_instrs.tab
@@ -43,27 +43,23 @@
 //  *
 //  */
 
-recv_mark(Dest) {
+i_recv_mark() {
     /*
-     * Save the current position in message buffer and the
-     * the label for the loop_rec/2 instruction for the
-     * the receive statement.
+     * Save the current position in message buffer.
      */
-    $SET_REL_I(c_p->msg.mark, $Dest);
     c_p->msg.saved_last = c_p->msg.last;
 }
 
 i_recv_set() {
     /*
-     * If the mark is valid (points to the loop_rec/2
-     * instruction that follows), we know that the saved
-     * position points to the first message that could
-     * possibly be matched out.
+     * If c_p->msg.saved_last is non-zero, it points to the first
+     * message that could possibly be matched out.
      *
-     * If the mark is invalid, we do nothing, meaning that
-     * we will look through all messages in the message queue.
+     * If c_p->msg.saved_last is zero, it means that it was invalidated
+     * because another receive was executed before this i_recv_set()
+     * instruction was reached.
      */
-    if (c_p->msg.mark == (BeamInstr *) ($NEXT_INSTRUCTION)) {
+    if (c_p->msg.saved_last) {
         c_p->msg.save = c_p->msg.saved_last;
     }
     SET_I($NEXT_INSTRUCTION);
@@ -131,6 +127,7 @@ i_loop_rec(Dest) {
             ASSERT(HTOP == c_p->htop && E == c_p->stop);
             /* TODO: Add DTrace probe for this bad message situation? */
             UNLINK_MESSAGE(c_p, msgp);
+            c_p->msg.saved_last = 0; /* Better safe than sorry. */
             msgp->next = NULL;
             erts_cleanup_messages(msgp);
             goto loop_rec__;
diff --git a/erts/emulator/beam/ops.tab b/erts/emulator/beam/ops.tab
index a560bde920..3df91056cb 100644
--- a/erts/emulator/beam/ops.tab
+++ b/erts/emulator/beam/ops.tab
@@ -1565,7 +1565,12 @@ on_load
 #
 # R14A.
 #
-recv_mark f
+# Modified in OTP 21 because it turns out that we don't need the
+# label after all.
+#
+
+recv_mark f => i_recv_mark
+i_recv_mark
 
 recv_set Fail | label Lbl | loop_rec Lf Reg => \
    i_recv_set | label Lbl | loop_rec Lf Reg
diff --git a/erts/emulator/test/receive_SUITE.erl b/erts/emulator/test/receive_SUITE.erl
index a12019ec83..1fe11428b4 100644
--- a/erts/emulator/test/receive_SUITE.erl
+++ b/erts/emulator/test/receive_SUITE.erl
@@ -25,14 +25,16 @@
 -include_lib("common_test/include/ct.hrl").
 
 -export([all/0, suite/0,
-	 call_with_huge_message_queue/1,receive_in_between/1]).
+	 call_with_huge_message_queue/1,receive_in_between/1,
+         receive_opt_exception/1,receive_opt_recursion/1]).
 
 suite() ->
     [{ct_hooks,[ts_install_cth]},
      {timetrap, {minutes, 3}}].
 
-all() -> 
-    [call_with_huge_message_queue, receive_in_between].
+all() ->
+    [call_with_huge_message_queue, receive_in_between,
+     receive_opt_exception, receive_opt_recursion].
 
 call_with_huge_message_queue(Config) when is_list(Config) ->
     Pid = spawn_link(fun echo_loop/0),
@@ -113,6 +115,60 @@ receive_one() ->
 	dummy -> ok
     end.
 
+receive_opt_exception(_Config) ->
+    Recurse = fun() ->
+                      %% Overwrite with the same mark,
+                      %% and never consume it.
+                      ThrowFun = fun() -> throw(aborted) end,
+                      aborted = (catch do_receive_opt_exception(ThrowFun)),
+                      ok
+              end,
+    do_receive_opt_exception(Recurse),
+
+    %% Eat the second message.
+    receive
+        Ref when is_reference(Ref) -> ok
+    end.
+
+do_receive_opt_exception(Disturber) ->
+    %% Create a receive mark.
+    Ref = make_ref(),
+    self() ! Ref,
+    Disturber(),
+    receive
+        Ref ->
+            ok
+    after 0 ->
+            error(the_expected_message_was_not_there)
+    end.
+
+receive_opt_recursion(_Config) ->
+    Recurse = fun() ->
+                      %% Overwrite with the same mark,
+                      %% and never consume it.
+                      NoOp = fun() -> ok end,
+                      BlackHole = spawn(NoOp),
+                      expected = do_receive_opt_recursion(BlackHole, NoOp, true),
+                      ok
+              end,
+    do_receive_opt_recursion(self(), Recurse, false),
+    ok.
+
+do_receive_opt_recursion(Recipient, Disturber, IsInner) ->
+    Ref = make_ref(),
+    Recipient ! Ref,
+    Disturber(),
+    receive
+        Ref -> ok
+    after 0 ->
+            case IsInner of
+                true ->
+                    expected;
+                false ->
+                    error(the_expected_message_was_not_there)
+            end
+    end.
+
 %%%
 %%% Common helpers.
 %%%