Merge branch 'lukas/erts/dist-optimize-large-data/PR-2291/OTP-15926' into maint

* lukas/erts/dist-optimize-large-data/PR-2291/OTP-15926:
  erts: Don't copy binary data from dist data
  erts: Move copy of payload to receiving process

5 files changed, 62 insertions, 40 deletions

diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index d8501ea6ac..4537e3e569 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -80,7 +80,7 @@ dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
     byte *extp = edep->data->extp;
     Eterm msg;
     Sint ctl_len;
-    Sint size = ctl_len = erts_decode_dist_ext_size(edep, 0);
+    Sint size = ctl_len = erts_decode_dist_ext_size(edep, 0, 0);
     if (size < 0) {
 	erts_fprintf(dbg_file,
 		     "DIST MSG DEBUG: erts_decode_dist_ext_size(%s) failed:\n",
@@ -1462,7 +1462,7 @@ int erts_net_message(Port *prt,
 #endif
         goto data_error;
     case ERTS_PREP_DIST_EXT_SUCCESS:
-	ctl_len = erts_decode_dist_ext_size(&ede, 1);
+	ctl_len = erts_decode_dist_ext_size(&ede, 1, 0);
         if (ctl_len < 0) {
 #ifdef ERTS_DIST_MSG_DBG
             erts_fprintf(dbg_file, "DIST MSG DEBUG: erts_decode_dist_ext_size(CTL) failed:\n");
@@ -1543,39 +1543,6 @@ int erts_net_message(Port *prt,
         edep = erts_get_dist_ext(&seq->hfrag);
         ede_hfrag = &seq->hfrag;
 
-        /* If the sequence consisted of more than 1 fragment we create one large
-           binary out of all of the fragments. This because erts_decode_ext
-           cannot handle a segmented buffer.
-           TODO: Move this copy to as late as possible, preferably in in the
-           erts_decode_dist_ext in the receiving process.
-        */
-        if (edep->data->frag_id > 1) {
-            Uint sz = 0;
-            Binary *bin;
-            int i;
-            byte *ep;
-
-            for (i = 0; i < edep->data->frag_id; i++)
-                sz += edep->data[i].ext_endp - edep->data[i].extp;
-
-            bin = erts_bin_nrml_alloc(sz);
-            ep = (byte*)bin->orig_bytes;
-
-            for (i = 0; i < edep->data->frag_id; i++) {
-                sys_memcpy(ep, edep->data[i].extp, edep->data[i].ext_endp - edep->data[i].extp);
-                ep += edep->data[i].ext_endp - edep->data[i].extp;
-                erts_bin_release(edep->data[i].binp);
-                edep->data[i].binp = NULL;
-                edep->data[i].extp = NULL;
-                edep->data[i].ext_endp = NULL;
-            }
-
-            edep->data->frag_id = 1;
-            edep->data->extp = (byte*)bin->orig_bytes;
-            edep->data->ext_endp = ep;
-            edep->data->binp = bin;
-        }
-
         break;
     }
     default:
diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c
index 6645341512..1bebf6efe2 100644
--- a/erts/emulator/beam/erl_message.c
+++ b/erts/emulator/beam/erl_message.c
@@ -527,7 +527,7 @@ erts_msg_attached_data_size_aux(ErtsMessage *msg)
 
     if (edep->heap_size < 0) {
 
-        sz = erts_decode_dist_ext_size(edep, 1);
+        sz = erts_decode_dist_ext_size(edep, 1, 1);
         if (sz < 0) {
             /* Bad external
              * We leave the message intact in this case as it's not worth the trouble
diff --git a/erts/emulator/beam/erl_proc_sig_queue.c b/erts/emulator/beam/erl_proc_sig_queue.c
index 55e469b553..fb900ca7ba 100644
--- a/erts/emulator/beam/erl_proc_sig_queue.c
+++ b/erts/emulator/beam/erl_proc_sig_queue.c
@@ -3028,7 +3028,7 @@ erts_proc_sig_decode_dist(Process *proc, ErtsProcLocks proc_locks,
     if (edep->heap_size >= 0)
 	need = edep->heap_size;
     else {
-	need = erts_decode_dist_ext_size(edep, 1);
+	need = erts_decode_dist_ext_size(edep, 1, 1);
 	if (need < 0) {
 	    /* bad signal; remove it... */
 	    return 0;
diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index ec67ab2aed..ce61cdf040 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -1062,11 +1062,38 @@ bad_dist_ext(ErtsDistExternal *edep)
 }
 
 Sint
-erts_decode_dist_ext_size(ErtsDistExternal *edep, int kill_connection)
+erts_decode_dist_ext_size(ErtsDistExternal *edep, int kill_connection, int payload)
 {
     Sint res;
     byte *ep;
 
+    if (edep->data->frag_id > 1 && payload) {
+        Uint sz = 0;
+        Binary *bin;
+        int i;
+        byte *ep;
+
+        for (i = 0; i < edep->data->frag_id; i++)
+            sz += edep->data[i].ext_endp - edep->data[i].extp;
+
+        bin = erts_bin_nrml_alloc(sz);
+        ep = (byte*)bin->orig_bytes;
+
+        for (i = 0; i < edep->data->frag_id; i++) {
+            sys_memcpy(ep, edep->data[i].extp, edep->data[i].ext_endp - edep->data[i].extp);
+            ep += edep->data[i].ext_endp - edep->data[i].extp;
+            erts_bin_release(edep->data[i].binp);
+            edep->data[i].binp = NULL;
+            edep->data[i].extp = NULL;
+            edep->data[i].ext_endp = NULL;
+        }
+
+        edep->data->frag_id = 1;
+        edep->data->extp = (byte*)bin->orig_bytes;
+        edep->data->ext_endp = ep;
+        edep->data->binp = bin;
+    }
+
     if (edep->data->extp >= edep->data->ext_endp)
 	goto fail;
 #ifndef ERTS_DEBUG_USE_DIST_SEP
@@ -1164,6 +1191,7 @@ Eterm erts_decode_ext(ErtsHeapFactory* factory, byte **ext, Uint32 flags)
     if (flags) {
         ASSERT(flags == ERTS_DIST_EXT_BTT_SAFE);
         ede.flags = flags; /* a dummy struct just for the flags */
+        ede.data = NULL;
         edep = &ede;
     } else {
         edep = NULL;
@@ -1233,8 +1261,10 @@ BIF_RETTYPE erts_debug_dist_ext_to_term_2(BIF_ALIST_2)
 
     ede.data->extp = binary_bytes(real_bin)+offset;
     ede.data->ext_endp = ede.data->extp + size;
+    ede.data->frag_id = 1;
+    ede.data->binp = NULL;
 
-    hsz = erts_decode_dist_ext_size(&ede, 1);
+    hsz = erts_decode_dist_ext_size(&ede, 1, 1);
     if (hsz < 0)
 	goto badarg;
 
@@ -1765,6 +1795,7 @@ static BIF_RETTYPE binary_to_term_int(Process* p, Eterm bin, B2TContext *ctx)
         case B2TDecodeBinary: {
 	    ErtsDistExternal fakedep;
             fakedep.flags = ctx->flags;
+            fakedep.data = NULL;
             dec_term(&fakedep, NULL, NULL, NULL, ctx);
             break;
 	}
@@ -3762,6 +3793,30 @@ dec_term_atom_common:
 		    hp += heap_bin_size(n);
 		    sys_memcpy(hb->data, ep, n);
 		    *objp = make_binary(hb);
+		} else if (edep && edep->data && edep->data->binp &&
+                           n > (edep->data->binp->orig_size / 4)) {
+                    /* If we decode a refc binary from a distribution data
+                       entry we know that it is a refc binary to begin with
+                       so we just increment it and use the reference. This
+                       means that the entire distribution data entry will
+                       remain until this binary is de-allocated so we only
+                       do it if a substantial part (> 25%) of the data
+                       is a binary. */
+                    ProcBin* pb = (ProcBin *) hp;
+                    Binary* bptr = edep->data->binp;
+                    erts_refc_inc(&bptr->intern.refc, 1);
+                    pb->thing_word = HEADER_PROC_BIN;
+                    pb->size = n;
+                    pb->next = factory->off_heap->first;
+                    factory->off_heap->first = (struct erl_off_heap_header*)pb;
+                    pb->val = bptr;
+                    pb->bytes = (byte*) ep;
+                    ERTS_ASSERT((byte*)(bptr->orig_bytes) < ep &&
+                                ep+n <= (byte*)(bptr->orig_bytes+bptr->orig_size));
+                    pb->flags = 0;
+                    OH_OVERHEAD(factory->off_heap, pb->size / sizeof(Eterm));
+                    hp += PROC_BIN_SIZE;
+                    *objp = make_binary(pb);
 		} else {
 		    Binary* dbin = erts_bin_nrml_alloc(n);
 
diff --git a/erts/emulator/beam/external.h b/erts/emulator/beam/external.h
index b556c9076c..e362a6c81f 100644
--- a/erts/emulator/beam/external.h
+++ b/erts/emulator/beam/external.h
@@ -199,7 +199,7 @@ typedef enum {
 
 ErtsPrepDistExtRes erts_prepare_dist_ext(ErtsDistExternal *, byte *, Uint, struct binary *,
                                          DistEntry *, Uint32, ErtsAtomCache *);
-Sint erts_decode_dist_ext_size(ErtsDistExternal *, int);
+Sint erts_decode_dist_ext_size(ErtsDistExternal *, int, int);
 Eterm erts_decode_dist_ext(ErtsHeapFactory*, ErtsDistExternal *, int);
 
 Sint erts_decode_ext_size(byte*, Uint);