Merge branch 'rickard/delayed-delete-node/OTP-12802'

* rickard/delayed-delete-node/OTP-12802:
  Delayed node table GC

11 files changed, 288 insertions, 59 deletions

diff --git a/erts/doc/src/erl.xml b/erts/doc/src/erl.xml
index 98d05dc7de..f41b6e6149 100644
--- a/erts/doc/src/erl.xml
+++ b/erts/doc/src/erl.xml
@@ -1351,6 +1351,18 @@
             give lower latency and higher throughput at the expense
             of higher memory usage.</p>
           </item>
+          <tag><marker id="+zdntgc"><c>+zdntgc time</c></marker></tag>
+          <item>
+            <p>Set the delayed node table garbage collection time
+	    (<seealso marker="erlang#system_info_delayed_node_table_gc">delayed_node_table_gc</seealso>)
+	    in seconds. Valid values are either <c>infinity</c> or
+	    an integer in the range [0-100000000]. Default is 60.</p>
+            <p>Node table entries that are not referred will linger
+	    in the table for at least the amount of time that this
+	    parameter determines. The lingering prevents repeated
+	    deletions and insertions in the tables from occurring.
+	    </p>
+          </item>
         </taglist>
       </item>
     </taglist>
diff --git a/erts/doc/src/erlang.xml b/erts/doc/src/erlang.xml
index 3fea64cef5..50a26781c4 100644
--- a/erts/doc/src/erlang.xml
+++ b/erts/doc/src/erlang.xml
@@ -6222,6 +6222,7 @@ ok
       <name name="system_info" arity="1" clause_i="64"/>
       <name name="system_info" arity="1" clause_i="65"/>
       <name name="system_info" arity="1" clause_i="66"/>
+      <name name="system_info" arity="1" clause_i="67"/>
       <fsummary>Information about the system</fsummary>
       <desc>
         <p>Returns various information about the current system
@@ -6291,6 +6292,15 @@ ok
 	       compiled; otherwise, <c>false</c>.
 	    </p>
           </item>
+          <tag><marker id="system_info_delayed_node_table_gc"><c>delayed_node_table_gc</c></marker></tag>
+          <item>
+            <p>Returns the amount of time in seconds that garbage collection
+	    of an entry in a node table will be delayed. This limit can be set
+	    on startup by passing the
+	    <seealso marker="erts:erl#+zdntgc">+zdntgc</seealso> command line
+	    flag to <c>erl</c>. For more information see the documentation of the
+	    command line flag.</p>
+          </item>
           <tag><marker id="system_info_dirty_cpu_schedulers"><c>dirty_cpu_schedulers</c></marker></tag>
           <item>
             <p>Returns the number of dirty CPU scheduler threads used by
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 142fcb3c00..cfdede793c 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -2522,7 +2522,7 @@ info_dist_entry(int to, void *arg, DistEntry *dep, int visible, int connected)
 
   erts_print(to, arg, "Name: %T", dep->sysname);
 #ifdef DEBUG
-  erts_print(to, arg, " (refc=%d)", erts_refc_read(&dep->refc, 1));
+  erts_print(to, arg, " (refc=%d)", erts_refc_read(&dep->refc, 0));
 #endif
   erts_print(to, arg, "\n");
   if (!connected && is_nil(dep->cid)) {
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index ae8a5c266a..23fc4f915e 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -2687,6 +2687,15 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
 	hp = hsz ? HAlloc(BIF_P, hsz) : NULL;
 	res = erts_bld_uint(&hp, NULL, erts_dist_buf_busy_limit);
 	BIF_RET(res);
+    } else if (ERTS_IS_ATOM_STR("delayed_node_table_gc", BIF_ARG_1)) {
+	Uint hsz = 0;
+	Uint dntgc = erts_delayed_node_table_gc();
+	if (dntgc == ERTS_NODE_TAB_DELAY_GC_INFINITY)
+	    BIF_RET(am_infinity);
+ 	(void) erts_bld_uint(NULL, &hsz, dntgc);
+	hp = hsz ? HAlloc(BIF_P, hsz) : NULL;
+	res = erts_bld_uint(&hp, NULL, dntgc);
+	BIF_RET(res);
     } else if (ERTS_IS_ATOM_STR("ethread_info", BIF_ARG_1)) {
 	BIF_RET(erts_get_ethread_info(BIF_P));
     }
