erts: Fix bug in setnode/2

that could lead strange things to happen when renaming a node with
a name that already exist in node and dist tables.

Problem:
erts_set_this_node() is a bit brutal and does not handle the case
when an old remote node name is reused as the new local node name.

Solution:
Treat erts_this_node and erts_this_dist_entry as all the others
with correct reference counting.

4 files changed, 95 insertions, 111 deletions

diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c
index 4e3a1cef69..6c61b0de6b 100644
--- a/erts/emulator/beam/bif.c
+++ b/erts/emulator/beam/bif.c
@@ -4231,6 +4231,7 @@ BIF_RETTYPE list_to_pid_1(BIF_ALIST_1)
 	  goto bad;
       
       enp = erts_find_or_insert_node(dep->sysname, dep->creation);
+      ASSERT(enp != erts_this_node);
 
       etp = (ExternalThing *) HAlloc(BIF_P, EXTERNAL_THING_HEAD_SIZE + 1);
       etp->header = make_external_pid_header(1);
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 5a0f8388f8..56a8633f85 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -2572,7 +2572,9 @@ int distribution_info(int to, void *arg)	/* Called by break handler */
     }
 
     for (dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
-	info_dist_entry(to, arg, dep, 0, 0);
+        if (dep != erts_this_dist_entry) {
+            info_dist_entry(to, arg, dep, 0, 0);
+        }
     }
 
     return(0);
@@ -3012,11 +3014,11 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
 
     erts_smp_rwmtx_rlock(&erts_dist_table_rwmtx);
 
-    ASSERT(erts_no_of_not_connected_dist_entries >= 0);
+    ASSERT(erts_no_of_not_connected_dist_entries > 0);
     ASSERT(erts_no_of_hidden_dist_entries >= 0);
     ASSERT(erts_no_of_visible_dist_entries >= 0);
     if(not_connected)
-      length += erts_no_of_not_connected_dist_entries;
+      length += (erts_no_of_not_connected_dist_entries - 1);
     if(hidden)
       length += erts_no_of_hidden_dist_entries;
     if(visible)
