1 files changed, 87 insertions, 47 deletions

diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index 3e17645997..ea4697815f 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -44,6 +44,21 @@
 #define DFLAG_UTF8_ATOMS          0x10000
 #define DFLAG_MAP_TAG             0x20000
 #define DFLAG_BIG_CREATION        0x40000
+#define DFLAG_SEND_SENDER         0x80000
+#define DFLAG_NO_MAGIC            0x100000
+
+/* Mandatory flags for distribution (sync with dist_util.erl) */
+#define DFLAG_DIST_MANDATORY (DFLAG_EXTENDED_REFERENCES         \
+                              | DFLAG_EXTENDED_PIDS_PORTS       \
+			      | DFLAG_UTF8_ATOMS                \
+			      | DFLAG_NEW_FUN_TAGS)
+
+/* Additional optimistic flags when encoding toward pending connection */
+#define DFLAG_DIST_HOPEFULLY (DFLAG_NO_MAGIC                    \
+                              | DFLAG_EXPORT_PTR_TAG            \
+                              | DFLAG_BIT_BINARIES              \
+                              | DFLAG_DIST_MONITOR              \
+                              | DFLAG_DIST_MONITOR_NAME)
 
 /* All flags that should be enabled when term_to_binary/1 is used. */
 #define TERM_TO_BINARY_DFLAGS (DFLAG_EXTENDED_REFERENCES	\
@@ -74,33 +89,30 @@
 #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;
-extern Export* dlink_trap;
-extern Export* dunlink_trap;
 extern Export* dmonitor_node_trap;
-extern Export* dgroup_leader_trap;
-extern Export* dexit_trap;
 extern Export* dmonitor_p_trap;
 
 typedef enum {
     ERTS_DSP_NO_LOCK,
-    ERTS_DSP_RLOCK,
-    ERTS_DSP_RWLOCK
+    ERTS_DSP_RLOCK
 } ErtsDSigPrepLock;
 
 
 typedef struct {
     Process *proc;
     DistEntry *dep;
+    Eterm node;   /* used if dep == NULL */
     Eterm cid;
     Eterm connection_id;
     int no_suspend;
 } 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)))
 
@@ -113,14 +125,10 @@ 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 appropriate actions
- * have been taken. Appropriate actions would typically be setting
- * up the connection.
+ * One of the values defined below are returned.
  */
 
-/* Connected; signal can be sent. */
+/* Connected; signals can be enqueued and sent. */
 #define ERTS_DSIG_PREP_CONNECTED	0
 /* Not connected; connection needs to be set up. */
 #define ERTS_DSIG_PREP_NOT_CONNECTED	1
@@ -128,46 +136,80 @@ extern int erts_is_alive;
 #define ERTS_DSIG_PREP_WOULD_SUSPEND	2
 /* System not alive (distributed) */
 #define ERTS_DSIG_PREP_NOT_ALIVE	3
+/* Pending connection; signals can be enqueued */
+#define ERTS_DSIG_PREP_PENDING	        4
 
 ERTS_GLB_INLINE int erts_dsig_prepare(ErtsDSigData *,
-				      DistEntry *,
+				      DistEntry*,
 				      Process *,
+                                      ErtsProcLocks,
 				      ErtsDSigPrepLock,
+				      int,
 				      int);
 
 ERTS_GLB_INLINE
 void erts_schedule_dist_command(Port *, DistEntry *);
 
+int erts_auto_connect(DistEntry* dep, Process *proc, ErtsProcLocks proc_locks);
+
 #if ERTS_GLB_INLINE_INCL_FUNC_DEF
 
 ERTS_GLB_INLINE int 
 erts_dsig_prepare(ErtsDSigData *dsdp,
 		  DistEntry *dep,
 		  Process *proc,
+                  ErtsProcLocks proc_locks,
 		  ErtsDSigPrepLock dspl,
-		  int no_suspend)
+		  int no_suspend,
+		  int connect)
 {
-    int failure;
+    int res;
+
     if (!erts_is_alive)
 	return ERTS_DSIG_PREP_NOT_ALIVE;
-    if (!dep)
-	return ERTS_DSIG_PREP_NOT_CONNECTED;
-    if (dspl == ERTS_DSP_RWLOCK)
-	erts_smp_de_rwlock(dep);
-    else
-	erts_smp_de_rlock(dep);
-    if (ERTS_DE_IS_NOT_CONNECTED(dep)) {
-	failure = ERTS_DSIG_PREP_NOT_CONNECTED;
+    if (!dep) {
+        ASSERT(!connect);
+        return ERTS_DSIG_PREP_NOT_CONNECTED;
+    }
+
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    if (connect) {
+        erts_proc_lc_might_unlock(proc, proc_locks);
+    }
+#endif
+
+retry:
+    erts_de_rlock(dep);
+
+    if (ERTS_DE_IS_CONNECTED(dep)) {
+	res = ERTS_DSIG_PREP_CONNECTED;
+    }
+    else if (dep->status & ERTS_DE_SFLG_PENDING) {
+	res = ERTS_DSIG_PREP_PENDING;
+    }
+    else if (dep->status & ERTS_DE_SFLG_EXITING) {
+	res = ERTS_DSIG_PREP_NOT_CONNECTED;
 	goto fail;
     }
+    else if (connect) {
+        ASSERT(dep->status == 0);
+        erts_de_runlock(dep);
+        if (!erts_auto_connect(dep, proc, proc_locks)) {
+            return ERTS_DSIG_PREP_NOT_ALIVE;
+        }
+	goto retry;
+    }
+    else {
+        ASSERT(dep->status == 0);
+	res = 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)
-	    failure = ERTS_DSIG_PREP_WOULD_SUSPEND;
-	erts_smp_mtx_unlock(&dep->qlock);
-	if (failure == ERTS_DSIG_PREP_WOULD_SUSPEND)
+	if (erts_atomic32_read_acqb(&dep->qflgs) & ERTS_DE_QFLG_BUSY) {
+	    res = ERTS_DSIG_PREP_WOULD_SUSPEND;
 	    goto fail;
+	}
     }
     dsdp->proc = proc;
     dsdp->dep = dep;
@@ -175,16 +217,12 @@ 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);
-    return ERTS_DSIG_PREP_CONNECTED;
+	erts_de_runlock(dep);
+    return res;
 
  fail:
-    if (dspl == ERTS_DSP_RWLOCK)
-	erts_smp_de_rwunlock(dep);
-    else
-	erts_smp_de_runlock(dep);
-    return failure;
-
+    erts_de_runlock(dep);
+    return res;
 }
 
 ERTS_GLB_INLINE
@@ -194,7 +232,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);
@@ -204,7 +242,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));
 
@@ -212,7 +250,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);
 }
 
@@ -238,7 +276,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;
@@ -248,7 +286,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
@@ -331,7 +369,7 @@ struct erts_dsig_send_context {
     Eterm ctl;
     Eterm msg;
     int force_busy;
-    Uint32 pass_through_size;
+    Uint32 max_finalize_prepend;
     Uint data_size, dhdr_ext_size;
     ErtsAtomCacheMap *acmp;
     ErtsDistOutputBuf *obuf;
@@ -345,10 +383,12 @@ struct erts_dsig_send_context {
 
 typedef struct {
     int suspend;
+    int connect;
 
     Eterm ctl_heap[6];
     ErtsDSigData dsd;
-    DistEntry* dep_to_deref;
+    DistEntry *dep;
+    int deref_dep;
     struct erts_dsig_send_context dss;
 
     Eterm return_term;