@@ -4069,6 +4078,17 @@ BIF_RETTYPE erts_debug_set_internal_state_2(BIF_ALIST_2)
 	    int res = erts_debug_set_unique_monotonic_integer_state(BIF_ARG_2);
 	    BIF_RET(res ? am_true : am_false);
 	}
+	else if (ERTS_IS_ATOM_STR("node_tab_delayed_delete", BIF_ARG_1)) {
+	    /* node_container_SUITE */
+	    Sint64 msecs;
+	    if (term_to_Sint64(BIF_ARG_2, &msecs)) {
+		/* Negative value restore original value... */
+		erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN);
+		erts_debug_test_node_tab_delayed_delete(msecs);
+		erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN);
+		BIF_RET(am_ok);
+	    }
+	}
     }
 
     BIF_ERROR(BIF_P, BADARG);
diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c
index 33417833a9..421c6670c6 100644
--- a/erts/emulator/beam/erl_init.c
+++ b/erts/emulator/beam/erl_init.c
@@ -141,7 +141,8 @@ static void erl_init(int ncpu,
 		     int port_tab_sz_ignore_files,
 		     int legacy_port_tab,
 		     int time_correction,
-		     ErtsTimeWarpMode time_warp_mode);
+		     ErtsTimeWarpMode time_warp_mode,
+		     int node_tab_delete_delay);
 
 static erts_atomic_t exiting;
 
@@ -314,7 +315,8 @@ erts_short_init(void)
 	     0,
 	     0,
 	     time_correction,
-	     time_warp_mode);
+	     time_warp_mode,
+	     ERTS_NODE_TAB_DELAY_GC_DEFAULT);
     erts_initialized = 1;
 }
 
