1 files changed, 129 insertions, 31 deletions

diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index f32b999198..d4765c50b8 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -1,18 +1,19 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 1996-2014. All Rights Reserved.
+ * Copyright Ericsson AB 1996-2017. All Rights Reserved.
  *
- * The contents of this file are subject to the Erlang Public License,
- * Version 1.1, (the "License"); you may not use this file except in
- * compliance with the License. You should have received a copy of the
- * Erlang Public License along with this software. If not, it can be
- * retrieved online at http://www.erlang.org/.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
  *
- * Software distributed under the License is distributed on an "AS IS"
- * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
- * the License for the specific language governing rights and limitations
- * under the License.
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
  *
  * %CopyrightEnd%
  */
@@ -22,6 +23,7 @@
 
 #include "erl_process.h"
 #include "erl_node_tables.h"
+#include "zlib.h"
 
 #define DFLAG_PUBLISHED           0x01
 #define DFLAG_ATOM_CACHE          0x02
@@ -41,6 +43,8 @@
 #define DFLAG_INTERNAL_TAGS       0x8000
 #define DFLAG_UTF8_ATOMS          0x10000
 #define DFLAG_MAP_TAG             0x20000
+#define DFLAG_BIG_CREATION        0x40000
+#define DFLAG_SEND_SENDER         0x80000
 
 /* All flags that should be enabled when term_to_binary/1 is used. */
 #define TERM_TO_BINARY_DFLAGS (DFLAG_EXTENDED_REFERENCES	\
@@ -49,7 +53,8 @@
 			       | DFLAG_EXTENDED_PIDS_PORTS	\
 			       | DFLAG_EXPORT_PTR_TAG		\
 			       | DFLAG_BIT_BINARIES             \
-			       | DFLAG_MAP_TAG)
+			       | DFLAG_MAP_TAG                  \
+                               | DFLAG_BIG_CREATION)
 
 /* opcodes used in distribution messages */
 #define DOP_LINK		1
@@ -70,6 +75,9 @@
 #define DOP_DEMONITOR_P		20
 #define DOP_MONITOR_P_EXIT	21
 
+#define DOP_SEND_SENDER         22
+#define DOP_SEND_SENDER_TT      23
+
 /* distribution trap functions */
 extern Export* dsend2_trap;
 extern Export* dsend3_trap;
@@ -96,7 +104,7 @@ typedef struct {
 } ErtsDSigData;
 
 #define ERTS_DE_IS_NOT_CONNECTED(DEP) \
-  (ERTS_SMP_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&(DEP)->rwmtx) \
+  (ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&(DEP)->rwmtx) \
 		      || erts_lc_rwmtx_is_rwlocked(&(DEP)->rwmtx)), \
    (is_nil((DEP)->cid) || ((DEP)->status & ERTS_DE_SFLG_EXITING)))
 
@@ -111,8 +119,8 @@ extern int erts_is_alive;
  * erts_dsig_prepare() prepares a send of a distributed signal.
  * One of the values defined below are returned. If the returned
  * value is another than ERTS_DSIG_PREP_CONNECTED, the
- * distributed signal cannot be sent before apropriate actions
- * have been taken. Apropriate actions would typically be setting
+ * distributed signal cannot be sent before appropriate actions
+ * have been taken. Appropriate actions would typically be setting
  * up the connection.
  */
 
