Consolidate distribution entry state transitions

* Make connection_id part of the distribution handle as {ConnId, DistEntry}
  in order for BIFs to verify correct connection.
* Make distribution handle opaque to net_kernel.
* Remove some unsafe lockless reads of DistEntry.flags
* Change state ERTS_DE_STATE_EXITING to be more of an internal state that
  prevents erts from enqueue, encode or schedule new data to be sent. Otherwise
  it should behave like ERTS_DE_STATE_CONNECTED.

1 files changed, 26 insertions, 21 deletions

diff --git a/erts/emulator/beam/external.c b/erts/emulator/beam/external.c
index 904993ceb6..1f6d3ef031 100644
--- a/erts/emulator/beam/external.c
+++ b/erts/emulator/beam/external.c
@@ -671,8 +671,8 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 		      byte *ext,
 		      Uint size,
 		      DistEntry *dep,
-		      ErtsAtomCache *cache,
-                      Uint32 *connection_id)
+                      Uint32 conn_id,
+		      ErtsAtomCache *cache)
 {
 #undef ERTS_EXT_FAIL
 #undef ERTS_EXT_HDR_FAIL
@@ -683,18 +683,34 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 #define ERTS_EXT_FAIL abort()
 #define ERTS_EXT_HDR_FAIL abort()
 #endif
+    register byte *ep;
+
+    edep->heap_size = -1;
+    edep->flags = 0;
+    edep->dep = dep;
+
+    ASSERT(dep);
+    erts_de_rlock(dep);
 
-    register byte *ep = ext;
     ASSERT(dep->flags & DFLAG_UTF8_ATOMS);
 
-    edep->heap_size = -1;
-    edep->ext_endp = ext+size;
+    if ((dep->state != ERTS_DE_STATE_CONNECTED &&
+         dep->state != ERTS_DE_STATE_PENDING)
+        || dep->connection_id != conn_id) {
+        erts_de_runlock(dep);
+        return ERTS_PREP_DIST_EXT_CLOSED;
+    }
 
-    if (size < 2)
-	ERTS_EXT_FAIL;
+    if (!(dep->flags & DFLAG_DIST_HDR_ATOM_CACHE)) {
+        /* Skip PASS_THROUGH */
+        ext++;
+        size--;
+    }
+    edep->ext_endp = ext + size;
+    ep = ext;
 
-    if (!dep)
-        ERTS_INTERNAL_ERROR("Invalid use");
+    if (size < 2)
+        goto fail;
 
     if (ep[0] != VERSION_MAGIC) {
 	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
@@ -706,20 +722,9 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
 	ERTS_EXT_FAIL;
     }
 
-    edep->flags = 0;
-    edep->dep = dep;
-
-    erts_de_rlock(dep);
-
-    if (dep->state != ERTS_DE_STATE_CONNECTED &&
-	dep->state != ERTS_DE_STATE_PENDING) {
-        erts_de_runlock(dep);
-        return ERTS_PREP_DIST_EXT_CLOSED;
-    }
     if (dep->flags & DFLAG_DIST_HDR_ATOM_CACHE)
         edep->flags |= ERTS_DIST_EXT_DFLAG_HDR;
 
-    *connection_id = dep->connection_id;
     edep->connection_id = dep->connection_id;
 
     if (ep[1] != DIST_HEADER) {
@@ -901,7 +906,7 @@ erts_prepare_dist_ext(ErtsDistExternal *edep,
     }
  fail: {
 	erts_de_runlock(dep);
-	erts_kill_dist_connection(dep, *connection_id);
+	erts_kill_dist_connection(dep, conn_id);
     }
     return ERTS_PREP_DIST_EXT_FAILED;
 }