@@ -326,7 +328,8 @@ erl_init(int ncpu,
 	 int port_tab_sz_ignore_files,
 	 int legacy_port_tab,
 	 int time_correction,
-	 ErtsTimeWarpMode time_warp_mode)
+	 ErtsTimeWarpMode time_warp_mode,
+	 int node_tab_delete_delay)
 {
     init_benchmarking();
 
@@ -367,7 +370,7 @@ erl_init(int ncpu,
     erts_init_binary(); /* Must be after init_emulator() */
     erts_bp_init();
     init_db(); /* Must be after init_emulator */
-    erts_init_node_tables();
+    erts_init_node_tables(node_tab_delete_delay);
     init_dist();
     erl_drv_thr_init();
     erts_init_async();
@@ -629,6 +632,9 @@ void erts_usage(void)
     erts_fprintf(stderr, "               see error_logger documentation for details\n");
     erts_fprintf(stderr, "-zdbbl size    set the distribution buffer busy limit in kilobytes\n");
     erts_fprintf(stderr, "               valid range is [1-%d]\n", INT_MAX/1024);
+    erts_fprintf(stderr, "-zdntgc time   set delayed node table gc in seconds\n");
+    erts_fprintf(stderr, "               valid values are infinity or intergers in the range [0-%d]\n",
+		 ERTS_NODE_TAB_DELAY_GC_MAX);
     erts_fprintf(stderr, "\n");
     erts_fprintf(stderr, "Note that if the emulator is started with erlexec (typically\n");
     erts_fprintf(stderr, "from the erl script), these flags should be specified with +.\n");
@@ -1219,6 +1225,7 @@ erl_start(int argc, char **argv)
     int legacy_port_tab = 0;
     int time_correction;
     ErtsTimeWarpMode time_warp_mode;
+    int node_tab_delete_delay = ERTS_NODE_TAB_DELAY_GC_DEFAULT;
 
     set_default_time_adj(&time_correction,
 			 &time_warp_mode);
@@ -2005,9 +2012,9 @@ erl_start(int argc, char **argv)
 
 	case 'z': {
 	    char *sub_param = argv[i]+2;
-	    int new_limit;
 
 	    if (has_prefix("dbbl", sub_param)) {
+		int new_limit;
 		arg = get_arg(sub_param+4, argv[i+1], &i);
 		new_limit = atoi(arg);
 		if (new_limit < 1 || INT_MAX/1024 < new_limit) {
@@ -2016,6 +2023,22 @@ erl_start(int argc, char **argv)
 		} else {
 		    erts_dist_buf_busy_limit = new_limit*1024;
 		}
+	    }
+	    else if (has_prefix("dntgc", sub_param)) {
+		long secs;
+
+		arg = get_arg(sub_param+5, argv[i+1], &i);
+		if (sys_strcmp(arg, "infinity") == 0)
+		    secs = ERTS_NODE_TAB_DELAY_GC_INFINITY;
+		else {
+		    errno = 0;
+		    secs = strtol(arg, NULL, 10);
+		    if (errno != 0 || secs < 0 || ERTS_NODE_TAB_DELAY_GC_MAX < secs) {
+			erts_fprintf(stderr, "Invalid delayed node table gc: %ld\n", secs);
+			erts_usage();
+		    }
+		}
+		node_tab_delete_delay = (int) secs;
 	    } else {
 		erts_fprintf(stderr, "bad -z option %s\n", argv[i]);
 		erts_usage();
@@ -2089,7 +2112,8 @@ erl_start(int argc, char **argv)
 	     port_tab_sz_ignore_files,
 	     legacy_port_tab,
 	     time_correction,
-	     time_warp_mode);
+	     time_warp_mode,
+	     node_tab_delete_delay);
 
     load_preloaded();
     erts_end_staging_code_ix();
diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c
index 964d830fc1..0950d7e7ef 100644
--- a/erts/emulator/beam/erl_node_tables.c
+++ b/erts/emulator/beam/erl_node_tables.c
@@ -51,6 +51,9 @@ static Uint dist_entries;
 
 static int references_atoms_need_init = 1;
 
+static ErtsMonotonicTime orig_node_tab_delete_delay;
+static ErtsMonotonicTime node_tab_delete_delay;
+
 /* -- The distribution table ---------------------------------------------- */
 
 #ifdef DEBUG
@@ -290,21 +293,46 @@ DistEntry *erts_find_dist_entry(Eterm sysname)
     return res;
 }
 
-void erts_delete_dist_entry(DistEntry *dep)
+static void try_delete_dist_entry(void *vdep)
+{
+    DistEntry *dep = (DistEntry *) vdep;
+    erts_aint_t refc;
+
+    erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
+    /*
+     * Another thread might have looked up this dist entry after
+     * we decided to delete it (refc became zero). If so, the other
+     * 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
+     * set up the timer. Delete the entry.
+     *
+     * If refc reach 0, the entry is currently not in use
+     * but has been used since we set up the timer. Set up a
+     * new timer.
+     *
+     * If refc > 0, the entry is in use. Keep the entry.
+     */
+    refc = erts_refc_dectest(&dep->refc, -1);
+    if (refc == -1)
+	(void) hash_erase(&erts_dist_table, (void *) dep);
+    erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+
+    if (refc == 0)
+	erts_schedule_delete_dist_entry(dep);
+}
+
+void erts_schedule_delete_dist_entry(DistEntry *dep)
 {
     ASSERT(dep != erts_this_dist_entry);
-    if(dep != erts_this_dist_entry) {
-	erts_smp_rwmtx_rwlock(&erts_dist_table_rwmtx);
-	/*
-	 * Another thread might have looked up this dist entry after
-	 * we decided to delete it (refc became zero). If so, the other
-	 * thread incremented refc twice. Once for the new reference
-	 * and once for this thread. Therefore, delete dist entry if
-	 * refc is 0 or -1 after a decrement.
-	 */
-	if (erts_refc_dectest(&dep->refc, -1) <= 0)
-	    (void) hash_erase(&erts_dist_table, (void *) dep);
-	erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx);
+    if (dep != erts_this_dist_entry) {
+	if (node_tab_delete_delay == 0)
+	    try_delete_dist_entry((void *) dep);
+	else if (node_tab_delete_delay > 0)
+	    erts_start_timer_callback(node_tab_delete_delay,
+				      try_delete_dist_entry,
+				      (void *) dep);
     }
 }
 
@@ -609,21 +637,46 @@ ErlNode *erts_find_or_insert_node(Eterm sysname, Uint creation)
     return res;
 }
 