@@ -149,21 +157,18 @@ erts_dsig_prepare(ErtsDSigData *dsdp,
     if (!dep)
 	return ERTS_DSIG_PREP_NOT_CONNECTED;
     if (dspl == ERTS_DSP_RWLOCK)
-	erts_smp_de_rwlock(dep);
+	erts_de_rwlock(dep);
     else
-	erts_smp_de_rlock(dep);
+	erts_de_rlock(dep);
     if (ERTS_DE_IS_NOT_CONNECTED(dep)) {
 	failure = ERTS_DSIG_PREP_NOT_CONNECTED;
 	goto fail;
     }
     if (no_suspend) {
-	failure = ERTS_DSIG_PREP_CONNECTED;
-	erts_smp_mtx_lock(&dep->qlock);
-	if (dep->qflgs & ERTS_DE_QFLG_BUSY)
+        if (erts_atomic32_read_acqb(&dep->qflgs) & ERTS_DE_QFLG_BUSY) {
 	    failure = ERTS_DSIG_PREP_WOULD_SUSPEND;
-	erts_smp_mtx_unlock(&dep->qlock);
-	if (failure == ERTS_DSIG_PREP_WOULD_SUSPEND)
 	    goto fail;
+        }
     }
     dsdp->proc = proc;
     dsdp->dep = dep;
@@ -171,14 +176,14 @@ erts_dsig_prepare(ErtsDSigData *dsdp,
     dsdp->connection_id = dep->connection_id;
     dsdp->no_suspend = no_suspend;
     if (dspl == ERTS_DSP_NO_LOCK)
-	erts_smp_de_runlock(dep);
+	erts_de_runlock(dep);
     return ERTS_DSIG_PREP_CONNECTED;
 
  fail:
     if (dspl == ERTS_DSP_RWLOCK)
-	erts_smp_de_rwunlock(dep);
+	erts_de_rwunlock(dep);
     else
-	erts_smp_de_runlock(dep);
+	erts_de_runlock(dep);
     return failure;
 
 }
@@ -190,7 +195,7 @@ void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry)
     Eterm id;
 
     if (prt) {
-	ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt));
+	ERTS_LC_ASSERT(erts_lc_is_port_locked(prt));
 	ASSERT((erts_atomic32_read_nob(&prt->state)
 		& ERTS_PORT_SFLGS_DEAD) == 0);
 	ASSERT(prt->dist_entry);
@@ -200,7 +205,7 @@ void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry)
     }
     else {
 	ASSERT(dist_entry);
-	ERTS_SMP_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&dist_entry->rwmtx)
+	ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(&dist_entry->rwmtx)
 			   || erts_lc_rwmtx_is_rwlocked(&dist_entry->rwmtx));
 	ASSERT(is_internal_port(dist_entry->cid));
 
@@ -208,7 +213,7 @@ void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry)
 	id = dep->cid;
     }
 
-    if (!erts_smp_atomic_xchg_mb(&dep->dist_cmd_scheduled, 1))
+    if (!erts_atomic_xchg_mb(&dep->dist_cmd_scheduled, 1))
 	erts_port_task_schedule(id, &dep->dist_cmd, ERTS_PORT_TASK_DIST_CMD);
 }
 
@@ -234,7 +239,7 @@ erts_remove_dist_link(ErtsDistLinkData *dldp,
 		      Eterm rid,
 		      DistEntry *dep)
 {
-    erts_smp_de_links_lock(dep);
+    erts_de_links_lock(dep);
     dldp->d_lnk = erts_lookup_link(dep->nlinks, lid);
     if (!dldp->d_lnk)
 	dldp->d_sub_lnk = NULL;
@@ -244,7 +249,7 @@ erts_remove_dist_link(ErtsDistLinkData *dldp,
 		       ? NULL
 		       : erts_remove_link(&dep->nlinks, lid));
     }
-    erts_smp_de_links_unlock(dep);
+    erts_de_links_unlock(dep);
 }
 
 ERTS_GLB_INLINE int
@@ -264,17 +269,106 @@ erts_destroy_dist_link(ErtsDistLinkData *dldp)
 
 #endif
 
