Add enif_ioq_peek_head

This introduces a way to retrieve erlang terms from NIF IO queues
without having to resort to copying.

OTP-14797

6 files changed, 103 insertions, 3 deletions

diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml
index ef3cdb89e9..23f0c66429 100644
--- a/erts/doc/src/erl_nif.xml
+++ b/erts/doc/src/erl_nif.xml
@@ -367,6 +367,8 @@ return term;</code>
             <c>enif_ioq_deq()</c></seealso></item>
             <item><seealso marker="#enif_ioq_peek">
             <c>enif_ioq_peek()</c></seealso></item>
+            <item><seealso marker="#enif_ioq_peek_head">
+            <c>enif_ioq_peek_head()</c></seealso></item>
             <item><seealso marker="#enif_inspect_iovec">
             <c>enif_inspect_iovec()</c></seealso></item>
             <item><seealso marker="#enif_free_iovec">
@@ -1683,8 +1685,7 @@ enif_inspect_iovec(env, max_elements, term, &amp;tail, &amp;iovec);
       <fsummary>Peek inside the IO Queue</fsummary>
       <desc>
         <p>Get the I/O queue as a pointer to an array of <c>SysIOVec</c>s.
-          It also returns the number of elements in <c>iovlen</c>.
-          This is the only way to get data out of the queue.</p>
+           It also returns the number of elements in <c>iovlen</c>.</p>
         <p>Nothing is removed from the queue by this function, that must be done
           with <seealso marker="#enif_ioq_deq"><c>enif_ioq_deq</c></seealso>.</p>
          <p>The returned array is suitable to use with the Unix system
@@ -1693,6 +1694,21 @@ enif_inspect_iovec(env, max_elements, term, &amp;tail, &amp;iovec);
     </func>
 
     <func>
+      <name><ret>int</ret>
+      <nametext>enif_ioq_peek_head(ErlNifEnv *env, ErlNifIOQueue *q, size_t *size, ERL_NIF_TERM *bin_term)</nametext></name>
+      <fsummary>Peek the head of the IO Queue.</fsummary>
+      <desc>
+        <p>Get the head of the IO Queue as a binary term.</p>
+        <p>If <c>size</c> is not <c>NULL</c>, the size of the head is placed
+           there.</p>
+        <p>Nothing is removed from the queue by this function, that must be done
+           with <seealso marker="#enif_ioq_deq"><c>enif_ioq_deq</c></seealso>.</p>
+        <p>Returns <c>true</c> on success, or <c>false</c> if the queue is
+           empty.</p>
+      </desc>
+    </func>
+
+    <func>
       <name><ret>size_t</ret>
       <nametext>enif_ioq_size(ErlNifIOQueue *q)</nametext></name>
       <fsummary>Get the current size of the IO Queue</fsummary>
diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index dbcc894ac9..abb490cf9c 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -3605,6 +3605,55 @@ int enif_ioq_deq(ErlNifIOQueue *q, size_t elems, size_t *size)
     return 1;
 }
 