-void erts_delete_node(ErlNode *enp)
+static void try_delete_node(void *venp)
+{
+    ErlNode *enp = (ErlNode *) venp;
+    erts_aint_t refc;
+
+    erts_smp_rwmtx_rwlock(&erts_node_table_rwmtx);
+    /*
+     * Another thread might have looked up this node after we
+     * decided to delete it (refc became zero). If so, the other
+     * 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
+     * set up the timer. Delete the entry.
+     *
+     * If refc reach 0, the entry is currently not in use
+     * but has been used since we set up the timer. Set up a
+     * new timer.
+     *
+     * If refc > 0, the entry is in use. Keep the entry.
+     */
+    refc = erts_refc_dectest(&enp->refc, -1);
+    if (refc == -1)
+	(void) hash_erase(&erts_node_table, (void *) enp);
+    erts_smp_rwmtx_rwunlock(&erts_node_table_rwmtx);
+
+    if (refc == 0)
+	erts_schedule_delete_node(enp);
+}
+
+void erts_schedule_delete_node(ErlNode *enp)
 {
     ASSERT(enp != erts_this_node);
-    if(enp != erts_this_node) {
-	erts_smp_rwmtx_rwlock(&erts_node_table_rwmtx);
-	/*
-	 * Another thread might have looked up this node after we
-	 * decided to delete it (refc became zero). If so, the other
-	 * thread incremented refc twice. Once for the new reference
-	 * and once for this thread. Therefore, delete node if refc
-	 * is 0 or -1 after a decrement.
-	 */
-	if (erts_refc_dectest(&enp->refc, -1) <= 0)
-	    (void) hash_erase(&erts_node_table, (void *) enp);
-	erts_smp_rwmtx_rwunlock(&erts_node_table_rwmtx);
+    if (enp != erts_this_node) {
+	if (node_tab_delete_delay == 0)
+	    try_delete_node((void *) enp);
+	else if (node_tab_delete_delay > 0)
+	    erts_start_timer_callback(node_tab_delete_delay,
+				      try_delete_node,
+				      (void *) enp);
     }
 }
 
@@ -651,7 +704,7 @@ static void print_node(void *venp, void *vpndp)
 	erts_print(pndp->to, pndp->to_arg, " %d", enp->creation);
 #ifdef DEBUG
 	erts_print(pndp->to, pndp->to_arg, " (refc=%ld)",
-		   erts_refc_read(&enp->refc, 1));
+		   erts_refc_read(&enp->refc, 0));
 #endif
 	pndp->no_sysname++;
     }
@@ -714,11 +767,28 @@ erts_set_this_node(Eterm sysname, Uint creation)
 
 }
 
