erts: Implement recv_mark/1 and recv_set/1 for real

The recv_mark/1 instruction will both save the current
position in the message queue and a mark (the address of the
loop_rec/2 instruction just following the recv_set/1
instruction). The recv_mark/1 instruction will only
use the saved position if the mark is correct.

The reason for saving and verifying the mark is that
the compiler does not need to guarantee that no other
receive instruction can be executed in between the
recv_mark/1 and recv_set/1 instructions (the mark will
be cleared by the remove_message/0 instruction when a message
is removed from the message queue). That means that arbitrary
function calls in between those instruction can be allowed.

3 files changed, 45 insertions, 2 deletions

diff --git a/erts/emulator/beam/beam_emu.c b/erts/emulator/beam/beam_emu.c
index 5b2f032afc..ee8ba54cb8 100644
--- a/erts/emulator/beam/beam_emu.c
+++ b/erts/emulator/beam/beam_emu.c
@@ -1539,6 +1539,10 @@ void process_main(void)
  /*
   * Skeleton for receive statement:
   *
+  *             recv_mark L1                     Optional
+  *             call make_ref/monitor            Optional
+  *             ...
+  *             recv_set L1                      Optional
   *      L1:          <-------------------+
   *                   <-----------+       |
   *     	     	       	  |   	  |
@@ -1557,6 +1561,34 @@ void process_main(void)
   *
   */
 
+ OpCase(recv_mark_f): {
+     /*
+      * Save the current position in message buffer and the
+      * the label for the loop_rec/2 instruction for the
+      * the receive statement.
+      */
+     c_p->msg.mark = (BeamInstr *) Arg(0);
+     c_p->msg.saved_last = c_p->msg.last;
+     Next(1);
+ }
+
+ OpCase(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 the mark is invalid, we do nothing, meaning that
+      * we will look through all messages in the message queue.
+      */
+     if (c_p->msg.mark == (BeamInstr *) (I+1)) {
+	 c_p->msg.save = c_p->msg.saved_last;
+     }
+     I++;
+     /* Fall through to the loop_rec/2 instruction */
+ }
+
     /*
      * Pick up the next message and place it in x(0).
      * If no message, jump to a wait or wait_timeout instruction.
diff --git a/erts/emulator/beam/erl_message.h b/erts/emulator/beam/erl_message.h
index 459c6363aa..489dee7b37 100644
--- a/erts/emulator/beam/erl_message.h
+++ b/erts/emulator/beam/erl_message.h
@@ -75,6 +75,13 @@ typedef struct {
     ErlMessage** last;  /* point to the last next pointer */
     ErlMessage** save;
     int len;            /* queue length */
+
+    /*
+     * The following two fields are used by the recv_mark/1 and
+     * recv_set/1 instructions.
+     */
+    BeamInstr* mark;		/* address to rec_loop/2 instruction */
+    ErlMessage** saved_last;	/* saved last pointer */
 } ErlMessageQueue;
 
 #ifdef ERTS_SMP
@@ -137,6 +144,7 @@ do {							\
      (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) */
diff --git a/erts/emulator/beam/ops.tab b/erts/emulator/beam/ops.tab
index ef6062203e..49280a60e0 100644
--- a/erts/emulator/beam/ops.tab
+++ b/erts/emulator/beam/ops.tab
@@ -1425,5 +1425,8 @@ on_load
 #
 # R14A.
 #
-recv_mark Lbl =>
-recv_set Lbl =>
+recv_mark f
+
+recv_set Fail | label Lbl | loop_rec Lf Reg => \
+   i_recv_set | label Lbl | loop_rec Lf Reg
+i_recv_set