Merge branch 'rickard/dist/OTP-14459' into rickard/dist/master/OTP-14459

Conflicts:
	erts/emulator/beam/bif.c
	erts/emulator/beam/dist.c
	erts/emulator/beam/dist.h
	erts/emulator/beam/erl_bif_info.c
	erts/emulator/beam/erl_node_tables.c
	erts/emulator/beam/erl_node_tables.h
	erts/emulator/beam/external.c

1 files changed, 30 insertions, 23 deletions

diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h
index 7974b25444..3bba673435 100644
--- a/erts/emulator/beam/erl_node_tables.h
+++ b/erts/emulator/beam/erl_node_tables.h
@@ -47,6 +47,9 @@
 #define ERTS_PORT_TASK_ONLY_BASIC_TYPES__
 #include "erl_port_task.h"
 #undef ERTS_PORT_TASK_ONLY_BASIC_TYPES__
+#define ERTS_BINARY_TYPES_ONLY__
+#include "erl_binary.h"
+#undef ERTS_BINARY_TYPES_ONLY__
 
 #define ERTS_NODE_TAB_DELAY_GC_DEFAULT (60)
 #define ERTS_NODE_TAB_DELAY_GC_MAX (100*1000*1000)
@@ -60,11 +63,17 @@
 #define ERTS_DE_SFLGS_ALL			(ERTS_DE_SFLG_CONNECTED \
 						 | ERTS_DE_SFLG_EXITING)
 
-#define ERTS_DE_QFLG_BUSY			(((Uint32) 1) <<  0)
-#define ERTS_DE_QFLG_EXIT			(((Uint32) 1) <<  1)
+#define ERTS_DE_QFLG_BUSY			(((erts_aint32_t) 1) <<  0)
+#define ERTS_DE_QFLG_EXIT			(((erts_aint32_t) 1) <<  1)
+#define ERTS_DE_QFLG_REQ_INFO			(((erts_aint32_t) 1) <<  2)
+#define ERTS_DE_QFLG_PORT_CTRL                  (((erts_aint32_t) 1) <<  3)
+#define ERTS_DE_QFLG_PROC_CTRL                  (((erts_aint32_t) 1) <<  4)
 
 #define ERTS_DE_QFLGS_ALL			(ERTS_DE_QFLG_BUSY \
-						 | ERTS_DE_QFLG_EXIT)
+						 | ERTS_DE_QFLG_EXIT \
+                                                 | ERTS_DE_QFLG_REQ_INFO \
+                                                 | ERTS_DE_QFLG_PORT_CTRL \
+                                                 | ERTS_DE_QFLG_PROC_CTRL)
 
 #if defined(ARCH_64)
 #define ERTS_DIST_OUTPUT_BUF_DBG_PATTERN ((Uint) 0xf713f713f713f713UL)
@@ -106,12 +115,13 @@ typedef struct dist_entry_ {
     HashBucket hash_bucket;     /* Hash bucket */
     struct dist_entry_ *next;	/* Next entry in dist_table (not sorted) */
     struct dist_entry_ *prev;	/* Previous entry in dist_table (not sorted) */
-    erts_refc_t refc;		/* Reference count */
 
-    erts_rwmtx_t rwmtx;     /* Protects all fields below until lck_mtx. */
+    erts_rwmtx_t rwmtx;         /* Protects all fields below until lck_mtx. */
     Eterm sysname;		/* name@host atom for efficiency */
     Uint32 creation;		/* creation of connected node */
-    Eterm cid;			/* connection handler (pid or port), NIL == free */
+    erts_atomic_t input_handler; /* Input handler */
+    Eterm cid;			/* connection handler (pid or port),
+                                   NIL == free */
     Uint32 connection_id;	/* Connection id incremented on connect */
     Uint32 status;		/* Slot status, like exiting reserved etc */
     Uint32 flags;		/* Distribution flags, like hidden, 
@@ -119,7 +129,7 @@ typedef struct dist_entry_ {
     unsigned long version;	/* Protocol version */
 
 
-    erts_mtx_t lnk_mtx;     /* Protects node_links, nlinks, and
+    erts_mtx_t lnk_mtx;         /* Protects node_links, nlinks, and
 				   monitors. */
     ErtsLink *node_links;       /* In a dist entry, node links are kept 
 				   in a separate tree, while they are 
@@ -131,12 +141,15 @@ typedef struct dist_entry_ {
     ErtsLink *nlinks;           /* Link tree with subtrees */
     ErtsMonitor *monitors;      /* Monitor tree */
 
-    erts_mtx_t qlock;       /* Protects qflgs and out_queue */
-    Uint32 qflgs;
-    Sint qsize;
+    erts_mtx_t qlock;           /* Protects qflgs and out_queue */
+    erts_atomic32_t qflgs;
+    erts_atomic_t qsize;
+    erts_atomic64_t in;
+    erts_atomic64_t out;
     ErtsDistOutputQueue out_queue;
     struct ErtsProcList_ *suspended;
 
+    ErtsDistOutputQueue tmp_out_queue;
     ErtsDistOutputQueue finalized_out_queue;
     erts_atomic_t dist_cmd_scheduled;
     ErtsPortTaskHandle dist_cmd;
@@ -144,6 +157,8 @@ typedef struct dist_entry_ {
     Uint (*send)(Port *prt, ErtsDistOutputBuf *obuf);
 
     struct cache* cache;	/* The atom cache */
+
+    ErtsThrPrgrLaterOp later_op;
 } DistEntry;
 
 typedef struct erl_node_ {
@@ -193,12 +208,12 @@ Eterm erts_get_node_and_dist_references(struct process *);
 int erts_lc_is_de_rwlocked(DistEntry *);
 int erts_lc_is_de_rlocked(DistEntry *);
 #endif
+int erts_dist_entry_destructor(Binary *bin);
+DistEntry *erts_dhandle_to_dist_entry(Eterm dhandle);
+Eterm erts_make_dhandle(Process *c_p, DistEntry *dep);
+void erts_ref_dist_entry(DistEntry *dep);
+void erts_deref_dist_entry(DistEntry *dep);
 
-#ifdef ERTS_ENABLE_LOCK_COUNT
-void erts_lcnt_update_distribution_locks(int enable);
-#endif
-
-ERTS_GLB_INLINE void erts_deref_dist_entry(DistEntry *dep);
 ERTS_GLB_INLINE void erts_deref_node_entry(ErlNode *np);
 ERTS_GLB_INLINE void erts_de_rlock(DistEntry *dep);
 ERTS_GLB_INLINE void erts_de_runlock(DistEntry *dep);
@@ -210,14 +225,6 @@ ERTS_GLB_INLINE void erts_de_links_unlock(DistEntry *dep);
 #if ERTS_GLB_INLINE_INCL_FUNC_DEF
 
 ERTS_GLB_INLINE void
-erts_deref_dist_entry(DistEntry *dep)
-{
-    ASSERT(dep);
-    if (erts_refc_dectest(&dep->refc, 0) == 0)
-	erts_schedule_delete_dist_entry(dep);
-}
-
-ERTS_GLB_INLINE void
 erts_deref_node_entry(ErlNode *np)
 {
     ASSERT(np);