-void erts_init_node_tables(void)
+Uint
+erts_delayed_node_table_gc(void)
+{
+    if (node_tab_delete_delay < 0)
+	return (Uint) ERTS_NODE_TAB_DELAY_GC_INFINITY;
+    if (node_tab_delete_delay == 0)
+	return (Uint) 0;
+    return (Uint) ((node_tab_delete_delay-1)/1000 + 1);
+}
+
+void erts_init_node_tables(int dd_sec)
 {
     erts_smp_rwmtx_opt_t rwmtx_opt = ERTS_SMP_RWMTX_OPT_DEFAULT_INITER;
     HashFunctions f;
 
+    if (dd_sec == ERTS_NODE_TAB_DELAY_GC_INFINITY)
+	node_tab_delete_delay = (ErtsMonotonicTime) -1;
+    else
+	node_tab_delete_delay = ((ErtsMonotonicTime) dd_sec)*1000;
+
+    orig_node_tab_delete_delay = node_tab_delete_delay;
+
     rwmtx_opt.type = ERTS_SMP_RWMTX_TYPE_FREQUENT_READ;
     rwmtx_opt.lived = ERTS_SMP_RWMTX_LONG_LIVED;
 
@@ -847,6 +917,7 @@ static Eterm AM_dist_references;
 static Eterm AM_node_references;
 static Eterm AM_system;
 static Eterm AM_timer;
+static Eterm AM_delayed_delete_timer;
 
 static void setup_reference_table(void);
 static Eterm reference_table_term(Uint **hpp, Uint *szp);
@@ -881,8 +952,10 @@ typedef struct dist_referrer_ {
     int heap_ref;
     int node_ref;
     int ctrl_ref;
+    int system_ref;
     Eterm id;
     Uint creation;
+    Uint id_heap[ID_HEAP_SIZE];
 } DistReferrer;
 
 typedef struct {
@@ -931,6 +1004,7 @@ erts_get_node_and_dist_references(struct process *proc)
 	INIT_AM(node_references);
 	INIT_AM(timer);
 	INIT_AM(system);
+	INIT_AM(delayed_delete_timer);
 	references_atoms_need_init = 0;
     }
 
@@ -988,17 +1062,25 @@ insert_dist_referrer(ReferredDist *referred_dist,
 					  sizeof(DistReferrer));
 	drp->next = referred_dist->referrers;
 	referred_dist->referrers = drp;
-	drp->id = id;
+	if(IS_CONST(id))
+	    drp->id = id;
+	else {
+	    Uint *hp = &drp->id_heap[0];
+	    ASSERT(is_tuple(id));
+	    drp->id = copy_struct(id, size_object(id), &hp, NULL);
+	}
 	drp->creation = creation;
 	drp->heap_ref = 0;
 	drp->node_ref = 0;
 	drp->ctrl_ref = 0;
+	drp->system_ref = 0;
     }
 
     switch (type) {
     case NODE_REF:	drp->node_ref++;	break;
     case CTRL_REF:	drp->ctrl_ref++;	break;
     case HEAP_REF:	drp->heap_ref++;	break;
+    case SYSTEM_REF:	drp->system_ref++;	break;
     default:		ASSERT(0);
     }
 }
@@ -1261,6 +1343,33 @@ insert_sys_msg(Eterm from, Eterm to, Eterm msg, ErlHeapFragment *bp)
 #endif
 
 static void
+insert_delayed_delete_node(void *state,
+			   ErtsMonotonicTime timeout_pos,
+			   void *vnp)
+{
+    DeclareTmpHeapNoproc(heap,3);
+    UseTmpHeapNoproc(3);
+    insert_node((ErlNode *) vnp,
+		SYSTEM_REF,
+		TUPLE2(&heap[0], AM_system, AM_delayed_delete_timer));
+    UnUseTmpHeapNoproc(3);
+}
+
+static void
+insert_delayed_delete_dist_entry(void *state,
+				 ErtsMonotonicTime timeout_pos,
+				 void *vdep)
+{
+    DeclareTmpHeapNoproc(heap,3);
+    UseTmpHeapNoproc(3);
+    insert_dist_entry((DistEntry *) vdep,
+		      SYSTEM_REF,
+		      TUPLE2(&heap[0], AM_system, AM_delayed_delete_timer),
+		      0);
+    UnUseTmpHeapNoproc(3);
+}
+
+static void
 setup_reference_table(void)
 {
     ErlHeapFragment *hfp;
@@ -1288,6 +1397,13 @@ setup_reference_table(void)
     /* Go through the hole system, and build a table of all references
        to ErlNode and DistEntry structures */
 
+    erts_debug_callback_timer_foreach(try_delete_node,
+				      insert_delayed_delete_node,
+				      NULL);
+    erts_debug_callback_timer_foreach(try_delete_dist_entry,
+				      insert_delayed_delete_dist_entry,
+				      NULL);
+
     UseTmpHeapNoproc(3);
     insert_node(erts_this_node,
 		SYSTEM_REF,
@@ -1601,7 +1717,7 @@ reference_table_term(Uint **hpp, Uint *szp)
 
 	tup = MK_2TUP(referred_nodes[i].node->sysname,
 		      MK_UINT(referred_nodes[i].node->creation));
-	tup = MK_3TUP(tup, MK_UINT(erts_refc_read(&referred_nodes[i].node->refc, 1)), nril);
+	tup = MK_3TUP(tup, MK_UINT(erts_refc_read(&referred_nodes[i].node->refc, 0)), nril);
 	nl = MK_CONS(tup, nl);
     }
 
@@ -1624,6 +1740,10 @@ reference_table_term(Uint **hpp, Uint *szp)
 		tup = MK_2TUP(AM_heap, MK_UINT(drp->heap_ref));
 		drl = MK_CONS(tup, drl);
 	    }
