erts: Remove a copy of distribution data payload

Before this change the inet driver was in list mode and
thus the data from it had to be copied when received by
the dist entry. This change puts the tcp port in binary mode
and makes the any refc binary created by it be used all the way
to the process where it is decoded.

Thus eliminating one copy of the entire message payload.

5 files changed, 63 insertions, 27 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 21f994fd3e..3b6c0e5db9 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -85,10 +85,10 @@ dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
     }
     else {
 	ErlHeapFragment *mbuf = new_message_buffer(size);
-	erts_factory_static_init(&factory, ctl, ctl_len, &mbuf->off_heap);
+	erts_factory_static_init(&factory, mbuf->mem, ctl_len, &mbuf->off_heap);
 	msg = erts_decode_dist_ext(&factory, edep);
 	if (is_value(msg))
-	    erts_fprintf(stderr, "    %s: %T\n", what, msg);
+	    erts_fprintf(stderr, "    %s: %.80T\n", what, msg);
 	else {
 	    erts_fprintf(stderr,
 			 "DIST MSG DEBUG: erts_decode_dist_ext(%s) failed:\n",
@@ -1211,6 +1211,7 @@ int erts_net_message(Port *prt,
                      Uint32 conn_id,
 		     byte *hbuf,
 		     ErlDrvSizeT hlen,
+                     Binary *bin,
 		     byte *buf,
 		     ErlDrvSizeT len)
 {
@@ -1259,7 +1260,7 @@ int erts_net_message(Port *prt,
     bw(buf, len);
 #endif
 
-    res = erts_prepare_dist_ext(&ede, buf, len, dep, conn_id, dep->cache);
+    res = erts_prepare_dist_ext(&ede, buf, len, bin, dep, conn_id, dep->cache);
 
     switch (res) {
     case ERTS_PREP_DIST_EXT_CLOSED:
@@ -1303,7 +1304,7 @@ int erts_net_message(Port *prt,
     }
 
 #ifdef ERTS_DIST_MSG_DBG
-    erts_fprintf(stderr, "<< CTL: %T\n", arg);
+    erts_fprintf(stderr, "<< CTL: %.80T\n", arg);
 #endif
 
     if (is_not_tuple(arg) || 
@@ -1830,9 +1831,9 @@ erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx)
 	    }
 
     #ifdef ERTS_DIST_MSG_DBG
-            erts_fprintf(stderr, ">> CTL: %T\n", ctx->ctl);
+            erts_fprintf(stderr, ">> CTL: %.80T\n", ctx->ctl);
             if (is_value(ctx->msg))
-                erts_fprintf(stderr, "    MSG: %T\n", ctx->msg);
+                erts_fprintf(stderr, "    MSG: %.80T\n", ctx->msg);
     #endif
 
 	    ctx->data_size = ctx->max_finalize_prepend;
@@ -2613,6 +2614,7 @@ dist_ctrl_put_data_2(BIF_ALIST_2)
     ErlDrvSizeT size;
     Eterm input_handler;
     Uint32 conn_id;
+    Binary *bin = NULL;
 
     if (is_binary(BIF_ARG_2))
         size = binary_size(BIF_ARG_2);
@@ -2638,13 +2640,27 @@ dist_ctrl_put_data_2(BIF_ALIST_2)
     if (size != 0) {
         byte *data, *temp_alloc = NULL;
 
-        data = (byte *) erts_get_aligned_binary_bytes(BIF_ARG_2, &temp_alloc);
+        if (binary_bitoffset(BIF_ARG_2))
+            data = (byte *) erts_get_aligned_binary_bytes(BIF_ARG_2, &temp_alloc);
+        else {
+            Eterm real_bin;
+            ProcBin *proc_bin;
+            Uint offset, bitoffs, bitsize;
+
+            ERTS_GET_REAL_BIN(BIF_ARG_2, real_bin, offset, bitoffs, bitsize);
+            ASSERT(bitoffs == 0);
+            data = binary_bytes(real_bin) + offset;
+            proc_bin = (ProcBin *)binary_val(real_bin);
+            if (proc_bin->thing_word == HEADER_PROC_BIN)
+                bin = proc_bin->val;
+        }
+
         if (!data)
             BIF_ERROR(BIF_P, BADARG);
 
         erts_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
 
-        (void) erts_net_message(NULL, dep, conn_id, NULL, 0, data, size);
+        (void) erts_net_message(NULL, dep, conn_id, NULL, 0, bin, data, size);
         /*
          * We ignore any decode failures. On fatal failures the
          * connection will be taken down by killing the
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index 9a66e491f3..4b8040b356 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -640,14 +640,15 @@ erts_make_dist_ext_copy(ErtsDistExternal *edep, Uint xsize)
 {
     size_t align_sz;
     size_t dist_ext_sz;
-    size_t ext_sz;
+    size_t ext_sz = 0;
     byte *ep;
     ErtsDistExternal *new_edep;
 
     dist_ext_sz = ERTS_DIST_EXT_SIZE(edep);
     ASSERT(edep->ext_endp && edep->extp);
     ASSERT(edep->ext_endp >= edep->extp);
-    ext_sz = edep->ext_endp - edep->extp;
+    if (edep->binp == NULL)
+        ext_sz = edep->ext_endp - edep->extp;
 
     align_sz = ERTS_EXTRA_DATA_ALIGN_SZ(dist_ext_sz + ext_sz);
 
@@ -656,20 +657,36 @@ erts_make_dist_ext_copy(ErtsDistExternal *edep, Uint xsize)
 
     ep = (byte *) new_edep;
     sys_memcpy((void *) ep, (void *) edep, dist_ext_sz);
-    ep += dist_ext_sz;
     if (new_edep->dep)
         erts_ref_dist_entry(new_edep->dep);
-    new_edep->extp = ep;
-    new_edep->ext_endp = ep + ext_sz;
     new_edep->heap_size = -1;
-    sys_memcpy((void *) ep, (void *) edep->extp, ext_sz);
+
+    if (ext_sz) {
+        ep += dist_ext_sz;
+        new_edep->extp = ep;
+        new_edep->ext_endp = ep + ext_sz;
+        sys_memcpy((void *) ep, (void *) edep->extp, ext_sz);
+    } else {
+        erts_refc_inc(&new_edep->binp->intern.refc, 2);
+    }
     return new_edep;
 }
 
+void
+erts_free_dist_ext_copy(ErtsDistExternal *edep)
+{
+    if (edep->dep)
+	erts_deref_dist_entry(edep->dep);
+    if (edep->binp)
+        erts_bin_release(edep->binp);
+    erts_free(ERTS_ALC_T_EXT_TERM_DATA, edep);
+}
+
 int
 erts_prepare_dist_ext(ErtsDistExternal *edep,
 		      byte *ext,
 		      Uint size,
+                      Binary *binp,
 		      DistEntry *dep,
                       Uint32 conn_id,
 		      ErtsAtomCache *cache)
@@ -685,6 +702,7 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 
     ASSERT(dep->flags & DFLAG_UTF8_ATOMS);
 
+
     if ((dep->state != ERTS_DE_STATE_CONNECTED &&
          dep->state != ERTS_DE_STATE_PENDING)
         || dep->connection_id != conn_id) {
@@ -697,7 +715,11 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
         ext++;
         size--;
     }
+
+    edep->heap_size = -1;
     edep->ext_endp = ext + size;
+    edep->binp = binp;
+
     ep = ext;
 
     if (size < 2)
diff --git a/erts/emulator/beam/external.h b/erts/emulator/beam/external.h
index edac177cc6..1223f90fcc 100644
--- a/erts/emulator/beam/external.h
+++ b/erts/emulator/beam/external.h
@@ -122,10 +122,13 @@ typedef struct {
 
 #define ERTS_DIST_CON_ID_MASK ((Uint32) 0x00ffffff) /* also in net_kernel.erl */
 
-typedef struct {
+struct binary;
+
+typedef struct erl_dist_external {
     DistEntry *dep;
     byte *extp;
     byte *ext_endp;
+    struct binary *binp;
     Sint heap_size;
     Uint32 connection_id;
     Uint32 flags;
@@ -171,7 +174,7 @@ Uint erts_encode_ext_size_ets(Eterm);
 void erts_encode_ext(Eterm, byte **);
 byte* erts_encode_ext_ets(Eterm, byte *, struct erl_off_heap_header** ext_off_heap);
 
-ERTS_GLB_INLINE void erts_free_dist_ext_copy(ErtsDistExternal *);
+void erts_free_dist_ext_copy(ErtsDistExternal *);
 ERTS_GLB_INLINE void *erts_dist_ext_trailer(ErtsDistExternal *);
 ErtsDistExternal *erts_make_dist_ext_copy(ErtsDistExternal *, Uint);
 void *erts_dist_ext_trailer(ErtsDistExternal *);
@@ -181,8 +184,8 @@ void erts_destroy_dist_ext_copy(ErtsDistExternal *);
 #define ERTS_PREP_DIST_EXT_SUCCESS      (0)
 #define ERTS_PREP_DIST_EXT_CLOSED       (1)
 
-int erts_prepare_dist_ext(ErtsDistExternal *, byte *, Uint,
-			  DistEntry *, Uint32 conn_id, ErtsAtomCache *);
+int erts_prepare_dist_ext(ErtsDistExternal *, byte *, Uint, struct binary *,
+			  DistEntry *, Uint32, ErtsAtomCache *);
 Sint erts_decode_dist_ext_size(ErtsDistExternal *);
 Eterm erts_decode_dist_ext(ErtsHeapFactory* factory, ErtsDistExternal *);
 
@@ -202,14 +205,6 @@ void transcode_free_ctx(DistEntry* dep);
 
 #if ERTS_GLB_INLINE_INCL_FUNC_DEF
 
-ERTS_GLB_INLINE void
-erts_free_dist_ext_copy(ErtsDistExternal *edep)
-{
-    if (edep->dep)
-	erts_deref_dist_entry(edep->dep);
-    erts_free(ERTS_ALC_T_EXT_TERM_DATA, edep);
-}
-
 ERTS_GLB_INLINE void *
 erts_dist_ext_trailer(ErtsDistExternal *edep)
 {
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index 29de162b42..52935d834c 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -1093,7 +1093,7 @@ extern int distribution_info(fmtfn_t, void *);
 extern int is_node_name_atom(Eterm a);
 
 extern int erts_net_message(Port *, DistEntry *, Uint32 conn_id,
-			    byte *, ErlDrvSizeT, byte *, ErlDrvSizeT);
+			    byte *, ErlDrvSizeT, Binary *, byte *, ErlDrvSizeT);
 
 extern void init_dist(void);
 extern int stop_dist(void);
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index 6a2be52e11..36824fe62c 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -6181,6 +6181,7 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
 				dep,
                                 conn_id,
 				(byte*) hbuf, hlen,
+                                ErlDrvBinary2Binary(bin),
 				(byte*) (bin->orig_bytes+offs), len);
     }
     else
@@ -6226,12 +6227,14 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen,
 				    dep,
                                     conn_id,
 				    NULL, 0,
+                                    NULL,
 				    (byte*) hbuf, hlen);
 	else
 	    return erts_net_message(prt,
 				    dep,
                                     conn_id,
 				    (byte*) hbuf, hlen,
+                                    NULL,
 				    (byte*) buf, len);
     }
     else if (state & ERTS_PORT_SFLG_LINEBUF_IO)