@@ -3038,8 +3040,10 @@ BIF_RETTYPE nodes_1(BIF_ALIST_1)
 #endif
     if(not_connected)
       for(dep = erts_not_connected_dist_entries; dep; dep = dep->next) {
-	result = CONS(hp, dep->sysname, result);
-	hp += 2;
+          if (dep != erts_this_dist_entry) {
+            result = CONS(hp, dep->sysname, result);
+            hp += 2;
+          }
       }
     if(hidden)
       for(dep = erts_hidden_dist_entries; dep; dep = dep->next) {
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index 865de1726d..a7d0511bf9 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -37,18 +37,18 @@ erts_smp_rwmtx_t erts_node_table_rwmtx;
 
 DistEntry *erts_hidden_dist_entries;
 DistEntry *erts_visible_dist_entries;
-DistEntry *erts_not_connected_dist_entries;
+DistEntry *erts_not_connected_dist_entries; /* including erts_this_dist_entry */
 Sint erts_no_of_hidden_dist_entries;
 Sint erts_no_of_visible_dist_entries;
-Sint erts_no_of_not_connected_dist_entries;
+Sint erts_no_of_not_connected_dist_entries; /* including erts_this_dist_entry */
 
 DistEntry *erts_this_dist_entry;
 ErlNode *erts_this_node;
 char erts_this_node_sysname_BUFFER[256],
     *erts_this_node_sysname = "uninitialized yet";
 
-static Uint node_entries;
-static Uint dist_entries;
+static Uint node_entries = 0;
+static Uint dist_entries = 0;
 
 static int references_atoms_need_init = 1;
 
@@ -91,9 +91,6 @@ dist_table_alloc(void *dep_tmpl)
     erts_smp_rwmtx_opt_t rwmtx_opt = ERTS_SMP_RWMTX_OPT_DEFAULT_INITER;
     rwmtx_opt.type = ERTS_SMP_RWMTX_TYPE_FREQUENT_READ;
 
-    if(((DistEntry *) dep_tmpl) == erts_this_dist_entry)
-	return dep_tmpl;
-
     sysname = ((DistEntry *) dep_tmpl)->sysname;
     chnl_nr = make_small((Uint) atom_val(sysname));
     dep = (DistEntry *) erts_alloc(ERTS_ALC_T_DIST_ENTRY, sizeof(DistEntry));
@@ -132,7 +129,9 @@ dist_table_alloc(void *dep_tmpl)
 
     /* Link in */
 
-    /* All new dist entries are "not connected" */
+    /* All new dist entries are "not connected".
+     * erts_this_dist_entry is also always included among "not connected"
+     */
     dep->next = erts_not_connected_dist_entries;
     if(erts_not_connected_dist_entries) {
 	ASSERT(erts_not_connected_dist_entries->prev == NULL);
@@ -149,9 +148,6 @@ dist_table_free(void *vdep)
 {
     DistEntry *dep = (DistEntry *) vdep;
 
-    if(dep == erts_this_dist_entry)
-	return;
-
     ASSERT(is_nil(dep->cid));
     ASSERT(dep->nlinks == NULL);
     ASSERT(dep->node_links == NULL);
@@ -186,7 +182,7 @@ dist_table_free(void *vdep)
 #endif
     erts_free(ERTS_ALC_T_DIST_ENTRY, (void *) dep);
 
-    ASSERT(dist_entries > 1);
+    ASSERT(dist_entries > 0);
     dist_entries--;
 }
 
@@ -306,7 +302,7 @@ static void try_delete_dist_entry(void *vdep)
      * thread incremented refc twice. Once for the new reference
      * and once for this thread.
      *
-     * If refc reach -1, noone has used the entry since we
+     * If refc reach -1, no one has used the entry since we
      * set up the timer. Delete the entry.
      *
      * If refc reach 0, the entry is currently not in use
@@ -369,8 +365,7 @@ erts_dist_table_size(void)
 
     ASSERT(dist_entries == (erts_no_of_visible_dist_entries
 			    + erts_no_of_hidden_dist_entries
-			    + erts_no_of_not_connected_dist_entries
-			    + 1 /* erts_this_dist_entry */));
+			    + erts_no_of_not_connected_dist_entries));
 #endif
 
     res = (hash_table_sz(&erts_dist_table)
@@ -543,9 +538,6 @@ node_table_alloc(void *venp_tmpl)
 {
     ErlNode *enp;
 
-    if(((ErlNode *) venp_tmpl) == erts_this_node)
-	return venp_tmpl;
-
     enp = (ErlNode *) erts_alloc(ERTS_ALC_T_NODE_ENTRY, sizeof(ErlNode));
 
     node_entries++;
@@ -563,8 +555,7 @@ node_table_free(void *venp)
 {
     ErlNode *enp = (ErlNode *) venp;
 
-    if(enp == erts_this_node)
-	return;
+    ERTS_SMP_LC_ASSERT(enp != erts_this_node || erts_thr_progress_is_blocking());
 
     erts_deref_dist_entry(enp->dist_entry);
 #ifdef DEBUG
@@ -572,7 +563,7 @@ node_table_free(void *venp)
 #endif
     erts_free(ERTS_ALC_T_NODE_ENTRY, venp);
 
-    ASSERT(node_entries > 1);
+    ASSERT(node_entries > 0);
     node_entries--;
 }
 
@@ -650,7 +641,7 @@ static void try_delete_node(void *venp)
      * thread incremented refc twice. Once for the new reference
      * and once for this thread.
      *
-     * If refc reach -1, noone has used the entry since we
+     * If refc reach -1, no one has used the entry since we
      * set up the timer. Delete the entry.
      *
      * If refc reach 0, the entry is currently not in use
@@ -747,25 +738,20 @@ void erts_print_node_info(int to,
 void
 erts_set_this_node(Eterm sysname, Uint creation)
 {
-    erts_smp_rwmtx_rwlock(&erts_node_table_rwmtx);
-    erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+    ERTS_SMP_LC_ASSERT(erts_thr_progress_is_blocking());
+    ASSERT(erts_refc_read(&erts_this_dist_entry->refc, 2));
 
-    (void) hash_erase(&erts_dist_table, (void *) erts_this_dist_entry);
-    erts_this_dist_entry->sysname = sysname;
-    erts_this_dist_entry->creation = creation;
-    (void) hash_put(&erts_dist_table, (void *) erts_this_dist_entry);
+    if (erts_refc_dectest(&erts_this_node->refc, 0) == 0)
+        try_delete_node(erts_this_node);
 
-    (void) hash_erase(&erts_node_table, (void *) erts_this_node);
-    erts_this_node->sysname = sysname;
-    erts_this_node->creation = creation;
-    erts_this_node_sysname = erts_this_node_sysname_BUFFER;
-    erts_snprintf(erts_this_node_sysname, sizeof(erts_this_node_sysname_BUFFER),
-                  "%T", sysname);
-    (void) hash_put(&erts_node_table, (void *) erts_this_node);
+    if (erts_refc_dectest(&erts_this_dist_entry->refc, 0) == 0)
+        try_delete_dist_entry(erts_this_dist_entry);
 
-    erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
-    erts_smp_rwmtx_rwunlock(&erts_node_table_rwmtx);
+    erts_this_node = NULL; /* to make sure refc is bumped for this node */
+    erts_this_node = erts_find_or_insert_node(sysname, creation);
+    erts_this_dist_entry = erts_this_node->dist_entry;
 
+    erts_refc_inc(&erts_this_dist_entry->refc, 2);
 }
 
 Uint
@@ -782,6 +768,7 @@ void erts_init_node_tables(int dd_sec)
 {
     erts_smp_rwmtx_opt_t rwmtx_opt = ERTS_SMP_RWMTX_OPT_DEFAULT_INITER;
     HashFunctions f;
+    ErlNode node_tmpl;
 
     if (dd_sec == ERTS_NODE_TAB_DELAY_GC_INFINITY)
 	node_tab_delete_delay = (ErtsMonotonicTime) -1;
@@ -793,16 +780,21 @@ void erts_init_node_tables(int dd_sec)
     rwmtx_opt.type = ERTS_SMP_RWMTX_TYPE_FREQUENT_READ;
     rwmtx_opt.lived = ERTS_SMP_RWMTX_LONG_LIVED;
 
+    erts_smp_rwmtx_init_opt(&erts_node_table_rwmtx, &rwmtx_opt, "node_table");
+    erts_smp_rwmtx_init_opt(&erts_dist_table_rwmtx, &rwmtx_opt, "dist_table");
+
     f.hash  = (H_FUN)			dist_table_hash;
     f.cmp   = (HCMP_FUN)		dist_table_cmp;
     f.alloc = (HALLOC_FUN)		dist_table_alloc;
     f.free  = (HFREE_FUN)		dist_table_free;
-
-    erts_this_dist_entry = erts_alloc(ERTS_ALC_T_DIST_ENTRY, sizeof(DistEntry));
-    dist_entries = 1;
-
     hash_init(ERTS_ALC_T_DIST_TABLE, &erts_dist_table, "dist_table", 11, f);
 
+    f.hash  = (H_FUN)      			node_table_hash;
+    f.cmp   = (HCMP_FUN)   			node_table_cmp;
+    f.alloc = (HALLOC_FUN) 			node_table_alloc;
+    f.free  = (HFREE_FUN)  			node_table_free;
+    hash_init(ERTS_ALC_T_NODE_TABLE, &erts_node_table, "node_table", 11, f);
+
     erts_hidden_dist_entries				= NULL;
     erts_visible_dist_entries				= NULL;
     erts_not_connected_dist_entries			= NULL;
@@ -810,69 +802,23 @@ void erts_init_node_tables(int dd_sec)
     erts_no_of_visible_dist_entries			= 0;
     erts_no_of_not_connected_dist_entries		= 0;
 
-    erts_this_dist_entry->next				= NULL;
-    erts_this_dist_entry->prev				= NULL;
-    erts_refc_init(&erts_this_dist_entry->refc, 1); /* erts_this_node */
-
-    erts_smp_rwmtx_init_opt_x(&erts_this_dist_entry->rwmtx,
-			      &rwmtx_opt,
-			      "dist_entry",
-			      make_small(ERST_INTERNAL_CHANNEL_NO));
-    erts_this_dist_entry->sysname			= am_Noname;
-    erts_this_dist_entry->cid				= NIL;
-    erts_this_dist_entry->connection_id			= 0;
-    erts_this_dist_entry->status			= 0;
-    erts_this_dist_entry->flags				= 0;
-    erts_this_dist_entry->version			= 0;
-
-    erts_smp_mtx_init_x(&erts_this_dist_entry->lnk_mtx,
-			"dist_entry_links",
-			make_small(ERST_INTERNAL_CHANNEL_NO));
-    erts_this_dist_entry->node_links			= NULL;
-    erts_this_dist_entry->nlinks			= NULL;
-    erts_this_dist_entry->monitors			= NULL;
-
-    erts_smp_mtx_init_x(&erts_this_dist_entry->qlock,
-			"dist_entry_out_queue",
-			make_small(ERST_INTERNAL_CHANNEL_NO));
-    erts_this_dist_entry->qflgs				= 0;
-    erts_this_dist_entry->qsize				= 0;
-    erts_this_dist_entry->out_queue.first		= NULL;
-    erts_this_dist_entry->out_queue.last		= NULL;
-    erts_this_dist_entry->suspended			= NULL;
-
-    erts_this_dist_entry->finalized_out_queue.first	= NULL;
-    erts_this_dist_entry->finalized_out_queue.last	= NULL;
-    erts_smp_atomic_init_nob(&erts_this_dist_entry->dist_cmd_scheduled, 0);
-    erts_port_task_handle_init(&erts_this_dist_entry->dist_cmd);
-    erts_this_dist_entry->send				= NULL;
-    erts_this_dist_entry->cache				= NULL;
-
-    (void) hash_put(&erts_dist_table, (void *) erts_this_dist_entry);
+    node_tmpl.sysname = am_Noname;
+    node_tmpl.creation = 0;
+    erts_this_node = hash_put(&erts_node_table, &node_tmpl);
+     /* +1 for erts_this_node */
+    erts_refc_init(&erts_this_node->refc, 1);
 
-    f.hash  = (H_FUN)      			node_table_hash;
-    f.cmp   = (HCMP_FUN)   			node_table_cmp;
-    f.alloc = (HALLOC_FUN) 			node_table_alloc;
-    f.free  = (HFREE_FUN)  			node_table_free;
-
-    hash_init(ERTS_ALC_T_NODE_TABLE, &erts_node_table, "node_table", 11, f);
+    ASSERT(erts_this_node->dist_entry != NULL);
+    erts_this_dist_entry = erts_this_node->dist_entry;
+    /* +1 for erts_this_dist_entry */
+    /* +1 for erts_this_node->dist_entry */
+    erts_refc_init(&erts_this_dist_entry->refc, 2);
 
-    erts_this_node = erts_alloc(ERTS_ALC_T_NODE_ENTRY, sizeof(ErlNode));
-    node_entries = 1;
 
-    erts_refc_init(&erts_this_node->refc, 1); /* The system itself */
-    erts_this_node->sysname			= am_Noname;
-    erts_this_node->creation			= 0;
-    erts_this_node->dist_entry			= erts_this_dist_entry;
     erts_this_node_sysname = erts_this_node_sysname_BUFFER;
     erts_snprintf(erts_this_node_sysname, sizeof(erts_this_node_sysname_BUFFER),
                   "%T", erts_this_node->sysname);
 
-    (void) hash_put(&erts_node_table, (void *) erts_this_node);
-
-    erts_smp_rwmtx_init_opt(&erts_node_table_rwmtx, &rwmtx_opt, "node_table");
-    erts_smp_rwmtx_init_opt(&erts_dist_table_rwmtx, &rwmtx_opt, "dist_table");
-
     references_atoms_need_init = 1;
 }
 
@@ -1410,6 +1356,10 @@ setup_reference_table(void)
 		SYSTEM_REF,
 		TUPLE2(&heap[0], AM_system, am_undefined));
 
+    insert_dist_entry(erts_this_dist_entry,
+                      SYSTEM_REF,
+                      TUPLE2(&heap[0], AM_system, am_undefined),
+                      erts_this_node->creation);
     UnUseTmpHeapNoproc(3);
 
     max = erts_ptab_max(&erts_proc);
diff --git a/erts/emulator/test/distribution_SUITE_data/run.erl b/erts/emulator/test/distribution_SUITE_data/run.erl
index f5169e160c..d5ed139369 100644
--- a/erts/emulator/test/distribution_SUITE_data/run.erl
+++ b/erts/emulator/test/distribution_SUITE_data/run.erl
@@ -30,16 +30,19 @@ from(H, [_ | T]) -> from(H, T);
 from(H, []) -> [].
 
 start() ->
-    net_kernel:start([fideridum,shortnames]),
-    {ok, Node} = slave:start(host(), heppel),
-    P = spawn(Node, a, b, []),
-    B1 = term_to_binary(P),
-    N1 = node(P),
-    ok = net_kernel:stop(),
-    N2 = node(P),
-    io:format("~w~n", [N1 == N2]),
+    Result = do_it(),
+
+    %% Do GCs and node_and_dist_references
+    %% in an attempt to crash the VM (without OTP-13076 fix)
+    lists:foreach(fun(P) -> erlang:garbage_collect(P) end,
+		  processes()),
+    erts_debug:set_internal_state(available_internal_state, true),
+    erts_debug:get_internal_state(node_and_dist_references),
+
+    io:format("~w~n", [Result]),
+
     if
-	N1 == N2 -> 
+	Result ->
 	    init:stop();
 	true ->
 	    %% Make sure that the io:format/2 output is really written
@@ -47,3 +50,29 @@ start() ->
 	    erlang:yield(),
 	    init:stop()
     end.
+
+
+do_it() ->
+    {ok, _} = net_kernel:start([fideridum,shortnames]),
+    {ok, Node} = slave:start(host(), heppel),
+    P = spawn(Node, net_kernel, stop, []),
+    B1 = term_to_binary(P),
+    N1 = node(P),
+    ok = net_kernel:stop(),
+    N2 = node(P),
+
+    %% OTP-13076
+    %% Restart distribution with same node name as previous remote node
+    %% Repeat to wrap around creation
+    Result = lists:foldl(fun(_, Acc) ->
+				 timer:sleep(2),  % give net_kernel:stop() time to take effect :-(
+				 {ok, _} = net_kernel:start([heppel,shortnames]),
+				 N3 = node(P),
+				 ok = net_kernel:stop(),
+				 N4 = node(P),
+				 Acc and (N3 =:= N1) and (N4 =:= N1)
+			 end,
+			 (N2 =:= N1),
+			 lists:seq(1,3)),
+
+    Result.