+int enif_ioq_peek_head(ErlNifEnv *env, ErlNifIOQueue *q, size_t *size, ERL_NIF_TERM *bin_term) {
+    SysIOVec *iov_entry;
+    Binary *ref_bin;
+
+    if (q->size == 0) {
+        return 0;
+    }
+
+    ASSERT(q->b_head != q->b_tail && q->v_head != q->v_tail);
+
+    ref_bin = &q->b_head[0]->nif;
+    iov_entry = &q->v_head[0];
+
+    if (size != NULL) {
+        *size = iov_entry->iov_len;
+    }
+
+    if (iov_entry->iov_len > ERL_ONHEAP_BIN_LIMIT) {
+        ProcBin *pb = (ProcBin*)alloc_heap(env, PROC_BIN_SIZE);
+
+        pb->thing_word = HEADER_PROC_BIN;
+        pb->next = MSO(env->proc).first;
+        pb->val = ref_bin;
+        pb->flags = 0;
+
+        ASSERT((byte*)iov_entry->iov_base >= (byte*)ref_bin->orig_bytes);
+        ASSERT(iov_entry->iov_len <= ref_bin->orig_size);
+
+        pb->bytes = (byte*)iov_entry->iov_base;
+        pb->size = iov_entry->iov_len;
+
+        MSO(env->proc).first = (struct erl_off_heap_header*) pb;
+        OH_OVERHEAD(&(MSO(env->proc)), pb->size / sizeof(Eterm));
+
+        erts_refc_inc(&ref_bin->intern.refc, 2);
+        *bin_term = make_binary(pb);
+    } else {
+        ErlHeapBin* hb = (ErlHeapBin*)alloc_heap(env, heap_bin_size(iov_entry->iov_len));
+
+        hb->thing_word = header_heap_bin(iov_entry->iov_len);
+        hb->size = iov_entry->iov_len;
+
+        sys_memcpy(hb->data, iov_entry->iov_base, iov_entry->iov_len);
+        *bin_term = make_binary(hb);
+    }
+
+    return 1;
+}
+
 SysIOVec *enif_ioq_peek(ErlNifIOQueue *q, int *iovlen)
 {
     return erts_ioq_peekq(q, iovlen);
diff --git a/erts/emulator/beam/erl_nif.h b/erts/emulator/beam/erl_nif.h
index 7fb447e4a8..053f7673c4 100644
--- a/erts/emulator/beam/erl_nif.h
+++ b/erts/emulator/beam/erl_nif.h
@@ -52,9 +52,10 @@
 ** 2.11: 19.0 enif_snprintf 
 ** 2.12: 20.0 add enif_select, enif_open_resource_type_x
 ** 2.13: 20.1 add enif_ioq
+** 2.14: 21.0 add enif_ioq_peek_head
 */
 #define ERL_NIF_MAJOR_VERSION 2
-#define ERL_NIF_MINOR_VERSION 13
+#define ERL_NIF_MINOR_VERSION 14
 
 /*
  * The emulator will refuse to load a nif-lib with a major version
diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h
index 9e573307d8..3750fd9b68 100644
--- a/erts/emulator/beam/erl_nif_api_funcs.h
+++ b/erts/emulator/beam/erl_nif_api_funcs.h
@@ -198,6 +198,7 @@ ERL_NIF_API_FUNC_DECL(SysIOVec*,enif_ioq_peek,(ErlNifIOQueue *q, int *iovlen));
 ERL_NIF_API_FUNC_DECL(int,enif_inspect_iovec,(ErlNifEnv *env, size_t max_length, ERL_NIF_TERM iovec_term, ERL_NIF_TERM *tail, ErlNifIOVec **iovec));
 ERL_NIF_API_FUNC_DECL(void,enif_free_iovec,(ErlNifIOVec *iov));
 
+ERL_NIF_API_FUNC_DECL(int,enif_ioq_peek_head,(ErlNifEnv *env, ErlNifIOQueue *q, size_t *size, ERL_NIF_TERM *head));
 
 /*
 ** ADD NEW ENTRIES HERE (before this comment) !!!
@@ -373,6 +374,7 @@ ERL_NIF_API_FUNC_DECL(void,enif_free_iovec,(ErlNifIOVec *iov));
 #  define enif_ioq_peek ERL_NIF_API_FUNC_MACRO(enif_ioq_peek)
 #  define enif_inspect_iovec ERL_NIF_API_FUNC_MACRO(enif_inspect_iovec)
 #  define enif_free_iovec ERL_NIF_API_FUNC_MACRO(enif_free_iovec)
+#  define enif_ioq_peek_head ERL_NIF_API_FUNC_MACRO(enif_ioq_peek_head)
 
 /*
 ** ADD NEW ENTRIES HERE (before this comment)
diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index bec2291867..2afeb7e4cf 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -3037,14 +3037,17 @@ nif_ioq(Config) ->
          {enqbraw,a},
          {enqbraw,a, 5},
          {peek,   a},
+         {peek_head,  a},
          {deq,    a, 42},
 
          %% Test enqv
          {enqv,   a, 2, 100},
+         {peek_head,  a},
          {deq,    a, all},
 
          %% This skips all elements but one in the iolist
          {enqv,   a, 5, iolist_size(nif_ioq_payload(5)) - 1},
+         {peek_head,  a},
          {peek,   a},
 
          %% Test to enqueue a bunch of refc binaries
@@ -3074,9 +3077,13 @@ nif_ioq(Config) ->
 
     Q = ioq_nif(create),
 
+    false = ioq_nif(peek_head, Q),
+
     {'EXIT', {badarg, _}} = (catch ioq_nif(deq, Q, 1)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, 1, 1234)),
 
+    false = ioq_nif(peek_head, Q),
+
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, [atom_in_list], 0)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, [make_ref()], 0)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, [256], 0)),
@@ -3085,6 +3092,8 @@ nif_ioq(Config) ->
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, [1 bsl 64], 0)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(enqv, Q, [{tuple}], 0)),
 
+    false = ioq_nif(peek_head, Q),
+
     {'EXIT', {badarg, _}} = (catch ioq_nif(inspect,  [atom_in_list], use_stack)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(inspect,  [make_ref()], no_stack)),
     {'EXIT', {badarg, _}} = (catch ioq_nif(inspect,  [256], use_stack)),
@@ -3145,6 +3154,20 @@ nif_ioq_run([{peek, Name} = H|T], State) ->
 
     true = iolist_to_binary(B) == iolist_to_binary(Data),
     nif_ioq_run(T, State);
+nif_ioq_run([{peek_head, Name} = H|T], State) ->
+    #{ q := IOQ, b := B } = maps:get(Name, State),
+    RefData = iolist_to_binary(B),
+
+    ct:log("~p", [H]),
+
+    {true, QueueHead} = ioq_nif(peek_head, IOQ),
+    true = byte_size(QueueHead) > 0,
+
+    {RefHead, _Tail} = split_binary(RefData, byte_size(QueueHead)),
+
+    true = QueueHead =:= RefHead,
+
+    nif_ioq_run(T, State);
 nif_ioq_run([{deq, Name, all}|T], State) ->
     #{ q := IOQ, b := B } = maps:get(Name, State),
     Size = ioq_nif(size, IOQ),
diff --git a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
index 79560a38aa..fa9ae1015c 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -3403,6 +3403,15 @@ static ERL_NIF_TERM ioq(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
 		return enif_make_badarg(env);
 	    else
 		return enif_make_atom(env, "true");
+        } else if (enif_is_identical(argv[0], enif_make_atom(env, "peek_head"))) {
+            ERL_NIF_TERM head_term;
+
+            if(enif_ioq_peek_head(env, ioq->q, NULL, &head_term)) {
+                return enif_make_tuple2(env,
+                    enif_make_atom(env, "true"), head_term);
+            }
+
+            return enif_make_atom(env, "false");
         } else if (enif_is_identical(argv[0], enif_make_atom(env, "peek"))) {
             int iovlen, num, i, off = 0;
             SysIOVec *iov = enif_ioq_peek(ioq->q, &iovlen);