+	    if(drp->system_ref) {
+		tup = MK_2TUP(AM_system, MK_UINT(drp->system_ref));
+		drl = MK_CONS(tup, drl);
+	    }
 
 	    if (is_internal_pid(drp->id)) {
 		ASSERT(!drp->node_ref);
@@ -1633,6 +1753,14 @@ reference_table_term(Uint **hpp, Uint *szp)
 		ASSERT(drp->ctrl_ref && !drp->node_ref);
 		tup = MK_2TUP(AM_port, drp->id);
 	    }
+	    else if (is_tuple(drp->id)) {
+		Eterm *t;
+		ASSERT(drp->system_ref && !drp->node_ref
+		       && !drp->ctrl_ref && !drp->heap_ref);
+		t = tuple_val(drp->id);
+		ASSERT(2 == arityval(t[0]));
+		tup = MK_2TUP(t[1], t[2]);
+	    }
 	    else {
 		ASSERT(!drp->ctrl_ref && drp->node_ref);
 		ASSERT(is_atom(drp->id));
@@ -1650,7 +1778,7 @@ reference_table_term(Uint **hpp, Uint *szp)
 
 	/* DistList = [{Dist, Refc, ReferenceIdList}] */
 	tup = MK_3TUP(referred_dists[i].dist->sysname,
-		      MK_UINT(erts_refc_read(&referred_dists[i].dist->refc, 1)),
+		      MK_UINT(erts_refc_read(&referred_dists[i].dist->refc, 0)),
 		      dril);
 	dl = MK_CONS(tup, dl);
     }
@@ -1705,3 +1833,15 @@ delete_reference_table(void)
     }
 }
 
+void
+erts_debug_test_node_tab_delayed_delete(Sint64 millisecs)
+{
+    erts_smp_thr_progress_block();
+
+    if (millisecs < 0)
+	node_tab_delete_delay = orig_node_tab_delete_delay;
+    else
+	node_tab_delete_delay = millisecs;
+
+    erts_smp_thr_progress_unblock();
+}
diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h
index af60071ea5..54c5cd1d11 100644
--- a/erts/emulator/beam/erl_node_tables.h
+++ b/erts/emulator/beam/erl_node_tables.h
@@ -46,6 +46,10 @@
 #define ERTS_PORT_TASK_ONLY_BASIC_TYPES__
 #include "erl_port_task.h"
 #undef ERTS_PORT_TASK_ONLY_BASIC_TYPES__
+
+#define ERTS_NODE_TAB_DELAY_GC_DEFAULT (60)
+#define ERTS_NODE_TAB_DELAY_GC_MAX (100*1000*1000)
+#define ERTS_NODE_TAB_DELAY_GC_INFINITY (ERTS_NODE_TAB_DELAY_GC_MAX+1)
  
 #define ERST_INTERNAL_CHANNEL_NO 0
 
@@ -166,20 +170,21 @@ extern DistEntry *erts_this_dist_entry;
 extern ErlNode *erts_this_node;
 extern char *erts_this_node_sysname; /* must match erl_node_tables.c */
 
+Uint erts_delayed_node_table_gc(void);
 DistEntry *erts_channel_no_to_dist_entry(Uint);
 DistEntry *erts_sysname_to_connected_dist_entry(Eterm);
 DistEntry *erts_find_or_insert_dist_entry(Eterm);
 DistEntry *erts_find_dist_entry(Eterm);
-void erts_delete_dist_entry(DistEntry *);
+void erts_schedule_delete_dist_entry(DistEntry *);
 Uint erts_dist_table_size(void);
 void erts_dist_table_info(int, void *);
 void erts_set_dist_entry_not_connected(DistEntry *);
 void erts_set_dist_entry_connected(DistEntry *, Eterm, Uint);
 ErlNode *erts_find_or_insert_node(Eterm, Uint);