+
+
+/* Define for testing */
+/* #define EXTREME_TTB_TRAPPING 1 */
+
+#ifndef EXTREME_TTB_TRAPPING
+#define TERM_TO_BINARY_LOOP_FACTOR 32
+#else
+#define TERM_TO_BINARY_LOOP_FACTOR 1
+#endif
+
+typedef enum { TTBSize, TTBEncode, TTBCompress } TTBState;
+typedef struct TTBSizeContext_ {
+    Uint flags;
+    int level;
+    Uint result;
+    Eterm obj;
+    ErtsWStack wstack;
+} TTBSizeContext;
+
+typedef struct TTBEncodeContext_ {
+    Uint flags;
+    int level;
+    byte* ep;
+    Eterm obj;
+    ErtsWStack wstack;
+    Binary *result_bin;
+} TTBEncodeContext;
+
+typedef struct {
+    Uint real_size;
+    Uint dest_len;
+    byte *dbytes;
+    Binary *result_bin;
+    Binary *destination_bin;
+    z_stream stream;
+} TTBCompressContext;
+
+typedef struct {
+    int alive;
+    TTBState state;
+    union {
+	TTBSizeContext sc;
+	TTBEncodeContext ec;
+	TTBCompressContext cc;
+    } s;
+} TTBContext;
+
+enum erts_dsig_send_phase {
+    ERTS_DSIG_SEND_PHASE_INIT,
+    ERTS_DSIG_SEND_PHASE_MSG_SIZE,
+    ERTS_DSIG_SEND_PHASE_ALLOC,
+    ERTS_DSIG_SEND_PHASE_MSG_ENCODE,
+    ERTS_DSIG_SEND_PHASE_FIN
+};
+
+struct erts_dsig_send_context {
+    enum erts_dsig_send_phase phase;
+    Sint reds;
+
+    Eterm ctl;
+    Eterm msg;
+    int force_busy;
+    Uint32 pass_through_size;
+    Uint data_size, dhdr_ext_size;
+    ErtsAtomCacheMap *acmp;
+    ErtsDistOutputBuf *obuf;
+    Uint32 flags;
+    Process *c_p;
+    union {
+	TTBSizeContext sc;
+	TTBEncodeContext ec;
+    }u;
+};
+
+typedef struct {
+    int suspend;
+
+    Eterm ctl_heap[6];
+    ErtsDSigData dsd;
+    DistEntry* dep_to_deref;
+    DistEntry *dep;
+    struct erts_dsig_send_context dss;
+
+    Eterm return_term;
+}ErtsSendContext;
+
+
 /*
  * erts_dsig_send_* return values.
  */
 #define ERTS_DSIG_SEND_OK	0
 #define ERTS_DSIG_SEND_YIELD	1
+#define ERTS_DSIG_SEND_CONTINUE 2
 
 extern int erts_dsig_send_link(ErtsDSigData *, Eterm, Eterm);
-extern int erts_dsig_send_msg(ErtsDSigData *, Eterm, Eterm);
+extern int erts_dsig_send_msg(Eterm, Eterm, ErtsSendContext*);
 extern int erts_dsig_send_exit_tt(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_unlink(ErtsDSigData *, Eterm, Eterm);
-extern int erts_dsig_send_reg_msg(ErtsDSigData *, Eterm, Eterm);
+extern int erts_dsig_send_reg_msg(Eterm, Eterm, ErtsSendContext*);
 extern int erts_dsig_send_group_leader(ErtsDSigData *, Eterm, Eterm);
 extern int erts_dsig_send_exit(ErtsDSigData *, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_exit2(ErtsDSigData *, Eterm, Eterm, Eterm);
@@ -282,6 +376,10 @@ extern int erts_dsig_send_demonitor(ErtsDSigData *, Eterm, Eterm, Eterm, int);
 extern int erts_dsig_send_monitor(ErtsDSigData *, Eterm, Eterm, Eterm);
 extern int erts_dsig_send_m_exit(ErtsDSigData *, Eterm, Eterm, Eterm, Eterm);
 
+extern int erts_dsig_send(ErtsDSigData *dsdp, struct erts_dsig_send_context* ctx);
+extern int erts_dsend_context_dtor(Binary*);
+extern Eterm erts_dsend_export_trap_context(Process* p, ErtsSendContext* ctx);
+
 extern int erts_dist_command(Port *prt, int reds);
 extern void erts_dist_port_not_busy(Port *prt);
 extern void erts_kill_dist_connection(DistEntry *dep, Uint32);