-void erts_delete_node(ErlNode *);
+void erts_schedule_delete_node(ErlNode *);
 void erts_set_this_node(Eterm, Uint);
 Uint erts_node_table_size(void);
-void erts_init_node_tables(void);
+void erts_init_node_tables(int);
 void erts_node_table_info(int, void *);
 void erts_print_node_info(int, void *, Eterm, int*, int*);
 Eterm erts_get_node_and_dist_references(struct process *);
@@ -204,7 +209,7 @@ erts_deref_dist_entry(DistEntry *dep)
 {
     ASSERT(dep);
     if (erts_refc_dectest(&dep->refc, 0) == 0)
-	erts_delete_dist_entry(dep);
+	erts_schedule_delete_dist_entry(dep);
 }
 
 ERTS_GLB_INLINE void
@@ -212,7 +217,7 @@ erts_deref_node_entry(ErlNode *np)
 {
     ASSERT(np);
     if (erts_refc_dectest(&np->refc, 0) == 0)
-	erts_delete_node(np);
+	erts_schedule_delete_node(np);
 }
 
 ERTS_GLB_INLINE void
@@ -253,5 +258,6 @@ erts_smp_de_links_unlock(DistEntry *dep)
 
 #endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */
 
+void erts_debug_test_node_tab_delayed_delete(Sint64 millisecs);
 
 #endif
diff --git a/erts/emulator/test/node_container_SUITE.erl b/erts/emulator/test/node_container_SUITE.erl
index 2f505893b4..479e6f200a 100644
--- a/erts/emulator/test/node_container_SUITE.erl
+++ b/erts/emulator/test/node_container_SUITE.erl
@@ -73,6 +73,8 @@ init_per_suite(Config) ->
     Config.
 
 end_per_suite(_Config) ->
+    erts_debug:set_internal_state(available_internal_state, true),
+    erts_debug:set_internal_state(node_tab_delayed_delete, -1), %% restore original value
     available_internal_state(false).
 
 init_per_group(_GroupName, Config) ->
@@ -419,6 +421,8 @@ node_table_gc(doc) ->
     ["Tests that node tables are garbage collected."];
 node_table_gc(suite) -> [];
 node_table_gc(Config) when is_list(Config) ->
+    erts_debug:set_internal_state(available_internal_state, true),
+    erts_debug:set_internal_state(node_tab_delayed_delete, 0),
     ?line PreKnown = nodes(known),
     ?line ?t:format("PreKnown = ~p~n", [PreKnown]),
     ?line make_node_garbage(0, 200000, 1000, []),
@@ -428,6 +432,7 @@ node_table_gc(Config) when is_list(Config) ->
     ?line ?t:format("PostAreas = ~p~n", [PostAreas]),
     ?line true = length(PostKnown) =< length(PreKnown),
     ?line nc_refc_check(node()),
+    erts_debug:set_internal_state(node_tab_delayed_delete, -1), %% restore original value
     ?line ok.
 
 make_node_garbage(N, L, I, Ps) when N < L ->
@@ -579,6 +584,8 @@ node_controller_refc(doc) ->
      "as they should for entities controlling a connections."];
 node_controller_refc(suite) -> [];
 node_controller_refc(Config) when is_list(Config) ->
+    erts_debug:set_internal_state(available_internal_state, true),
+    erts_debug:set_internal_state(node_tab_delayed_delete, 0),
     ?line NodeFirstName = get_nodefirstname(),
     ?line ?line {ok, Node} = start_node(NodeFirstName),
     ?line true = lists:member(Node, nodes()),
@@ -606,6 +613,7 @@ node_controller_refc(Config) when is_list(Config) ->
     ?line false = get_dist_references(Node),
     ?line false = lists:member(Node, nodes(known)),
     ?line nc_refc_check(node()),
+    erts_debug:set_internal_state(node_tab_delayed_delete, -1), %% restore original value
     ?line ok.
 
 %%
@@ -686,9 +694,6 @@ timer_refc(doc) ->
      "as they should for data stored in bif timers."];
 timer_refc(suite) -> [];
 timer_refc(Config) when is_list(Config) ->
-    {skipped, "Test needs to be UPDATED for new timer implementation"}.
-
-timer_refc_test(Config) when is_list(Config) ->
     ?line RNode = {get_nodename(), 1},
     ?line RPid = mk_pid(RNode, 4711, 2),
     ?line RPort = mk_port(RNode, 4711),
@@ -992,23 +997,32 @@ check_nd_refc({ThisNodeName, ThisCreation}, NodeRefs, DistRefs, Fail) ->
 check_refc(ThisNodeName,ThisCreation,Table,EntryList) when is_list(EntryList) ->
     lists:foreach(
       fun ({Entry, Refc, ReferrerList}) ->
-	      FoundRefs =
+	      {DelayedDeleteTimer,
+	       FoundRefs} =
 		  lists:foldl(
-		    fun ({_Referrer, ReferencesList}, A1) ->
-			    A1 + lists:foldl(fun ({_T,Rs},A2) ->
-						     A2+Rs
-					     end,
-					     0,
-					     ReferencesList)
+		    fun ({Referrer, ReferencesList}, {DDT, A1}) ->
+			    {case Referrer of
+				 {system,delayed_delete_timer} ->
+				     true;
+				 _ ->
+				     DDT
+			     end,
+			     A1 + lists:foldl(fun ({_T,Rs},A2) ->
+						      A2+Rs
+					      end,
+					      0,
+					      ReferencesList)}
 		    end,
-		    0,
+		    {false, 0},
 		    ReferrerList),
 	      
-	      %% Reference count equals found references ?
-	      case Refc =:= FoundRefs of
-		  true ->
+	      %% Reference count equals found references?
+	      case {Refc, FoundRefs, DelayedDeleteTimer} of
+		  {X, X, _} ->
+		      ok;
+		  {0, 1, true} ->
 		      ok;
-		  false ->
+		  _ ->
 		      exit({invalid_reference_count, Table, Entry})
 	      end,
 
@@ -1016,7 +1030,8 @@ check_refc(ThisNodeName,ThisCreation,Table,EntryList) when is_list(EntryList) ->
 	      case {Entry, Refc} of
 		  {ThisNodeName, 0} -> ok;
 		  {{ThisNodeName, ThisCreation}, 0} -> ok;
-		  {_, 0} -> exit({not_referred_entry_in_table, Table, Entry});
+		  {_, 0} when DelayedDeleteTimer == false ->
+		      exit({not_referred_entry_in_table, Table, Entry});
 		  {_, _} -> ok 
 	      end
 
diff --git a/erts/etc/common/erlexec.c b/erts/etc/common/erlexec.c
index d6544a2829..ba379f1da5 100644
--- a/erts/etc/common/erlexec.c
+++ b/erts/etc/common/erlexec.c
@@ -157,6 +157,7 @@ static char *plusr_val_switches[] = {
 /* +z arguments with values */
 static char *plusz_val_switches[] = {
     "dbbl",
+    "dntgc",
     NULL
 };
 
diff --git a/erts/preloaded/ebin/erlang.beam b/erts/preloaded/ebin/erlang.beam
index b99dc7339a..863a5e61ef 100644
--- a/erts/preloaded/ebin/erlang.beam
+++ b/erts/preloaded/ebin/erlang.beam
diff --git a/erts/preloaded/src/erlang.erl b/erts/preloaded/src/erlang.erl
index 9d9208d1b5..cf941ea6ca 100644
--- a/erts/preloaded/src/erlang.erl
+++ b/erts/preloaded/src/erlang.erl
@@ -2392,6 +2392,7 @@ tuple_to_list(_Tuple) ->
       CpuTopology :: cpu_topology();
          (creation) -> integer();
          (debug_compiled) -> boolean();
+         (delayed_node_table_gc) -> infinity | non_neg_integer();
          (dirty_cpu_schedulers) -> non_neg_integer();
          (dirty_cpu_schedulers_online) -> non_neg_integer();
          (dirty_io_schedulers) -> non_neg_integer();