Decentralized counters for ETS ordered_set with write_concurrency

Previously, all ETS tables used centralized counter variables to keep
track of the number of items stored and the amount of memory
consumed. These counters can cause scalability problems (especially on
big NUMA systems). This commit adds an implementation of a
decentralized counter and modifies the implementation of ETS so that
ETS tables of type ordered_set with write_concurrency enabled use the
decentralized counter. [Experiments][1] indicate that this change
substantially improves the scalability of ETS ordered_set tables with
write_concurrency enabled in scenarios with frequent `ets:insert/2`
and `ets:delete/2` calls.

The new counter is implemented in the module erts_flxctr
(`erts_flxctr.h` and `erts_flxctr.c`). The module has the suffix
flxctr as it contains the implementation of a flexible counter (i.e.,
counter instances can be configured to be either centralized or
decentralized). Counters that are configured to be centralized are
implemented with a single counter variable which is modified with
atomic operations. Decentralized counters are spread over several
cache lines (how many can be configured with the parameter
`+dcg`). The scheduler threads are mapped to cache lines so that there
is no single point of contention when decentralized counters are
updated. The thread progress functionality of the Erlang VM is
utilized to implement support for linearizable snapshots of
decentralized counters. The snapshot functionality is used by the
`ets:info/1` and `ets:info/2` functions.

[1]: http://winsh.me/ets_catree_benchmark/flxctr_res.html

18 files changed, 1357 insertions, 104 deletions

diff --git a/erts/doc/src/erl.xml b/erts/doc/src/erl.xml
index 88ddb03e97..471d7caa5a 100644
--- a/erts/doc/src/erl.xml
+++ b/erts/doc/src/erl.xml
@@ -636,6 +636,29 @@
           produces a crash dump. On Unix systems, sending an emulator process
           a <c>SIGUSR1</c> signal also forces a crash dump.</p>
       </item>
+      <tag><marker id="+dcg"/><c><![CDATA[+rg DecentralizedCounterGroupsLimit]]></c></tag>
+      <item>
+        <p>Limits the number of decentralized counter groups used by
+           decentralized counters optimized for update operations in the
+           Erlang runtime system. By default, the limit is 256.</p>
+        <p>When the number of schedulers is less than or equal to the
+           limit, each scheduler has its own group. When the
+           number of schedulers is larger than the groups limit,
+           schedulers share groups. Shared groups degrade
+           the performance for updating counters while many reader groups
+           degrade the performance for reading counters. So, the limit is a tradeoff
+           between performance for update operations and performance for
+           read operations. Each group consumes 64 bytes in each
+           counter.</p>
+        <p>Notice that a runtime system using decentralized
+           counter groups benefits from <seealso marker="#+sbt">binding
+           schedulers to logical processors</seealso>, as the groups are
+           distributed better between schedulers with this option.</p>
+        <p>This option only affects decentralized counters used for
+           the counters that are keeping track of the memory consumption
+           and the number of terms in ETS tables of type ordered_set with
+           the write_concurrency option activated.</p>
+      </item>
       <tag><marker id="+e"/><c><![CDATA[+e Number]]></c></tag>
       <item>
         <p>Sets the maximum number of ETS tables. This limit is
diff --git a/erts/emulator/Makefile.in b/erts/emulator/Makefile.in
index 21351df656..448f41b523 100644
--- a/erts/emulator/Makefile.in
+++ b/erts/emulator/Makefile.in
@@ -894,7 +894,7 @@ RUN_OBJS += \
 	$(OBJDIR)/erl_ptab.o		$(OBJDIR)/erl_map.o \
 	$(OBJDIR)/erl_msacc.o		$(OBJDIR)/erl_lock_flags.o \
 	$(OBJDIR)/erl_io_queue.o	$(OBJDIR)/erl_db_catree.o \
-	$(ESOCK_RUN_OBJS)
+	$(ESOCK_RUN_OBJS)		$(OBJDIR)/erl_flxctr.o
 
 LTTNG_OBJS = $(OBJDIR)/erlang_lttng.o
 
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 39d42d9757..a2f6284c9d 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -3028,6 +3028,8 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
 	BIF_RET(erts_nif_taints(BIF_P));
     } else if (ERTS_IS_ATOM_STR("reader_groups_map", BIF_ARG_1)) {
 	BIF_RET(erts_get_reader_groups_map(BIF_P));
+    } else if (ERTS_IS_ATOM_STR("decentralized_counter_groups_map", BIF_ARG_1)) {
+	BIF_RET(erts_get_decentralized_counter_groups_map(BIF_P));
     } else if (ERTS_IS_ATOM_STR("dist_buf_busy_limit", BIF_ARG_1)) {
 	Uint hsz = 0;
 
diff --git a/erts/emulator/beam/erl_cpu_topology.c b/erts/emulator/beam/erl_cpu_topology.c
index 6f8d2f8c35..6a4f43297e 100644
--- a/erts/emulator/beam/erl_cpu_topology.c
+++ b/erts/emulator/beam/erl_cpu_topology.c
@@ -34,6 +34,7 @@
 #include "error.h"
 #include "bif.h"
 #include "erl_cpu_topology.h"
+#include "erl_flxctr.h"
 
 #define ERTS_MAX_READER_GROUPS 64
 
@@ -58,6 +59,7 @@ static erts_cpu_info_t *cpuinfo;
 
 static int max_main_threads;
 static int reader_groups;
+static int decentralized_counter_groups;
 
 static ErtsCpuBindData *scheduler2cpu_map;
 static erts_rwmtx_t cpuinfo_rwmtx;
@@ -127,6 +129,8 @@ static erts_cpu_groups_map_t *cpu_groups_maps;
 
 static erts_cpu_groups_map_t *reader_groups_map;
 
+static erts_cpu_groups_map_t *decentralized_counter_groups_map;
+
 #define ERTS_TOPOLOGY_CG ERTS_TOPOLOGY_MAX_DEPTH
 
 #define ERTS_MAX_CPU_TOPOLOGY_ID ((int) 0xffff)
@@ -138,6 +142,7 @@ static void cpu_bind_order_sort(erts_cpu_topology_t *cpudata,
 static void write_schedulers_bind_change(erts_cpu_topology_t *cpudata, int size);
 
 static void reader_groups_callback(int, ErtsSchedulerData *, int, void *);
+static void flxctr_groups_callback(int, ErtsSchedulerData *, int, void *);
 static erts_cpu_groups_map_t *add_cpu_groups(int groups,
 					     erts_cpu_groups_callback_t callback,
 					     void *arg);
@@ -1646,7 +1651,8 @@ erts_get_logical_processors(int *conf, int *onln, int *avail)
 }
 
 void
-erts_pre_early_init_cpu_topology(int *max_rg_p,
+erts_pre_early_init_cpu_topology(int *max_dcg_p,
+                                 int *max_rg_p,
 				 int *conf_p,
 				 int *onln_p,
 				 int *avail_p)
@@ -1654,6 +1660,7 @@ erts_pre_early_init_cpu_topology(int *max_rg_p,
     cpu_groups_maps = NULL;
     no_cpu_groups_callbacks = 0;
     *max_rg_p = ERTS_MAX_READER_GROUPS;
+    *max_dcg_p = ERTS_MAX_FLXCTR_GROUPS;
     cpuinfo = erts_cpu_info_create();
     get_logical_processors(conf_p, onln_p, avail_p);
 }
@@ -1662,7 +1669,9 @@ void
 erts_early_init_cpu_topology(int no_schedulers,
 			     int *max_main_threads_p,
 			     int max_reader_groups,
-			     int *reader_groups_p)
+			     int *reader_groups_p,
+                             int max_decentralized_counter_groups,
+                             int *decentralized_counter_groups_p)
 {
     user_cpudata = NULL;
     user_cpudata_size = 0;
@@ -1687,6 +1696,12 @@ erts_early_init_cpu_topology(int no_schedulers,
 	max_main_threads = no_schedulers;
     *max_main_threads_p = max_main_threads;
 
+    decentralized_counter_groups = max_main_threads;
+    if (decentralized_counter_groups <= 1 || max_decentralized_counter_groups <= 1)
+	decentralized_counter_groups = 1;
+    if (decentralized_counter_groups > max_decentralized_counter_groups)
+	decentralized_counter_groups = max_decentralized_counter_groups;
+    *decentralized_counter_groups_p = decentralized_counter_groups;
     reader_groups = max_main_threads;
     if (reader_groups <= 1 || max_reader_groups <= 1)
 	reader_groups = 0;
@@ -1718,6 +1733,9 @@ erts_init_cpu_topology(void)
     reader_groups_map = add_cpu_groups(reader_groups,
 				       reader_groups_callback,
 				       NULL);
+    decentralized_counter_groups_map = add_cpu_groups(decentralized_counter_groups,
+                                                      flxctr_groups_callback,
+                                                      NULL);
 
     if (cpu_bind_order == ERTS_CPU_BIND_NONE)
 	erts_rwmtx_rwunlock(&cpuinfo_rwmtx);
@@ -1789,6 +1807,15 @@ reader_groups_callback(int suspending,
 	erts_rwmtx_set_reader_group(suspending ? 0 : group+1);
 }
 
+void
+flxctr_groups_callback(int suspending,
+		       ErtsSchedulerData *esdp,
+		       int group,
+		       void *unused)
+{
+    erts_flxctr_set_slot(suspending ? 0 : group+1);
+}
+
 static Eterm get_cpu_groups_map(Process *c_p,
 				erts_cpu_groups_map_t *map,
 				int offset);
@@ -1821,6 +1848,16 @@ erts_get_reader_groups_map(Process *c_p)
     return res;
 }
 
+Eterm
+erts_get_decentralized_counter_groups_map(Process *c_p)
+{
+    Eterm res;
+    erts_rwmtx_rlock(&cpuinfo_rwmtx);
+    res = get_cpu_groups_map(c_p, decentralized_counter_groups_map, 1);
+    erts_rwmtx_runlock(&cpuinfo_rwmtx);
+    return res;
+}
+
 /*
  * CPU groups
  */
diff --git a/erts/emulator/beam/erl_cpu_topology.h b/erts/emulator/beam/erl_cpu_topology.h
index 88bcad79ab..4a428d7972 100644
--- a/erts/emulator/beam/erl_cpu_topology.h
+++ b/erts/emulator/beam/erl_cpu_topology.h
@@ -27,14 +27,19 @@
 #ifndef ERL_CPU_TOPOLOGY_H__
 #define ERL_CPU_TOPOLOGY_H__
 
-void erts_pre_early_init_cpu_topology(int *max_rg_p,
-				      int *conf_p,
-				      int *onln_p,
-				      int *avail_p);
-void erts_early_init_cpu_topology(int no_schedulers,
-				  int *max_main_threads_p,
-				  int max_reader_groups,
-				  int *reader_groups_p);
+void
+erts_pre_early_init_cpu_topology(int *max_dcg_p,
+                                 int *max_rg_p,
+				 int *conf_p,
+				 int *onln_p,
+				 int *avail_p);
+void
+erts_early_init_cpu_topology(int no_schedulers,
+			     int *max_main_threads_p,
+			     int max_reader_groups,
+			     int *reader_groups_p,
+                             int max_decentralized_counter_groups,
+                             int *decentralized_counter_groups_p);
 void erts_init_cpu_topology(void);
 
 
@@ -70,6 +75,7 @@ Eterm erts_bind_schedulers(Process *c_p, Eterm how);
 Eterm erts_get_schedulers_binds(Process *c_p);
 
 Eterm erts_get_reader_groups_map(Process *c_p);
+Eterm erts_get_decentralized_counter_groups_map(Process *c_p);
 
 Eterm erts_set_cpu_topology(Process *c_p, Eterm term);
 Eterm erts_get_cpu_topology_term(Process *c_p, Eterm which);
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 0a50af4d1a..c0f5c506f4 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -42,6 +42,7 @@
 #include "bif.h"
 #include "big.h"
 #include "erl_binary.h"
+#include "bif.h"
 
 
 erts_atomic_t erts_ets_misc_mem_size;
@@ -64,6 +65,11 @@ do {                                                                     \
     }                                                                    \
 }while(0)
 
+#define DB_GET_APPROX_NITEMS(DB)                                        \
+    erts_flxctr_read_approx(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define DB_GET_APPROX_MEM_CONSUMED(DB)                                  \
+    erts_flxctr_read_approx(&(DB)->common.counters, ERTS_DB_TABLE_MEM_COUNTER_ID)
+
 static BIF_RETTYPE db_bif_fail(Process* p, Uint freason,
                                Uint bif_ix, Export* bif_exp)
 {
@@ -398,8 +404,9 @@ static void
 free_dbtable(void *vtb)
 {
     DbTable *tb = (DbTable *) vtb;
-
-    ASSERT(erts_atomic_read_nob(&tb->common.memory_size) == sizeof(DbTable));
+    ASSERT(erts_flxctr_is_snapshot_ongoing(&tb->common.counters) ||
+           sizeof(DbTable) == erts_flxctr_read_approx(&tb->common.counters,
+                                                      ERTS_DB_TABLE_MEM_COUNTER_ID));
 
     erts_rwmtx_destroy(&tb->common.rwlock);
     erts_mtx_destroy(&tb->common.fixlock);
@@ -408,7 +415,8 @@ free_dbtable(void *vtb)
     if (tb->common.btid)
         erts_bin_release(tb->common.btid);
 
-    erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
+    erts_flxctr_destroy(&tb->common.counters, ERTS_ALC_T_DB_TABLE);
+    erts_free(ERTS_ALC_T_DB_TABLE, tb);
 }
 
 static void schedule_free_dbtable(DbTable* tb)
@@ -1731,12 +1739,16 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
      */
     {
         DbTable init_tb;
-
-	erts_atomic_init_nob(&init_tb.common.memory_size, 0);
+        erts_flxctr_init(&init_tb.common.counters, 0, 2, ERTS_ALC_T_DB_TABLE);
 	tb = (DbTable*) erts_db_alloc(ERTS_ALC_T_DB_TABLE,
 				      &init_tb, sizeof(DbTable));
-	erts_atomic_init_nob(&tb->common.memory_size,
-				 erts_atomic_read_nob(&init_tb.common.memory_size));
+        erts_flxctr_init(&tb->common.counters,
+                         status & DB_CA_ORDERED_SET,
+                         2,
+                         ERTS_ALC_T_DB_TABLE);
+        erts_flxctr_add(&tb->common.counters,
+                        ERTS_DB_TABLE_MEM_COUNTER_ID,
+                        DB_GET_APPROX_MEM_CONSUMED(&init_tb));
     }
 
     tb->common.meth = meth;
@@ -1750,8 +1762,6 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
     tb->common.owner = BIF_P->common.id;
     set_heir(BIF_P, tb, heir, heir_data);
 
-    erts_atomic_init_nob(&tb->common.nitems, 0);
-
     tb->common.fixing_procs = NULL;
     tb->common.compress = is_compressed;
 #ifdef ETS_DBG_FORCE_TRAP
@@ -2128,19 +2138,18 @@ BIF_RETTYPE ets_internal_delete_all_2(BIF_ALIST_2)
 {
     SWord initial_reds = ERTS_BIF_REDS_LEFT(BIF_P);
     SWord reds = initial_reds;
-    Eterm nitems;
+    Eterm nitems_holder = THE_NON_VALUE;
     DbTable* tb;
-
     CHECK_TABLES();
 
     DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE, BIF_ets_internal_delete_all_2);
 
     if (BIF_ARG_2 == am_undefined) {
-        nitems = erts_make_integer(erts_atomic_read_nob(&tb->common.nitems),
-                                   BIF_P);
-
-        reds = tb->common.meth->db_delete_all_objects(BIF_P, tb, reds);
-
+        reds = tb->common.meth->db_delete_all_objects(BIF_P,
+                                                      tb,
+                                                      reds,
+                                                      &nitems_holder);
+        ASSERT(nitems_holder != THE_NON_VALUE);
         ASSERT(!(tb->common.status & DB_BUSY));
 
         if (reds < 0) {
@@ -2159,7 +2168,7 @@ BIF_RETTYPE ets_internal_delete_all_2(BIF_ALIST_2)
             db_unlock(tb, LCK_WRITE);
             BUMP_ALL_REDS(BIF_P);
             BIF_TRAP2(bif_export[BIF_ets_internal_delete_all_2], BIF_P,
-                      BIF_ARG_1, nitems);
+                      BIF_ARG_1, nitems_holder);
         }
         else {
             /* Done, no trapping needed */
@@ -2169,15 +2178,19 @@ BIF_RETTYPE ets_internal_delete_all_2(BIF_ALIST_2)
     }
     else {
         /*
-         * The table lookup succeeded and second argument is nitems
+         * The table lookup succeeded and second argument is nitems_holder
          * and not 'undefined', which means we have trapped at least once
          * and are now done.
          */
-        nitems = BIF_ARG_2;
+        nitems_holder = BIF_ARG_2;
     }
-
     db_unlock(tb, LCK_WRITE);
+    {
+    Eterm nitems =
+        tb->common.meth->db_delete_all_objects_get_nitems_from_holder(BIF_P,
+                                                                      nitems_holder);
     BIF_RET(nitems);
+    }
 }
 
 static void delete_all_objects_continue(Process* p, DbTable* tb)
@@ -2190,7 +2203,7 @@ static void delete_all_objects_continue(Process* p, DbTable* tb)
     if ((tb->common.status & (DB_DELETE|DB_BUSY)) != DB_BUSY)
         return;
 
-    reds = tb->common.meth->db_delete_all_objects(p, tb, reds);
+    reds = tb->common.meth->db_delete_all_objects(p, tb, reds, NULL);
     
     if (reds < 0) {
         BUMP_ALL_REDS(p);
@@ -3277,13 +3290,29 @@ BIF_RETTYPE ets_info_1(BIF_ALIST_1)
     int i;
     Eterm* hp;
     Uint freason;
+    Sint size = -1;
+    Sint memory = -1;
+    Eterm table;
+    int is_ctrs_read_result_set = 0;
     /*Process* rp = NULL;*/
     /* If/when we implement lockless private tables:
     Eterm owner;
     */
-
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ, &freason)) == NULL) {
-	if (freason == BADARG && (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)))
+    if(is_tuple(BIF_ARG_1) &&
+       is_tuple_arity(BIF_ARG_1, 2) &&
+       erts_flxctr_is_snapshot_result(tuple_val(BIF_ARG_1)[1])) {
+        Eterm counter_read_result  = tuple_val(BIF_ARG_1)[1];
+        table = tuple_val(BIF_ARG_1)[2];
+        size = erts_flxctr_get_snapshot_result_after_trap(counter_read_result,
+                                                          ERTS_DB_TABLE_NITEMS_COUNTER_ID);
+        memory = erts_flxctr_get_snapshot_result_after_trap(counter_read_result,
+                                                            ERTS_DB_TABLE_MEM_COUNTER_ID);
+        is_ctrs_read_result_set = 1;
+    } else {
+        table = BIF_ARG_1;
+    }
+    if ((tb = db_get_table(BIF_P, table, DB_INFO, LCK_READ, &freason)) == NULL) {
+	if (freason == BADARG && (is_atom(table) || is_ref(table)))
 	    BIF_RET(am_undefined);
 	else
             return db_bif_fail(BIF_P, freason, BIF_ets_info_1, NULL);
@@ -3314,9 +3343,35 @@ BIF_RETTYPE ets_info_1(BIF_ALIST_1)
 	    BIF_ERROR(BIF_P, BADARG);
 	}
     }*/
+
+    if (!is_ctrs_read_result_set) {
+        ErtsFlxCtrSnapshotResult res =
+            erts_flxctr_snapshot(&tb->common.counters, ERTS_ALC_T_DB_TABLE, BIF_P);
+        if (ERTS_FLXCTR_GET_RESULT_AFTER_TRAP == res.type) {
+            Eterm tuple;
+            db_unlock(tb, LCK_READ);
+            hp = HAlloc(BIF_P, 3);
+            tuple = TUPLE2(hp, res.trap_resume_state, table);
+            BIF_TRAP1(bif_export[BIF_ets_info_1], BIF_P, tuple);
+        } else if (res.type == ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP) {
+            db_unlock(tb, LCK_READ);
+            BIF_TRAP1(bif_export[BIF_ets_info_1], BIF_P, table);
+        } else {
+            size = res.result[ERTS_DB_TABLE_NITEMS_COUNTER_ID];
+            memory = res.result[ERTS_DB_TABLE_MEM_COUNTER_ID];
+            is_ctrs_read_result_set = 1;
+        }
+    }
     for (i = 0; i < sizeof(fields)/sizeof(Eterm); i++) {
-	results[i] = table_info(BIF_P, tb, fields[i]);
-	ASSERT(is_value(results[i]));
+        if (is_ctrs_read_result_set && am_size == fields[i]) {
+            results[i] = erts_make_integer(size, BIF_P);
+        } else if (is_ctrs_read_result_set && am_memory == fields[i]) {
+            Sint words = (Sint) ((memory + sizeof(Sint) - 1) / sizeof(Sint));
+            results[i] = erts_make_integer(words, BIF_P);
+        } else {
+            results[i] = table_info(BIF_P, tb, fields[i]);
+            ASSERT(is_value(results[i]));
+        }
     }
     db_unlock(tb, LCK_READ);
 
@@ -3344,14 +3399,43 @@ BIF_RETTYPE ets_info_2(BIF_ALIST_2)
     DbTable* tb;
     Eterm ret = THE_NON_VALUE;
     Uint freason;
-
+    if (erts_flxctr_is_snapshot_result(BIF_ARG_1)) {
+        Sint res;
+        if (am_memory == BIF_ARG_2) {
+            res = erts_flxctr_get_snapshot_result_after_trap(BIF_ARG_1,
+                                                             ERTS_DB_TABLE_MEM_COUNTER_ID);
+            res = (Sint) ((res + sizeof(Sint) - 1) / sizeof(Sint));
+        } else {
+            res = erts_flxctr_get_snapshot_result_after_trap(BIF_ARG_1,
+                                                             ERTS_DB_TABLE_NITEMS_COUNTER_ID);
+        }
+        BIF_RET(erts_make_integer(res, BIF_P));
+    }
     if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ, &freason)) == NULL) {
 	if (freason == BADARG && (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)))
 	    BIF_RET(am_undefined);
         else
             return db_bif_fail(BIF_P, freason, BIF_ets_info_2, NULL);
     }
-    ret = table_info(BIF_P, tb, BIF_ARG_2);
+    if (BIF_ARG_2 == am_size || BIF_ARG_2 == am_memory) {
+        ErtsFlxCtrSnapshotResult res =
+            erts_flxctr_snapshot(&tb->common.counters, ERTS_ALC_T_DB_TABLE, BIF_P);
+        if (ERTS_FLXCTR_GET_RESULT_AFTER_TRAP == res.type) {
+            db_unlock(tb, LCK_READ);
+            BIF_TRAP2(bif_export[BIF_ets_info_2], BIF_P, res.trap_resume_state, BIF_ARG_2);
+        } else if (res.type == ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP) {
+            db_unlock(tb, LCK_READ);
+            BIF_TRAP2(bif_export[BIF_ets_info_2], BIF_P, BIF_ARG_1, BIF_ARG_2);
+        } else if (BIF_ARG_2 == am_size) {
+            ret = erts_make_integer(res.result[ERTS_DB_TABLE_NITEMS_COUNTER_ID], BIF_P);
+        } else { /* BIF_ARG_2 == am_memory */
+            Sint r = res.result[ERTS_DB_TABLE_MEM_COUNTER_ID];
+            r = (Sint) ((r + sizeof(Sint) - 1) / sizeof(Sint));
+            ret = erts_make_integer(r, BIF_P);
+        }
+    } else {
+        ret = table_info(BIF_P, tb, BIF_ARG_2);
+    }
     db_unlock(tb, LCK_READ);
     if (is_non_value(ret)) {
 	BIF_ERROR(BIF_P, BADARG);
@@ -4121,7 +4205,8 @@ static Eterm table_info(Process* p, DbTable* tb, Eterm What)
     int use_monotonic;
 
     if (What == am_size) {
-	ret = make_small(erts_atomic_read_nob(&tb->common.nitems));
+        Uint size = (Uint) (DB_GET_APPROX_NITEMS(tb));
+        ret = erts_make_integer(size, p);
     } else if (What == am_type) {
 	if (tb->common.status & DB_SET)  {
 	    ret = am_set;
@@ -4136,7 +4221,7 @@ static Eterm table_info(Process* p, DbTable* tb, Eterm What)
 	    ret = am_bag;
 	}
     } else if (What == am_memory) {
-	Uint words = (Uint) ((erts_atomic_read_nob(&tb->common.memory_size)
+	Uint words = (Uint) ((DB_GET_APPROX_MEM_CONSUMED(tb)
 			      + sizeof(Uint)
 			      - 1)
 			     / sizeof(Uint));
@@ -4294,9 +4379,9 @@ static void print_table(fmtfn_t to, void *to_arg, int show,  DbTable* tb)
 
     tb->common.meth->db_print(to, to_arg, show, tb);
 
-    erts_print(to, to_arg, "Objects: %d\n", (int)erts_atomic_read_nob(&tb->common.nitems));
+    erts_print(to, to_arg, "Objects: %d\n", (int)DB_GET_APPROX_NITEMS(tb));
     erts_print(to, to_arg, "Words: %bpu\n",
-	       (Uint) ((erts_atomic_read_nob(&tb->common.memory_size)
+	       (Uint) ((DB_GET_APPROX_MEM_CONSUMED(tb)
 			+ sizeof(Uint)
 			- 1)
 		       / sizeof(Uint)));
diff --git a/erts/emulator/beam/erl_db.h b/erts/emulator/beam/erl_db.h
index dc77fbb60c..b22f35a5ef 100644
--- a/erts/emulator/beam/erl_db.h
+++ b/erts/emulator/beam/erl_db.h
@@ -160,7 +160,9 @@ do {									\
     erts_aint_t sz__ = (((erts_aint_t) (ALLOC_SZ))			\
 			- ((erts_aint_t) (FREE_SZ)));			\
     ASSERT((TAB));							\
-    erts_atomic_add_nob(&(TAB)->common.memory_size, sz__);		\
+    erts_flxctr_add(&(TAB)->common.counters,                            \
+                    ERTS_DB_TABLE_MEM_COUNTER_ID,                       \
+                    sz__);                                              \
 } while (0)
 
 #define ERTS_ETS_MISC_MEM_ADD(SZ) \
@@ -305,10 +307,10 @@ erts_db_free(ErtsAlcType_t type, DbTable *tab, void *ptr, Uint size)
     ASSERT(ptr != 0);
     ASSERT(size == ERTS_ALC_DBG_BLK_SZ(ptr));
     ERTS_DB_ALC_MEM_UPDATE_(tab, size, 0);
-
-    ASSERT(((void *) tab) != ptr
-	   || erts_atomic_read_nob(&tab->common.memory_size) == 0);
-
+    ASSERT(((void *) tab) != ptr ||
+           tab->common.counters.is_decentralized ||
+           0 == erts_flxctr_read_centralized(&tab->common.counters,
+                                             ERTS_DB_TABLE_MEM_COUNTER_ID));
     erts_free(type, ptr);
 }
 
diff --git a/erts/emulator/beam/erl_db_catree.c b/erts/emulator/beam/erl_db_catree.c
index 0402c6b7b4..962fe4c4f8 100644
--- a/erts/emulator/beam/erl_db_catree.c
+++ b/erts/emulator/beam/erl_db_catree.c
@@ -149,7 +149,12 @@ static SWord db_free_table_continue_catree(DbTable *tbl, SWord);
 static void db_foreach_offheap_catree(DbTable *,
                                       void (*)(ErlOffHeap *, void *),
                                       void *);
-static SWord db_delete_all_objects_catree(Process* p, DbTable* tbl, SWord reds);
+static SWord db_delete_all_objects_catree(Process* p,
+                                          DbTable* tbl,
+                                          SWord reds,
+                                          Eterm* nitems_holder_wb);
+static Eterm db_delete_all_objects_get_nitems_from_holder_catree(Process* p,
+                                                                 Eterm nitems_holder);
 static int
 db_lookup_dbterm_catree(Process *, DbTable *, Eterm key, Eterm obj,
                         DbUpdateHandle*);
@@ -191,6 +196,7 @@ DbTableMethod db_catree =
     db_select_replace_continue_catree,
     db_take_catree,
     db_delete_all_objects_catree,
+    db_delete_all_objects_get_nitems_from_holder_catree,
     db_free_table_catree,
     db_free_table_continue_catree,
     db_print_catree,
@@ -1357,6 +1363,8 @@ static SWord do_free_base_node_cont(DbTableCATree *tb, SWord num_left)
                 PUSH_NODE(&tb->free_stack_elems, root);
                 root = p;
             } else {
+                DEC_NITEMS((DbTable*)tb);
+                tb->nr_of_deleted_items++;
                 free_term((DbTable*)tb, root);
                 if (--num_left >= 0) {
                     break;
@@ -1397,6 +1405,7 @@ int db_create_catree(Process *p, DbTable *tbl)
     root = create_base_node(tb, NULL);
     tb->deletion = 0;
     tb->base_nodes_to_free_list = NULL;
+    tb->nr_of_deleted_items = 0;
     erts_atomic_init_relb(&(tb->root), (erts_aint_t)root);
     return DB_ERROR_NONE;
 }
@@ -2050,6 +2059,7 @@ static SWord db_free_table_continue_catree(DbTable *tbl, SWord reds)
         PUSH_NODE(&tb->free_stack_rnodes, GET_ROOT(tb));
         tb->is_routing_nodes_freed = 0;
         tb->base_nodes_to_free_list = NULL;
+        tb->nr_of_deleted_items = 0;
     }
     if ( ! tb->is_routing_nodes_freed ) {
         reds = do_free_routing_nodes_catree_cont(tb, reds);
@@ -2079,13 +2089,57 @@ static SWord db_free_table_continue_catree(DbTable *tbl, SWord reds)
     return 1;
 }
 
-static SWord db_delete_all_objects_catree(Process* p, DbTable* tbl, SWord reds)
+static
+int db_catree_nr_of_items_deleted_wb_dtor(Binary *context_bin) {
+    (void)context_bin;
+    return 1;
+}
+
+typedef struct {
+    Uint nr_of_deleted_items;
+} DbCATreeNrOfItemsDeletedWb;
+
+static Eterm
+create_and_install_num_of_deleted_items_wb_bin(Process *p, DbTableCATree *tb)
+{
+    Binary* bin =
+        erts_create_magic_binary(sizeof(DbCATreeNrOfItemsDeletedWb),
+                                 db_catree_nr_of_items_deleted_wb_dtor);
+    DbCATreeNrOfItemsDeletedWb* data = ERTS_MAGIC_BIN_DATA(bin);
+    Eterm* hp = HAlloc(p, ERTS_MAGIC_REF_THING_SIZE);
+    Eterm mref = erts_mk_magic_ref(&hp, &MSO(p), bin);
+    data->nr_of_deleted_items = 0;
+    tb->nr_of_deleted_items_wb = bin;
+    erts_refc_inctest(&bin->intern.refc, 2);   
+    return mref;
+}
+
+static Eterm db_delete_all_objects_get_nitems_from_holder_catree(Process* p,
+                                                                 Eterm mref)
 {
+    Binary* bin = erts_magic_ref2bin(mref);
+    DbCATreeNrOfItemsDeletedWb* data = ERTS_MAGIC_BIN_DATA(bin);
+    return erts_make_integer(data->nr_of_deleted_items, p);
+}
+
+static SWord db_delete_all_objects_catree(Process* p,
+                                          DbTable* tbl,
+                                          SWord reds,
+                                          Eterm* nitems_holder_wb)
+{
+    DbTableCATree *tb = &tbl->catree;
+    DbCATreeNrOfItemsDeletedWb* data;
+    if (!tb->deletion) {
+        *nitems_holder_wb =
+            create_and_install_num_of_deleted_items_wb_bin(p, tb);
+    }
     reds = db_free_table_continue_catree(tbl, reds);
     if (reds < 0)
         return reds;
+    data = ERTS_MAGIC_BIN_DATA(tb->nr_of_deleted_items_wb);
+    data->nr_of_deleted_items = tb->nr_of_deleted_items;
+    erts_bin_release(tb->nr_of_deleted_items_wb);
     db_create_catree(p, tbl);
-    erts_atomic_set_nob(&tbl->catree.common.nitems, 0);
     return reds;
 }
 
diff --git a/erts/emulator/beam/erl_db_catree.h b/erts/emulator/beam/erl_db_catree.h
index 418837be8e..fde442eaf5 100644
--- a/erts/emulator/beam/erl_db_catree.h
+++ b/erts/emulator/beam/erl_db_catree.h
@@ -87,6 +87,10 @@ typedef struct db_table_catree {
     CATreeNodeStack free_stack_rnodes;
     DbTableCATreeNode *base_nodes_to_free_list;
     int is_routing_nodes_freed;
+    /* The fields below are used by delete_all_objects and
+       select_delete(DeleteAll)*/
+    Uint nr_of_deleted_items;
+    Binary* nr_of_deleted_items_wb;
 } DbTableCATree;
 
 typedef struct {
@@ -104,7 +108,6 @@ void db_initialize_catree(void);
 
 int db_create_catree(Process *p, DbTable *tbl);
 
-
 TreeDbTerm** catree_find_root(Eterm key, CATreeRootIterator*);
 
 TreeDbTerm** catree_find_next_from_pb_key_root(Eterm key, CATreeRootIterator*);
diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index f225730029..ceaccf7e44 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -85,6 +85,14 @@
 
 #include "erl_db_hash.h"
 
+#define ADD_NITEMS(DB, TO_ADD)                                          \
+    erts_flxctr_add(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID, TO_ADD)
+#define INC_NITEMS(DB)                                                  \
+    erts_flxctr_inc_read_centralized(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define DEC_NITEMS(DB)                                                  \
+    erts_flxctr_dec_read_centralized(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define RESET_NITEMS(DB)                                                \
+    erts_flxctr_reset(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
 /* 
  * The following symbols can be manipulated to "tune" the linear hash array 
  */
@@ -121,7 +129,9 @@
      : ((struct segment**) erts_atomic_read_nob(&(tb)->segtab)))
 #endif
 #define NACTIVE(tb) ((int)erts_atomic_read_nob(&(tb)->nactive))
-#define NITEMS(tb) ((int)erts_atomic_read_nob(&(tb)->common.nitems))
+#define NITEMS(tb) \
+    ((Sint)erts_flxctr_read_centralized(&(tb)->common.counters,             \
+                                        ERTS_DB_TABLE_NITEMS_COUNTER_ID))
 
 #define SLOT_IX_TO_SEG_IX(i) (((i)+(EXT_SEGSZ-FIRST_SEGSZ)) >> EXT_SEGSZ_EXP)
 
@@ -444,7 +454,12 @@ static void db_foreach_offheap_hash(DbTable *,
 				    void (*)(ErlOffHeap *, void *),
 				    void *);
 
-static SWord db_delete_all_objects_hash(Process* p, DbTable* tbl, SWord reds);
+static SWord db_delete_all_objects_hash(Process* p,
+                                        DbTable* tbl,
+                                        SWord reds,
+                                        Eterm* nitems_holder_wb);
+static Eterm db_delete_all_objects_get_nitems_from_holder_hash(Process* p,
+                                                               Eterm nitems_holder);
 #ifdef HARDDEBUG
 static void db_check_table_hash(DbTableHash *tb);
 #endif
@@ -548,6 +563,7 @@ DbTableMethod db_hash =
     db_select_replace_continue_hash,
     db_take_hash,
     db_delete_all_objects_hash,
+    db_delete_all_objects_get_nitems_from_holder_hash,
     db_free_empty_table_hash,
     db_free_table_continue_hash,
     db_print_hash,
@@ -806,7 +822,7 @@ int db_put_hash(DbTable *tbl, Eterm obj, int key_clash_fail)
     if (tb->common.status & DB_SET) {
 	HashDbTerm* bnext = b->next;
 	if (is_pseudo_deleted(b)) {
-	    erts_atomic_inc_nob(&tb->common.nitems);
+            INC_NITEMS(tb);
             b->pseudo_deleted = 0;
 	}
 	else if (key_clash_fail) {
@@ -835,7 +851,7 @@ int db_put_hash(DbTable *tbl, Eterm obj, int key_clash_fail)
 	do {
 	    if (db_eq(&tb->common,obj,&q->dbterm)) {
 		if (is_pseudo_deleted(q)) {
-		    erts_atomic_inc_nob(&tb->common.nitems);
+		    INC_NITEMS(tb);
                     q->pseudo_deleted = 0;
 		    ASSERT(q->hvalue == hval);
 		    if (q != b) { /* must move to preserve key insertion order */
@@ -858,7 +874,7 @@ Lnew:
     q->pseudo_deleted = 0;
     q->next = b;
     *bp = q;
-    nitems = erts_atomic_inc_read_nob(&tb->common.nitems);
+    nitems = INC_NITEMS(tb);
     WUNLOCK_HASH(lck);
     {
 	int nactive = NACTIVE(tb);       
@@ -1056,7 +1072,7 @@ int db_erase_hash(DbTable *tbl, Eterm key, Eterm *ret)
     }
     WUNLOCK_HASH(lck);
     if (nitems_diff) {
-	erts_atomic_add_nob(&tb->common.nitems, nitems_diff);
+        ADD_NITEMS(tb, nitems_diff);
 	try_shrink(tb);
     }
     free_term_list(tb, free_us);
@@ -1117,7 +1133,7 @@ static int db_erase_object_hash(DbTable *tbl, Eterm object, Eterm *ret)
     }
     WUNLOCK_HASH(lck);
     if (nitems_diff) {
-	erts_atomic_add_nob(&tb->common.nitems, nitems_diff);
+        ADD_NITEMS(tb, nitems_diff);
 	try_shrink(tb);
     }
     free_term_list(tb, free_us);
@@ -2023,7 +2039,7 @@ static int select_delete_on_match_res(traverse_context_t* ctx_base, Sint slot_ix
         del->next = ctx->free_us;
         ctx->free_us = del;
     }
-    erts_atomic_dec_nob(&ctx->base.tb->common.nitems);
+    DEC_NITEMS(ctx->base.tb);
 
     return 1;
 }
@@ -2300,7 +2316,7 @@ static int db_take_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
     }
     WUNLOCK_HASH(lck);
     if (nitems_diff) {
-        erts_atomic_add_nob(&tb->common.nitems, nitems_diff);
+        ADD_NITEMS(tb, nitems_diff);
         try_shrink(tb);
     }
     free_term_list(tb, free_us);
@@ -2360,7 +2376,7 @@ static SWord db_mark_all_deleted_hash(DbTable *tbl, SWord reds)
     fixdel->slot = NACTIVE(tb) - 1;
     fixdel->all = 1;
     fixdel->trap = 0;
-    erts_atomic_set_nob(&tb->common.nitems, 0);
+    RESET_NITEMS(tb);
     return loops < 0 ? 0 : loops / LOOPS_PER_REDUCTION;
 }
 
@@ -2468,7 +2484,8 @@ static SWord db_free_table_continue_hash(DbTable *tbl, SWord reds)
 		     (void*)tb->locks, sizeof(DbTableHashFineLocks));
 	tb->locks = NULL;
     }
-    ASSERT(erts_atomic_read_nob(&tb->common.memory_size) == sizeof(DbTable));
+    ASSERT(sizeof(DbTable) == erts_flxctr_read_approx(&tb->common.counters,
+                                                      ERTS_DB_TABLE_MEM_COUNTER_ID));
     return reds;			/* Done */
 }
 
@@ -3080,7 +3097,7 @@ db_lookup_dbterm_hash(Process *p, DbTable *tbl, Eterm key, Eterm obj,
             ASSERT(q->hvalue == hval);
             q->pseudo_deleted = 0;
             *bp = b = q;
-            erts_atomic_inc_nob(&tb->common.nitems);
+            INC_NITEMS(tb);
         }
 
         HRelease(p, hend, htop);
@@ -3123,7 +3140,7 @@ db_finalize_dbterm_hash(int cret, DbUpdateHandle* handle)
         }
 
         WUNLOCK_HASH(lck);
-        erts_atomic_dec_nob(&tb->common.nitems);
+        DEC_NITEMS(tb);
         try_shrink(tb);
     } else {
         if (handle->flags & DB_MUST_RESIZE) {
@@ -3132,7 +3149,7 @@ db_finalize_dbterm_hash(int cret, DbUpdateHandle* handle)
         }
         if (handle->flags & DB_INC_TRY_GROW) {
             int nactive;
-            int nitems = erts_atomic_inc_read_nob(&tb->common.nitems);
+            int nitems = INC_NITEMS(tb);
             WUNLOCK_HASH(lck);
             nactive = NACTIVE(tb);
 
@@ -3153,8 +3170,17 @@ db_finalize_dbterm_hash(int cret, DbUpdateHandle* handle)
     return;
 }
 
-static SWord db_delete_all_objects_hash(Process* p, DbTable* tbl, SWord reds)
+static SWord db_delete_all_objects_hash(Process* p,
+                                        DbTable* tbl,
+                                        SWord reds,
+                                        Eterm* nitems_holder_wb)
 {
+    if (nitems_holder_wb != NULL) {
+        Uint nr_of_items =
+            erts_flxctr_read_centralized(&tbl->common.counters,
+                                         ERTS_DB_TABLE_NITEMS_COUNTER_ID);
+        *nitems_holder_wb = erts_make_integer(nr_of_items, p);
+    }
     if (IS_FIXED(tbl)) {
 	reds = db_mark_all_deleted_hash(tbl, reds);
     } else {
@@ -3163,11 +3189,16 @@ static SWord db_delete_all_objects_hash(Process* p, DbTable* tbl, SWord reds)
             return reds;
 
 	db_create_hash(p, tbl);
-	erts_atomic_set_nob(&tbl->hash.common.nitems, 0);
+        RESET_NITEMS(tbl);
     }
     return reds;
 }
 
+static Eterm db_delete_all_objects_get_nitems_from_holder_hash(Process* p,
+                                                               Eterm nitems_holder){
+    return nitems_holder;
+}
+
 void db_foreach_offheap_hash(DbTable *tbl,
 			     void (*func)(ErlOffHeap *, void *),
 			     void * arg)
diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
index f9ba04f399..492ea81b63 100644
--- a/erts/emulator/beam/erl_db_tree.c
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -51,9 +51,20 @@
 #include "erl_db_tree_util.h"
 
 #define GETKEY_WITH_POS(Keypos, Tplp) (*((Tplp) + Keypos))
-#define NITEMS(tb) ((int)erts_atomic_read_nob(&(tb)->common.nitems))
 
-#define TREE_MAX_ELEMENTS 0xFFFFFFFFUL
+#define NITEMS_CENTRALIZED(tb)                                          \
+    ((Sint)erts_flxctr_read_centralized(&(tb)->common.counters,          \
+                                        ERTS_DB_TABLE_NITEMS_COUNTER_ID))
+#define ADD_NITEMS(DB, TO_ADD)                                          \
+    erts_flxctr_add(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID, TO_ADD)
+#define INC_NITEMS(DB)                                                  \
+    erts_flxctr_inc(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define INC_NITEMS_CENTRALIZED(DB)                                      \
+    erts_flxctr_inc_read_centralized(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define RESET_NITEMS(DB)                                                \
+    erts_flxctr_reset(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+#define IS_CENTRALIZED_CTR(tb) (!(tb)->common.counters.is_decentralized)
+#define APPROX_MEM_CONSUMED(tb) erts_flxctr_read_approx(&(tb)->common.counters, ERTS_DB_TABLE_MEM_COUNTER_ID)
 
 #define TOPN_NODE(Dtt, Pos)                   \
      (((Pos) < Dtt->pos) ? 			\
@@ -296,7 +307,7 @@ int tree_balance_right(TreeDbTerm **this);
 static int delsub(TreeDbTerm **this); 
 static TreeDbTerm *slot_search(Process *p, TreeDbTerm *root, Sint slot,
                                DbTable *tb, DbTableTree *stack_container,
-                               CATreeRootIterator *iter);
+                               CATreeRootIterator *iter, int* is_EOT);
 static TreeDbTerm *find_node(DbTableCommon *tb, TreeDbTerm *root,
                              Eterm key, DbTableTree *stack_container);
 static TreeDbTerm **find_node2(DbTableCommon *tb, TreeDbTerm **root, Eterm key);
@@ -433,8 +444,12 @@ static void db_foreach_offheap_tree(DbTable *,
 				    void (*)(ErlOffHeap *, void *),
 				    void *);
 
-static SWord db_delete_all_objects_tree(Process* p, DbTable* tbl, SWord reds);
-
+static SWord db_delete_all_objects_tree(Process* p,
+                                        DbTable* tbl,
+                                        SWord reds,
+                                        Eterm* nitems_holder_wb);
+static Eterm db_delete_all_objects_get_nitems_from_holder_tree(Process* p,
+                                                               Eterm nitems_holder);
 #ifdef HARDDEBUG
 static void db_check_table_tree(DbTable *tbl);
 #endif
@@ -478,6 +493,7 @@ DbTableMethod db_tree =
     db_select_replace_continue_tree,
     db_take_tree,
     db_delete_all_objects_tree,
+    db_delete_all_objects_get_nitems_from_holder_tree,
     db_free_empty_table_tree,
     db_free_table_continue_tree,
     db_print_tree,
@@ -595,7 +611,8 @@ int db_last_tree_common(Process *p, DbTable *tbl, TreeDbTerm *root,
     }
     if (stack) {
 	PUSH_NODE(stack, this);
-	stack->slot = NITEMS(tbl);
+        /* Always centralized counters when static stack is used */
+	stack->slot = NITEMS_CENTRALIZED(tbl);
 	release_stack(tbl,stack_container,stack);
     }
     *ret = db_copy_key(p, tbl, &this->dbterm);
@@ -661,10 +678,7 @@ int db_put_tree_common(DbTableCommon *tb, TreeDbTerm **root, Eterm obj,
     for (;;)
 	if (!*this) { /* Found our place */
 	    state = 1;
-	    if (erts_atomic_inc_read_nob(&tb->nitems) >= TREE_MAX_ELEMENTS) {
-		erts_atomic_dec_nob(&tb->nitems);
-		return DB_ERROR_SYSRES;
-	    }
+            INC_NITEMS(((DbTable*)tb));
 	    *this = new_dbterm(tb, obj);
 	    (*this)->balance = 0;
 	    (*this)->left = (*this)->right = NULL;
@@ -888,7 +902,7 @@ int db_slot_tree_common(Process *p, DbTable *tbl, TreeDbTerm *root,
     TreeDbTerm *st;
     Eterm *hp, *hend;
     Eterm copy;
-
+    int is_EOT = 0;
     /*
      * The notion of a "slot" is not natural in a tree, but we try to
      * simulate it by giving the n'th node in the tree instead.
@@ -899,10 +913,10 @@ int db_slot_tree_common(Process *p, DbTable *tbl, TreeDbTerm *root,
 
     if (is_not_small(slot_term) ||
 	((slot = signed_val(slot_term)) < 0) ||
-	(slot > NITEMS(tbl)))
+	(IS_CENTRALIZED_CTR(tbl) && slot > NITEMS_CENTRALIZED(tbl)))
 	return DB_ERROR_BADPARAM;
 
-    if (slot == NITEMS(tbl)) {
+    if (IS_CENTRALIZED_CTR(tbl) && slot == NITEMS_CENTRALIZED(tbl)) {
 	*ret = am_EOT;
 	return DB_ERROR_NONE;
     }
@@ -912,7 +926,11 @@ int db_slot_tree_common(Process *p, DbTable *tbl, TreeDbTerm *root,
      * are counted from 1 and up.
      */
     ++slot;
-    st = slot_search(p, root, slot, tbl, stack_container, iter);
+    st = slot_search(p, root, slot, tbl, stack_container, iter, &is_EOT);
+    if (is_EOT) {
+        *ret = am_EOT;
+	return DB_ERROR_NONE;
+    }
     if (st == NULL) {
 	*ret = am_false;
 	return DB_ERROR_UNSPEC;
@@ -2244,7 +2262,8 @@ void db_print_tree_common(fmtfn_t to, void *to_arg,
 	erts_print(to, to_arg, "\n"
 		   "------------------------------------------------\n");
 #else
-    erts_print(to, to_arg, "Ordered set (AVL tree), Elements: %d\n", NITEMS(tbl));
+    erts_print(to, to_arg, "Ordered set (AVL tree), Elements: %d\n",
+               erts_flxctr_read_approx(&tbl->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID));
 #endif
 }
 
@@ -2281,24 +2300,41 @@ static SWord db_free_table_continue_tree(DbTable *tbl, SWord reds)
 		     (DbTable *) tb,
 		     (void *) tb->static_stack.array,
 		     sizeof(TreeDbTerm *) * STACK_NEED);
-	ASSERT((erts_atomic_read_nob(&tb->common.memory_size)
-	       == sizeof(DbTable)) ||
-               (erts_atomic_read_nob(&tb->common.memory_size)
-                == (sizeof(DbTable) + sizeof(DbFixation))));
+	ASSERT(erts_flxctr_is_snapshot_ongoing(&tb->common.counters) ||
+               ((APPROX_MEM_CONSUMED(tb)
+                 == sizeof(DbTable)) ||
+                (APPROX_MEM_CONSUMED(tb)
+                 == (sizeof(DbTable) + sizeof(DbFixation)))));
     }
     return reds;
 }
 
-static SWord db_delete_all_objects_tree(Process* p, DbTable* tbl, SWord reds)
+static SWord db_delete_all_objects_tree(Process* p,
+                                        DbTable* tbl,
+                                        SWord reds,
+                                        Eterm* nitems_holder_wb)
 {
+    if (nitems_holder_wb != NULL) {
+        Uint nr_of_items =
+            erts_flxctr_read_centralized(&tbl->common.counters,
+                                         ERTS_DB_TABLE_NITEMS_COUNTER_ID);
+        *nitems_holder_wb = erts_make_integer(nr_of_items, p);
+    }
     reds = db_free_table_continue_tree(tbl, reds);
     if (reds < 0)
         return reds;
     db_create_tree(p, tbl);
-    erts_atomic_set_nob(&tbl->tree.common.nitems, 0);
+    RESET_NITEMS(tbl);
     return reds;
 }
 
+static Eterm db_delete_all_objects_get_nitems_from_holder_tree(Process* p,
+                                                               Eterm holder)
+{
+    (void)p;
+    return holder;
+}
+
 static void do_db_tree_foreach_offheap(TreeDbTerm *,
 				       void (*)(ErlOffHeap *, void *),
 				       void *);
@@ -2383,7 +2419,7 @@ static TreeDbTerm *linkout_tree(DbTableCommon *tb, TreeDbTerm **root,
 		tstack[tpos++] = this;
 		state = delsub(this);
 	    }
-	    erts_atomic_dec_nob(&tb->nitems);
+            DEC_NITEMS(((DbTable*)tb));
 	    break;
 	}
     }
@@ -2450,7 +2486,7 @@ static TreeDbTerm *linkout_object_tree(DbTableCommon *tb,  TreeDbTerm **root,
 		tstack[tpos++] = this;
 		state = delsub(this);
 	    }
-	    erts_atomic_dec_nob(&tb->nitems);
+            DEC_NITEMS(((DbTable*)tb));
 	    break;
 	}
     }
@@ -2745,7 +2781,8 @@ static int delsub(TreeDbTerm **this)
 static TreeDbTerm *slot_search(Process *p, TreeDbTerm *root,
                                Sint slot, DbTable *tb,
                                DbTableTree *stack_container,
-                               CATreeRootIterator *iter)
+                               CATreeRootIterator *iter,
+                               int* is_EOT)
 {
     TreeDbTerm *this;
     TreeDbTerm *tmp;
@@ -2837,8 +2874,12 @@ static TreeDbTerm *slot_search(Process *p, TreeDbTerm *root,
         break;
 
 next_root:
-        if (!iter)
+        if (!iter) {
+            if (stack->slot == (slot-1)) {
+                *is_EOT = 1;
+            }
             break; /* EOT */
+        }
 
         ASSERT(slot > stack->slot);
         if (lastobj) {
@@ -2846,8 +2887,12 @@ next_root:
             lastobj = NULL;
         }
         pp = catree_find_next_root(iter, &lastkey);
-        if (!pp)
+        if (!pp) {
+            if (stack->slot == (slot-1)) {
+                *is_EOT = 1;
+            }
             break; /* EOT */
+        }
         root = *pp;
         stack->pos = 0;
         find_next(&tb->common, root, stack, lastkey);
diff --git a/erts/emulator/beam/erl_db_tree_util.h b/erts/emulator/beam/erl_db_tree_util.h
index 02df74678d..ba4a8f79e5 100644
--- a/erts/emulator/beam/erl_db_tree_util.h
+++ b/erts/emulator/beam/erl_db_tree_util.h
@@ -25,6 +25,8 @@
 ** Internal functions and macros used by both the CA tree and the AVL tree
 */
 
+
+#if defined(ARCH_32)
 /*
 ** A stack of this size is enough for an AVL tree with more than
 ** 0xFFFFFFFF elements. May be subject to change if
@@ -34,8 +36,19 @@
 ** Where n denotes the number of nodes, h(n) the height of the tree
 ** with n nodes and log is the binary logarithm.
 */
-  
 #define STACK_NEED 50
+#elif defined(ARCH_64)
+/*
+** A stack of this size is enough for an AVL tree with more than
+** 2^61 elements. 
+** The Maximal height of an AVL tree is calculated as above.
+*/
+#define STACK_NEED 90
+#else
+#error "Unsported architecture"
+#endif
+
+
 
 #define PUSH_NODE(Dtt, Tdt)                     \
     ((Dtt)->array[(Dtt)->pos++] = Tdt)
@@ -50,6 +63,9 @@
 
 #define EMPTY_NODE(Dtt) (TOP_NODE(Dtt) == NULL)
 
+#define DEC_NITEMS(DB)                                                  \
+    erts_flxctr_dec(&(DB)->common.counters, ERTS_DB_TABLE_NITEMS_COUNTER_ID)
+
 static ERTS_INLINE void free_term(DbTable *tb, TreeDbTerm* p)
 {
     db_free_term(tb, p, offsetof(TreeDbTerm, dbterm));
diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h
index e3d3c0e804..97f2848679 100644
--- a/erts/emulator/beam/erl_db_util.h
+++ b/erts/emulator/beam/erl_db_util.h
@@ -21,6 +21,7 @@
 #ifndef _DB_UTIL_H
 #define _DB_UTIL_H
 
+#include "erl_flxctr.h"
 #include "global.h"
 #include "erl_message.h"
 #include "erl_bif_unique.h"
@@ -207,8 +208,12 @@ typedef struct db_table_method
             enum DbIterSafety*);
     int (*db_take)(Process *, DbTable *, Eterm, Eterm *);
 
-    SWord (*db_delete_all_objects)(Process* p, DbTable* db, SWord reds);
-
+    SWord (*db_delete_all_objects)(Process* p,
+                                   DbTable* db,
+                                   SWord reds,
+                                   Eterm* nitems_holder_wb);
+    Eterm (*db_delete_all_objects_get_nitems_from_holder)(Process* p,
+                                                          Eterm nitems_holder);
     int (*db_free_empty_table)(DbTable* db);
     SWord (*db_free_table_continue)(DbTable* db, SWord reds);
     
@@ -257,6 +262,9 @@ typedef struct {
     DbTable *prev;
 } DbTableList;
 
+#define ERTS_DB_TABLE_NITEMS_COUNTER_ID 0
+#define ERTS_DB_TABLE_MEM_COUNTER_ID 1
+
 /*
  * This structure contains data for all different types of database
  * tables. Note that these fields must match the same fields
@@ -281,8 +289,11 @@ typedef struct db_table_common {
     Eterm the_name;           /* an atom */
     Binary *btid;
     DbTableMethod* meth;      /* table methods */
-    erts_atomic_t nitems; /* Total number of items in table */
-    erts_atomic_t memory_size;/* Total memory size. NOTE: in bytes! */
+    /* The ErtsFlxCtr below contains:
+     * - Total number of items in table
+     * - Total memory size (NOTE: in bytes!) */
+    ErtsFlxCtr counters;
+    char extra_for_flxctr[ERTS_FLXCTR_NR_OF_EXTRA_BYTES(2)];
     struct {                  /* Last fixation time */
 	ErtsMonotonicTime monotonic;
 	ErtsMonotonicTime offset;
diff --git a/erts/emulator/beam/erl_flxctr.c b/erts/emulator/beam/erl_flxctr.c
new file mode 100644
index 0000000000..35f4a21508
--- /dev/null
+++ b/erts/emulator/beam/erl_flxctr.c
@@ -0,0 +1,370 @@
+/*
+ * %CopyrightBegin%
+ *
+ * Copyright Ericsson AB 2019. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * %CopyrightEnd%
+ */
+
+/*
+ * Author: Kjell Winblad
+ */
+
+#include "erl_flxctr.h"
+
+static int reader_groups_array_size = 0;
+#define ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS (reader_groups_array_size)
+
+static int erts_flxctr_read_ctx_bin_dtor(Binary *context_bin);
+static int erts_flxctr_wait_dtor(Binary *context_bin);
+
+typedef struct {
+    ErtsThrPrgrLaterOp later_op;
+    Process* process;
+    ErtsFlxCtrDecentralizedCtrArray* array;
+    ErtsFlxCtrDecentralizedCtrArray* next_array;
+    ErtsAlcType_t alloc_type;
+    int nr_of_counters;
+    Sint result[ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE];
+} DecentralizedReadSnapshotInfo;
+
+typedef enum {
+    ERTS_FLXCTR_SNAPSHOT_NOT_ONGOING = 0,
+    ERTS_FLXCTR_SNAPSHOT_ONGOING = 1,
+    ERTS_FLXCTR_SNAPSHOT_ONGOING_TP_THREAD_DO_FREE = 2
+} erts_flxctr_snapshot_status;
+
+static void
+thr_prg_wake_up_and_count(void* bin_p)
+{
+    Binary* bin = bin_p;
+    DecentralizedReadSnapshotInfo* info = ERTS_MAGIC_BIN_DATA(bin);
+    Process* p = info->process;
+    ErtsFlxCtrDecentralizedCtrArray* array = info->array;
+    ErtsFlxCtrDecentralizedCtrArray* next = info->next_array;
+    int i, sched;
+    /* Reset result array */
+    for (i = 0; i < info->nr_of_counters; i++) {
+        info->result[i] = 0;
+    }
+    /* Read result from snapshot */
+    for (sched = 0; sched < ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS; sched++) {
+        for (i = 0; i < info->nr_of_counters; i++) {
+            info->result[i] = info->result[i] +
+                erts_atomic_read_nob(&array->array[sched].counters[i]);
+        }
+    }
+    /* Update the next decentralized counter array */
+    for (i = 0; i < info->nr_of_counters; i++) {
+        erts_atomic_add_nob(&next->array[0].counters[i], info->result[i]);
+    }
+    /* Announce that the snapshot is done */
+    {
+    Sint expected = ERTS_FLXCTR_SNAPSHOT_ONGOING;
+    if (expected != erts_atomic_cmpxchg_mb(&next->snapshot_status,
+                                           ERTS_FLXCTR_SNAPSHOT_NOT_ONGOING,
+                                           expected)) {
+        /* The CAS failed which means that this thread need to free the next array. */
+        erts_free(info->alloc_type, next->block_start);
+    }
+    }
+    /* Resume the process that requested the snapshot */
+    erts_proc_lock(p, ERTS_PROC_LOCK_STATUS);
+    if (!ERTS_PROC_IS_EXITING(p)) {
+        erts_resume(p, ERTS_PROC_LOCK_STATUS);
+    }
+    /* Free the memory that is no longer needed */
+    erts_free(info->alloc_type, array->block_start);
+    erts_proc_unlock(p, ERTS_PROC_LOCK_STATUS);
+    erts_proc_dec_refc(p);
+    erts_bin_release(bin);
+}
+
+typedef struct {
+    ErtsThrPrgrLaterOp later_op;
+    Process* process;
+} ErtsFlxCtrWakeUpLaterInfo;
+
+static void
+thr_prg_wake_up_later(void* bin_p)
+{
+    Binary* bin = bin_p;
+    ErtsFlxCtrWakeUpLaterInfo* info = ERTS_MAGIC_BIN_DATA(bin);
+    Process* p = info->process;
+    /* Resume the requesting process */
+    erts_proc_lock(p, ERTS_PROC_LOCK_STATUS);
+    if (!ERTS_PROC_IS_EXITING(p)) {
+        erts_resume(p, ERTS_PROC_LOCK_STATUS);
+    }
+    erts_proc_unlock(p, ERTS_PROC_LOCK_STATUS);
+    /* Free data */
+    erts_proc_dec_refc(p);
+    erts_bin_release(bin);
+}
+
+static
+int erts_flxctr_read_ctx_bin_dtor(Binary *context_bin) {
+    (void)context_bin;
+    return 1;
+}
+
+static
+int erts_flxctr_wait_dtor(Binary *context_bin) {
+    (void)context_bin;
+    return 1;
+}
+
+static void suspend_until_thr_prg(Process* p)
+{
+    Binary* state_bin;
+    ErtsFlxCtrWakeUpLaterInfo* info;
+    state_bin = erts_create_magic_binary(sizeof(ErtsFlxCtrWakeUpLaterInfo),
+                                         erts_flxctr_wait_dtor);
+    info = ERTS_MAGIC_BIN_DATA(state_bin);
+    info->process = p;
+    erts_refc_inctest(&state_bin->intern.refc, 1);
+    erts_suspend(p, ERTS_PROC_LOCK_MAIN, NULL);
+    erts_proc_inc_refc(p);
+    ERTS_VBUMP_ALL_REDS(p);
+    erts_schedule_thr_prgr_later_op(thr_prg_wake_up_later, state_bin, &info->later_op);
+}
+
+
+static ErtsFlxCtrDecentralizedCtrArray*
+create_decentralized_ctr_array(ErtsAlcType_t alloc_type, Uint nr_of_counters) {
+    /* Allocate an ErtsFlxCtrDecentralizedCtrArray and make sure that
+       the array field is located at the start of a cache line */
+    char* bytes =
+        erts_alloc(alloc_type,
+                   sizeof(ErtsFlxCtrDecentralizedCtrArray) +
+                   (sizeof(ErtsFlxCtrDecentralizedCtrArrayElem) *
+                    ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS) +
+                   ERTS_CACHE_LINE_SIZE);
+    void* block_start = bytes;
+    int bytes_to_next_cacheline_border;
+    ErtsFlxCtrDecentralizedCtrArray* array;
+    int i, sched;
+    bytes = &bytes[offsetof(ErtsFlxCtrDecentralizedCtrArray, array)];
+    bytes_to_next_cacheline_border =
+        ERTS_CACHE_LINE_SIZE - (((Uint)bytes) % ERTS_CACHE_LINE_SIZE);
+    array = (ErtsFlxCtrDecentralizedCtrArray*)
+        (&bytes[bytes_to_next_cacheline_border -
+                (int)offsetof(ErtsFlxCtrDecentralizedCtrArray, array)]);
+    ASSERT(((Uint)array->array) % ERTS_CACHE_LINE_SIZE == 0);
+    ASSERT(((Uint)array - (Uint)block_start) <= ERTS_CACHE_LINE_SIZE);
+    /* Initialize fields */
+    erts_atomic_init_nob(&array->snapshot_status, ERTS_FLXCTR_SNAPSHOT_ONGOING);
+    for (sched = 0; sched < ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS; sched++) {
+        for (i = 0; i < nr_of_counters; i++) {
+            erts_atomic_init_nob(&array->array[sched].counters[i], 0);
+        }
+    }
+    array->block_start = block_start;
+    return array;
+}
+
+void erts_flxctr_setup(int decentralized_counter_groups)
+{
+    reader_groups_array_size = decentralized_counter_groups+1;
+}
+
+void erts_flxctr_init(ErtsFlxCtr* c,
+                      int is_decentralized,
+                      Uint nr_of_counters,
+                      ErtsAlcType_t alloc_type)
+{
+    ASSERT(nr_of_counters <= ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE);
+    c->is_decentralized = is_decentralized;
+    c->nr_of_counters = nr_of_counters;
+    if (c->is_decentralized) {
+        ErtsFlxCtrDecentralizedCtrArray* array =
+            create_decentralized_ctr_array(alloc_type, nr_of_counters);
+        erts_atomic_set_nob(&array->snapshot_status,
+                            ERTS_FLXCTR_SNAPSHOT_NOT_ONGOING);
+        erts_atomic_init_nob(&c->u.counters_ptr, (Sint)array);
+        ASSERT(((Uint)array->array) % ERTS_CACHE_LINE_SIZE == 0);
+    } else {
+        int i;
+        for (i = 0; i < nr_of_counters; i++) {
+            erts_atomic_init_nob(&c->u.counters[i], 0);
+        }
+    }
+}
+
+void erts_flxctr_destroy(ErtsFlxCtr* c, ErtsAlcType_t type)
+{
+    if (c->is_decentralized) {
+        if (erts_flxctr_is_snapshot_ongoing(c)) {
+            ErtsFlxCtrDecentralizedCtrArray* array =
+                ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c);
+            /* Try to delegate the resposibilty of freeing to
+               thr_prg_wake_up_and_count */
+            Sint expected = ERTS_FLXCTR_SNAPSHOT_ONGOING;
+            if (expected !=
+                erts_atomic_cmpxchg_mb(&array->snapshot_status,
+                                       ERTS_FLXCTR_SNAPSHOT_ONGOING_TP_THREAD_DO_FREE,
+                                       expected)) {
+                /* The delegation was unsuccessful which means that no
+                   snapshot is ongoing anymore and the freeing needs
+                   to be done here */
+                ERTS_ASSERT(!erts_flxctr_is_snapshot_ongoing(c));
+                erts_free(type, array->block_start);
+            }
+        } else {
+            erts_free(type, ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c)->block_start);
+        }
+    }
+}
+
+ErtsFlxCtrSnapshotResult
+erts_flxctr_snapshot(ErtsFlxCtr* c,
+                     ErtsAlcType_t alloc_type,
+                     Process* p)
+{
+    if (c->is_decentralized) {
+        ErtsFlxCtrDecentralizedCtrArray* array = ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c);
+        if (erts_flxctr_is_snapshot_ongoing(c)) {
+            /* Let the caller try again later */
+            ErtsFlxCtrSnapshotResult res =
+                {.type = ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP};
+            suspend_until_thr_prg(p);
+            return res;
+        } else {
+            Eterm* hp;
+            Binary* state_bin;
+            Eterm state_mref;
+            DecentralizedReadSnapshotInfo* info;
+            ErtsFlxCtrDecentralizedCtrArray* new_array =
+                create_decentralized_ctr_array(alloc_type, c->nr_of_counters);
+            int success =
+                ((Sint)array) == erts_atomic_cmpxchg_mb(&c->u.counters_ptr,
+                                                        (Sint)new_array,
+                                                        (Sint)array);
+            if (!success) {
+                /* Let the caller try again later */
+                ErtsFlxCtrSnapshotResult res =
+                    {.type = ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP};
+                suspend_until_thr_prg(p);
+                erts_free(alloc_type, new_array->block_start);
+                return res;
+            }
+            /* Create binary with info about the operation that can be
+               sent to the caller and to a thread progress function */
+            state_bin =
+                erts_create_magic_binary(sizeof(DecentralizedReadSnapshotInfo),
+                                         erts_flxctr_read_ctx_bin_dtor);
+            hp = HAlloc(p, ERTS_MAGIC_REF_THING_SIZE);
+            state_mref = erts_mk_magic_ref(&hp, &MSO(p), state_bin);
+            info = ERTS_MAGIC_BIN_DATA(state_bin);
+            info->alloc_type = alloc_type;
+            info->array = array;
+            info->next_array = new_array;
+            info->process = p;
+            info->nr_of_counters = c->nr_of_counters;
+            erts_proc_inc_refc(p);
+            erts_refc_inctest(&state_bin->intern.refc, 2);
+            erts_suspend(p, ERTS_PROC_LOCK_MAIN, NULL);
+            ERTS_VBUMP_ALL_REDS(p);
+            erts_schedule_thr_prgr_later_op(thr_prg_wake_up_and_count,
+                                            state_bin,
+                                            &info->later_op);
+            {
+                ErtsFlxCtrSnapshotResult res = {
+                    .type = ERTS_FLXCTR_GET_RESULT_AFTER_TRAP,
+                    .trap_resume_state = state_mref};
+                return res;
+            }
+        }
+    } else {
+        ErtsFlxCtrSnapshotResult res;
+        int i;
+        res.type = ERTS_FLXCTR_DONE;
+        for (i = 0; i < c->nr_of_counters; i++){
+            res.result[i] = erts_flxctr_read_centralized(c, i);
+        }
+        return res;
+    }
+}
+
+
+Sint erts_flxctr_get_snapshot_result_after_trap(Eterm result_holder,
+                                            Uint counter_nr)
+{
+    Binary* bin = erts_magic_ref2bin(result_holder);
+    DecentralizedReadSnapshotInfo* data = ERTS_MAGIC_BIN_DATA(bin);;
+    return data->result[counter_nr];
+}
+
+int erts_flxctr_is_snapshot_result(Eterm term)
+{
+    if (is_internal_magic_ref(term)) {
+        Binary* bin = erts_magic_ref2bin(term);
+        return ERTS_MAGIC_BIN_DESTRUCTOR(bin) ==  erts_flxctr_read_ctx_bin_dtor;
+    } else return 0;
+}
+
+Sint erts_flxctr_read_approx(ErtsFlxCtr* c,
+                             Uint counter_nr)
+{
+    if (c->is_decentralized) {
+        ErtsFlxCtrDecentralizedCtrArray* counter = ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c);
+        Sint sum = 0;
+        int sched;
+        for (sched = 0; sched < ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS; sched++) {
+            sum = sum + erts_atomic_read_nob(&counter->array[sched].counters[counter_nr]);
+        }
+        return sum;
+    } else {
+        return erts_flxctr_read_centralized(c, counter_nr);
+    }
+}
+
+int erts_flxctr_is_snapshot_ongoing(ErtsFlxCtr* c)
+{
+    return c->is_decentralized &&
+        (ERTS_FLXCTR_SNAPSHOT_NOT_ONGOING !=
+         erts_atomic_read_acqb(&ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c)->snapshot_status));
+}
+
+int erts_flxctr_suspend_until_thr_prg_if_snapshot_ongoing(ErtsFlxCtr* c, Process* p)
+{
+    if (erts_flxctr_is_snapshot_ongoing(c)) {
+        suspend_until_thr_prg(p);
+        return 1;
+    } else {
+        return 0;
+    }
+}
+
+void erts_flxctr_reset(ErtsFlxCtr* c,
+                       Uint counter_nr)
+{
+    if (c->is_decentralized) {
+        int sched;
+        ErtsFlxCtrDecentralizedCtrArray* counter =
+            ERTS_FLXCTR_GET_CTR_ARRAY_PTR(c);
+        for (sched = 0; sched < ERTS_FLXCTR_DECENTRALIZED_NO_SLOTS; sched++) {
+            erts_atomic_set_nob(&counter->array[sched].counters[counter_nr], 0);
+        }
+    } else {
+        erts_atomic_set_nob(&c->u.counters[counter_nr], 0);
+    }
+}
+
+
+void erts_flxctr_set_slot(int group) {
+    ErtsSchedulerData *esdp = erts_get_scheduler_data();
+    esdp->flxctr_slot_no = group;
+}
diff --git a/erts/emulator/beam/erl_flxctr.h b/erts/emulator/beam/erl_flxctr.h
new file mode 100644
index 0000000000..5cab02b9eb
--- /dev/null
+++ b/erts/emulator/beam/erl_flxctr.h
@@ -0,0 +1,406 @@
+/*
+ * %CopyrightBegin%
+ *
+ * Copyright Ericsson AB 2019. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * %CopyrightEnd%
+ */
+
+/**
+ * @file erl_flxctr.h
+ *
+ * @brief This file contains the API of a flexible counter. The
+ * counter can be configured during its initialization to be
+ * centralized or decentralized. The centralized configuration makes
+ * it possible to read the counter value extremely efficiently, but
+ * updates of the counter value can easily cause contention. The
+ * decentralized configuration has the reverse trade-off (i.e.,
+ * updates are efficient and scalable but reading the counter value is
+ * slow and may cause contention).
+ *
+ * @author Kjell Winblad
+ */
+
+#ifndef ERL_FLXCTR_H__
+#define ERL_FLXCTR_H__
+
+#ifdef HAVE_CONFIG_H
+#  include "config.h"
+#endif
+#include "sys.h"
+#include "erl_vm.h"
+#include "global.h"
+#include "error.h"
+#include "bif.h"
+#include "big.h"
+#include "erl_binary.h"
+#include "bif.h"
+#include <stddef.h>
+
+/* Public Interface */
+
+#define ERTS_MAX_FLXCTR_GROUPS 256
+#define ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE (ERTS_CACHE_LINE_SIZE / sizeof(erts_atomic_t))
+
+typedef struct {
+    int nr_of_counters;
+    int is_decentralized;
+    union {
+        erts_atomic_t counters_ptr;
+        erts_atomic_t counters[1];
+    } u;
+} ErtsFlxCtr;
+
+#define ERTS_FLXCTR_NR_OF_EXTRA_BYTES(NR_OF_COUNTERS)   \
+    ((NR_OF_COUNTERS-1) * sizeof(erts_atomic_t))
+
+/* Called by early_init */
+void erts_flxctr_setup(int decentralized_counter_groups);
+
+/**
+ * @brief Initializes an ErtsFlxCtr. The macro
+ * ERTS_FLXCTR_NR_OF_EXTRA_BYTES should be used to determine how much
+ * extra space that needs to be allocated directly after the
+ * ErtsFlxCtr when is_decentralized is set to zero.  Each ErtsFlxCtr
+ * instance may contain up to ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE
+ * counters. These counters are numbered from zero to
+ * (ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE-1). Most of the functions in
+ * this module take a parameter named counter_nr that controls which
+ * of the ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE counters in the given
+ * ErtsFlxCtr that should be operated on.
+ *
+ * @param c The counter to initialize
+ * @param is_decentralized Non-zero value to make c decentralized
+ * @param nr_of_counters The number of counters included in c
+ *                       (max ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE)
+ * @param alloc_type 
+ */
+void erts_flxctr_init(ErtsFlxCtr* c,
+                      int is_decentralized,
+                      Uint nr_of_counters,
+                      ErtsAlcType_t alloc_type);
+
+/**
+ * @brief Destroys an initialized counter.
+ *
+ * @param c The counter that should be destroyed
+ * @param alloc_type The allocation type (needs to be the same as the
+ *                   one passed to erts_flxctr_init when c was
+ *                   initialized)
+ */
+void erts_flxctr_destroy(ErtsFlxCtr* c, ErtsAlcType_t alloc_type);
+
+/**
+ * @brief Adds to_add to the counter with counter_nr in c
+ *
+ * @param c the ErtsFlxCtr to operate on
+ * @param counter_nr The number of the counter in c to modify
+ * @param to_add The amount that should be added to the specified counter
+ */
+ERTS_GLB_INLINE
+void erts_flxctr_add(ErtsFlxCtr* c,
+                     Uint counter_nr,
+                     int to_add);
+
+/**
+ * @brief Increases the specified counter by 1
+ *
+ * @param c The ErtsFlxCtr instance to operate on
+ * @param counter_nr The number of the counter within c to operate on
+ */
+ERTS_GLB_INLINE
+void erts_flxctr_inc(ErtsFlxCtr* c,
+                     Uint counter_nr);
+
+/**
+ * @brief Decreases the specified counter by 1
+ */
+ERTS_GLB_INLINE
+void erts_flxctr_dec(ErtsFlxCtr* c,
+                     Uint counter_nr);
+
+/**
+ * @brief This function tries to return the current value of the
+ * specified counter but may return an incorrect result if the counter
+ * is decentralized and other threads are accessing the counter
+ * concurrently.
+ *
+ * @param c The ErtsFlxCtr instance to operate on
+ * @param counter_nr The number of the counter within c to operate on
+ *
+ * @return A snapshot of the specifed counter if c is centralized or a
+ *         possibly incorrect estimate of the counter value if c is
+ *         decentralized
+ */
+Sint erts_flxctr_read_approx(ErtsFlxCtr* c,
+                             Uint counter_nr);
+
+/**
+ * @brief This function can only be used together with an ErtsFlxCtr
+ * that is configured to be centralized. The function increments the
+ * specified counter by 1 and returns the value of the counter after
+ * the increment.
+ */
+ERTS_GLB_INLINE
+Sint erts_flxctr_inc_read_centralized(ErtsFlxCtr* c,
+                                      Uint counter_nr);
+
+/**
+ * @brief This function can only be used together with a ErtsFlxCtr
+ * that is configured to be centralized. The function decrements the
+ * specified counter by 1 and returns the value of the counter after
+ * the operation.
+ */
+ERTS_GLB_INLINE
+Sint erts_flxctr_dec_read_centralized(ErtsFlxCtr* c,
+                                      Uint counter_nr);
+
+/**
+ * @brief This function can only be used together with an ErtsFlxCtr
+ * that is configured to be centralized. The function returns the
+ * current value of the specified counter.
+ */
+ERTS_GLB_INLINE
+Sint erts_flxctr_read_centralized(ErtsFlxCtr* c,
+                                  Uint counter_nr);
+
+
+typedef enum {
+    ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP,
+    ERTS_FLXCTR_DONE,
+    ERTS_FLXCTR_GET_RESULT_AFTER_TRAP
+} ErtsFlxctrSnapshotResultType;
+
+typedef struct {
+    ErtsFlxctrSnapshotResultType type;
+    Eterm trap_resume_state;
+    Sint result[ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE];
+} ErtsFlxCtrSnapshotResult;
+
+/**
+ * @brief This function initiates an atomic snapshot of an ErtsFlxCtr
+ * to read out the values of one or more of the counters that are
+ * stored in the given ErtsFlxCtr. The caller needs to perform
+ * different actions after the return of this function depending on
+ * the value of the type field in the returned struct:
+ *
+ * - The caller needs to trap and try again after the trap if the
+ *   return value has the type ERTS_FLXCTR_TRY_AGAIN_AFTER_TRAP.
+ *
+ * - The caller can get the result directly from the result field of
+ *   the returned struct if the return value has the type
+ *   ERTS_FLXCTR_DONE. The value at index i in the result field
+ *   correspond to counter number i.
+ *
+ * - Finally, if the return value has the type
+ *   ERTS_FLXCTR_GET_RESULT_AFTER_TRAP, then the caller needs to save
+ *   the value of the field trap_resume_state from the returned struct
+ *   and trap. After the trap, the values of the counters can be
+ *   obtained by using the function
+ *   erts_flxctr_get_snapshot_result_after_trap. Note that the
+ *   function erts_flxctr_is_snapshot_result can be used to check if a
+ *   value is obtained from the trap_resume_state field in the
+ *   returned struct (this can be useful when the calling function
+ *   wakes up again after the trap).
+ *
+ * The snapshot operation that is initiated by this function should be
+ * considered to be ongoing from the issuing of this function until a
+ * struct with the type field set to ERTS_FLXCTR_DONE has been
+ * returned from the function or until the caller of this function has
+ * woken up after trapping.
+ *
+ * @param c The ErtsFlxCtr that the snapshot shall be taken from
+ * @param alloc_type The allocation type (needs to be the same as the
+ *                   type passed to erts_flxctr_init when c was
+ *                   initialized)
+ * @param p The Erlang process that is doing the call
+ *
+ * @return See the description above
+ *
+ */
+ErtsFlxCtrSnapshotResult
+erts_flxctr_snapshot(ErtsFlxCtr* c,
+                     ErtsAlcType_t alloc_type,
+                     Process* p);
+
+/**
+ * @brief Checks if the parameter term is a snapshot result (i.e.,
+ * something obtained from the trap_resume_state field of an
+ * ErtsFlxCtrSnapshotResult struct that has been returned from
+ * erts_flxctr_snapshot).
+ *
+ * @param term The term to check 
+ *
+ * @return A nonzero value iff the term is a snapshot result
+ */
+int erts_flxctr_is_snapshot_result(Eterm term);
+
+/**
+ * @brief Returns the result of a snapshot for a counter given a
+ * snapshot result returned by a call to erts_flxctr_snapshot (i.e.,
+ * the value stored in the trap_resume_state field of a struct
+ * returned by erts_flxctr_snapshot). The caller needs to trap between
+ * the return of erts_flxctr_snapshot and the call to this function.
+ */
+Sint erts_flxctr_get_snapshot_result_after_trap(Eterm trap_resume_state,
+                                                Uint counter_nr);
+
+/**
+ * @brief Resets the specified counter to 0. This function is unsafe
+ * to call while a snapshot operation may be active (initiated with
+ * the erts_flxctr_snapshot function).
+ */
+void erts_flxctr_reset(ErtsFlxCtr* c,
+                       Uint counter_nr);
+
+/**
+ * @brief Checks if a snapshot operation is active (snapshots are
+ * initiated with the erts_flxctr_snapshot function).
+ *
+ * @return nonzero value iff a snapshot was active at some point
+ * between the invocation and return of the function
+ */
+int erts_flxctr_is_snapshot_ongoing(ErtsFlxCtr* c);
+
+/**
+ * @brief This function checks if a snapshot operation is ongoing
+ * (snapshots are initiated with the erts_flxctr_snapshot function)
+ * and suspend the given process until thread progress has happened if
+ * it detected an ongoing snapshot operation. The caller needs to trap
+ * if a non-zero value is returned.
+ *
+ * @param c The ErtsFlxCtr to check
+ * @param p The calling process
+ *
+ * @return nonzero value if the given process has got suspended
+ */
+int erts_flxctr_suspend_until_thr_prg_if_snapshot_ongoing(ErtsFlxCtr* c, Process* p);
+
+/* End: Public Interface */
+
+/* Internal Declarations */
+
+#define ERTS_FLXCTR_GET_CTR_ARRAY_PTR(C)                                \
+    ((ErtsFlxCtrDecentralizedCtrArray*) erts_atomic_read_acqb(&(C)->u.counters_ptr))
+#define ERTS_FLXCTR_GET_CTR_PTR(C, SCHEDULER_ID, COUNTER_ID)            \
+    &(ERTS_FLXCTR_GET_CTR_ARRAY_PTR(C))->array[SCHEDULER_ID].counters[COUNTER_ID]
+
+
+typedef union {
+    erts_atomic_t counters[ERTS_FLXCTR_ATOMICS_PER_CACHE_LINE];
+    char pad[ERTS_CACHE_LINE_SIZE];
+} ErtsFlxCtrDecentralizedCtrArrayElem;
+
+typedef struct ErtsFlxCtrDecentralizedCtrArray {
+    void* block_start;
+    erts_atomic_t snapshot_status;
+    ErtsFlxCtrDecentralizedCtrArrayElem array[];
+} ErtsFlxCtrDecentralizedCtrArray;
+
+void erts_flxctr_set_slot(int group);
+
+ERTS_GLB_INLINE
+int erts_flxctr_get_slot_index(void);
+
+/* End: Internal Declarations */
+
+
+/* Implementation of inlined functions */
+
+#if ERTS_GLB_INLINE_INCL_FUNC_DEF
+
+ERTS_GLB_INLINE
+int erts_flxctr_get_slot_index(void)
+{
+    ErtsSchedulerData *esdp = erts_get_scheduler_data();
+    ASSERT(esdp && !ERTS_SCHEDULER_IS_DIRTY(esdp));
+    ASSERT(esdp->flxctr_slot_no > 0);
+    return esdp->flxctr_slot_no;
+}
+
+ERTS_GLB_INLINE
+void erts_flxctr_add(ErtsFlxCtr* c,
+                     Uint counter_nr,
+                     int to_add)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    if (c->is_decentralized) {
+        erts_atomic_add_nob(ERTS_FLXCTR_GET_CTR_PTR(c,
+                                                    erts_flxctr_get_slot_index(),
+                                                    counter_nr),
+                            to_add);
+    } else {
+        erts_atomic_add_nob(&c->u.counters[counter_nr], to_add);
+    }
+}
+
+ERTS_GLB_INLINE
+void erts_flxctr_inc(ErtsFlxCtr* c,
+                     Uint counter_nr)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    if (c->is_decentralized) {
+        erts_atomic_inc_nob(ERTS_FLXCTR_GET_CTR_PTR(c,
+                                                    erts_flxctr_get_slot_index(),
+                                                    counter_nr));
+    } else {
+        erts_atomic_inc_read_nob(&c->u.counters[counter_nr]);
+    }
+}
+
+ERTS_GLB_INLINE
+void erts_flxctr_dec(ErtsFlxCtr* c,
+                     Uint counter_nr)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    if (c->is_decentralized) {
+        erts_atomic_dec_nob(ERTS_FLXCTR_GET_CTR_PTR(c,
+                                                    erts_flxctr_get_slot_index(),
+                                                    counter_nr));
+    } else {
+        erts_atomic_dec_nob(&c->u.counters[counter_nr]);
+    }
+}
+
+ERTS_GLB_INLINE
+Sint erts_flxctr_inc_read_centralized(ErtsFlxCtr* c,
+                                      Uint counter_nr)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    ASSERT(!c->is_decentralized);
+    return erts_atomic_inc_read_nob(&c->u.counters[counter_nr]);
+}
+
+ERTS_GLB_INLINE
+Sint erts_flxctr_dec_read_centralized(ErtsFlxCtr* c,
+                                      Uint counter_nr)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    ASSERT(!c->is_decentralized);
+    return erts_atomic_dec_read_nob(&c->u.counters[counter_nr]);
+}
+
+ERTS_GLB_INLINE
+Sint erts_flxctr_read_centralized(ErtsFlxCtr* c,
+                                  Uint counter_nr)
+{
+    ASSERT(counter_nr < c->nr_of_counters);
+    ASSERT(!c->is_decentralized);   
+    return erts_atomic_read_nob(&((erts_atomic_t*)(c->u.counters))[counter_nr]);
+}
+
+#endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */
+
+#endif /* ERL_FLXCTR_H__ */
diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c
index 12750b9aa6..547e4064a2 100644
--- a/erts/emulator/beam/erl_init.c
+++ b/erts/emulator/beam/erl_init.c
@@ -593,6 +593,7 @@ void erts_usage(void)
     erts_fprintf(stderr, "               no_time_warp|single_time_warp|multi_time_warp\n");
     erts_fprintf(stderr, "-d             don't write a crash dump for internally detected errors\n");
     erts_fprintf(stderr, "               (halt(String) will still produce a crash dump)\n");
+    erts_fprintf(stderr, "-dcg           set the limit for the number of decentralized counter groups\n");
     erts_fprintf(stderr, "-fn[u|a|l]     Control how filenames are interpreted\n");
     erts_fprintf(stderr, "-hms size      set minimum heap size in words (default %d)\n",
 	       H_DEFAULT_SIZE);
@@ -785,6 +786,8 @@ early_init(int *argc, char **argv) /*
     int dirty_io_scheds;
     int max_reader_groups;
     int reader_groups;
+    int max_decentralized_counter_groups;
+    int decentralized_counter_groups;
     char envbuf[21]; /* enough for any 64-bit integer */
     size_t envbufsz;
 
@@ -804,7 +807,8 @@ early_init(int *argc, char **argv) /*
 
     erts_initialized = 0;
 
-    erts_pre_early_init_cpu_topology(&max_reader_groups,
+    erts_pre_early_init_cpu_topology(&max_decentralized_counter_groups,
+                                     &max_reader_groups,
 				     &ncpu,
 				     &ncpuonln,
 				     &ncpuavail);
@@ -865,6 +869,24 @@ early_init(int *argc, char **argv) /*
 	    }
 	    if (argv[i][0] == '-') {
 		switch (argv[i][1]) {
+		case 'd': {
+		    char *sub_param = argv[i]+2;
+		    if (has_prefix("cg", sub_param)) {
+			char *arg = get_arg(sub_param+2, argv[i+1], &i);
+			if (sscanf(arg, "%d", &max_decentralized_counter_groups) != 1) {
+			    erts_fprintf(stderr,
+					 "bad decentralized counter groups limit: %s\n", arg);
+			    erts_usage();
+			}
+			if (max_decentralized_counter_groups < 0) {
+			    erts_fprintf(stderr,
+					 "bad decentralized counter groups limit: %d\n",
+					 max_decentralized_counter_groups);
+			    erts_usage();
+			}
+		    }
+		    break;
+		}
 		case 'r': {
 		    char *sub_param = argv[i]+2;
 		    if (has_prefix("g", sub_param)) {
@@ -1186,8 +1208,10 @@ early_init(int *argc, char **argv) /*
     erts_early_init_cpu_topology(no_schedulers,
 				 &max_main_threads,
 				 max_reader_groups,
-				 &reader_groups);
-
+				 &reader_groups,
+                                 max_decentralized_counter_groups,
+                                 &decentralized_counter_groups);
+    erts_flxctr_setup(decentralized_counter_groups);
     {
 	erts_thr_late_init_data_t elid = ERTS_THR_LATE_INIT_DATA_DEF_INITER;
 	elid.mem.std.alloc = ethr_std_alloc;
diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 711b73417d..6118c671ee 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -641,6 +641,7 @@ struct ErtsSchedulerData_ {
     ErtsSchedType type;
     Uint no;			/* Scheduler number for normal schedulers */
     Uint dirty_no;  /* Scheduler number for dirty schedulers */
+    int flxctr_slot_no; /* slot nr when a flxctr is used */
     struct enif_environment_t *current_nif;
     Process *dirty_shadow_process;
     Port *current_port;
@@ -1847,6 +1848,7 @@ int erts_resume_processes(ErtsProcList *);
 void erts_deep_process_dump(fmtfn_t, void *);
 
 Eterm erts_get_reader_groups_map(Process *c_p);
+Eterm erts_get_decentralized_counter_groups_map(Process *c_p);
 Eterm erts_debug_reader_groups_map(Process *c_p, int groups);
 
 Uint erts_debug_nbalance(void);
diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl
index 87ca9bd32c..cd32f25abd 100644
--- a/lib/stdlib/test/ets_SUITE.erl
+++ b/lib/stdlib/test/ets_SUITE.erl
@@ -42,6 +42,8 @@
          select_bound_chunk/1,
 	 t_delete_all_objects/1, t_insert_list/1, t_test_ms/1,
 	 t_select_delete/1,t_select_replace/1,t_select_replace_next_bug/1,t_ets_dets/1]).
+-export([test_table_size_concurrency/1,test_table_memory_concurrency/1,
+         test_delete_table_while_size_snapshot/1, test_delete_table_while_size_snapshot_helper/0]).
 
 -export([ordered/1, ordered_match/1, interface_equality/1,
 	 fixtable_next/1, fixtable_insert/1, rename/1, rename_unnamed/1, evil_rename/1,
@@ -156,7 +158,11 @@ all() ->
      whereis_table,
      delete_unfix_race,
      test_throughput_benchmark,
-     {group, benchmark}].
+     {group, benchmark},
+     test_table_size_concurrency,
+     test_table_memory_concurrency,
+     test_delete_table_while_size_snapshot].
+
 
 groups() ->
     [{new, [],
@@ -828,7 +834,11 @@ adjust_xmem([_T1,_T2,_T3,_T4], {A0,B0,C0,D0} = _Mem0, EstCnt) ->
 
     {TabSz, EstSz} = erts_debug:get_internal_state('DbTable_words'),
     HTabSz = TabSz + EstCnt*EstSz,
-    {A0+TabSz, B0+HTabSz, C0+HTabSz, D0+HTabSz}.
+    OrdSetExtra = case erlang:system_info(wordsize) of
+                      8 -> 40; % larger stack on 64 bit architectures
+                      _ -> 0
+                  end,
+    {A0+TabSz+OrdSetExtra, B0+HTabSz, C0+HTabSz, D0+HTabSz}.
 
 %% Misc. whitebox tests
 t_whitebox(Config) when is_list(Config) ->
@@ -4102,6 +4112,11 @@ slot_do(Opts) ->
     fill_tab(Tab,foo),
     Elts = ets:info(Tab,size),
     Elts = slot_loop(Tab,0,0),
+    case ets:info(Tab, type) of
+        ordered_set ->
+            '$end_of_table' = ets:slot(Tab,Elts);
+        _ -> ok
+    end,
     true = ets:delete(Tab),
     verify_etsmem(EtsMem).
 
@@ -4453,6 +4468,127 @@ info_do(Opts) ->
     undefined = ets:info(non_existing_table_xxyy,safe_fixed),
     verify_etsmem(EtsMem).
 
+size_loop(_T, 0, _, _) ->
+    ok;
+size_loop(T, I, PrevSize, WhatToTest) ->
+    Size = ets:info(T, WhatToTest),
+    case Size < PrevSize of
+        true -> ct:fail("Bad ets:info/2");
+        _ -> ok
+    end,
+    size_loop(T, I -1, Size, WhatToTest).
+
+add_loop(_T, 0) ->
+    ok;
+add_loop(T, I) ->
+    ets:insert(T, {I}),
+    add_loop(T, I -1).
+
+
+test_table_counter_concurrency(WhatToTest) ->
+    ItemsToAdd = 1000000,
+    SizeLoopSize = 1000,
+    T = ets:new(k, [public, ordered_set, {write_concurrency, true}]),
+    0 = ets:info(T, size),
+    P = self(),
+    SpawnedSizeProcs =
+        [spawn(fun() -> 
+                       size_loop(T, SizeLoopSize, 0, WhatToTest),
+                       P ! done
+               end)
+         || _ <- lists:seq(1, 6)],
+    spawn(fun() -> 
+                  add_loop(T, ItemsToAdd),
+                  P ! done_add
+          end),
+    [receive
+         done -> ok;
+         done_add -> ok
+     end
+     || _ <- [ok|SpawnedSizeProcs]],
+    case WhatToTest =:= size of
+        true ->
+            ItemsToAdd = ets:info(T, size);
+        _ ->
+            ok
+    end,
+    ok.
+
+test_table_size_concurrency(Config) when is_list(Config) ->
+    test_table_counter_concurrency(size).
+
+test_table_memory_concurrency(Config) when is_list(Config) ->
+    test_table_counter_concurrency(memory).
+
+%% Tests that calling the ets:delete operation on a table T with
+%% decentralized counters works while ets:info(T, size) operations are
+%% active
+test_delete_table_while_size_snapshot(Config) when is_list(Config) ->
+    %% Run test case in a slave node as other test suites in stdlib
+    %% depend on that pids are ordered in creation order which is no
+    %% longer the case when many processes have been started before
+    Node = start_slave(),
+    ok = rpc:call(Node, ?MODULE, test_delete_table_while_size_snapshot_helper, []),
+    test_server:stop_node(Node),
+    ok.
+
+test_delete_table_while_size_snapshot_helper()->
+    TopParent = self(),
+    repeat_par(
+      fun() ->
+              Table = ets:new(t, [public, ordered_set,
+                                  {write_concurrency, true}]),
+              Parent = self(),
+              NrOfSizeProcs = 100,
+              Pids = [ spawn(fun()-> size_process(Table, Parent) end)
+                       || _ <- lists:seq(1, NrOfSizeProcs)],
+              timer:sleep(1),
+              ets:delete(Table),
+              [receive 
+                   table_gone ->  ok;
+                   Problem -> TopParent ! Problem
+               end || _ <- Pids]
+      end,
+      15000),
+    receive
+        Problem -> throw(Problem)
+    after 0 -> ok
+    end.
+
+size_process(Table, Parent) ->
+    try ets:info(Table, size) of
+        N when is_integer(N) ->
+            size_process(Table, Parent);
+        undefined -> Parent ! table_gone;
+        E -> Parent ! {got_unexpected, E}
+    catch
+        E -> Parent ! {got_unexpected_exception, E}
+    end.
+
+start_slave() ->
+    MicroSecs = erlang:monotonic_time(),
+    Name = "ets_" ++ integer_to_list(MicroSecs),
+    Pa = filename:dirname(code:which(?MODULE)),
+    {ok, Node} = test_server:start_node(list_to_atom(Name), slave, [{args, "-pa " ++ Pa}]),
+    Node.
+
+repeat_par(FunToRepeat, NrOfTimes) ->
+    repeat_par_help(FunToRepeat, NrOfTimes, NrOfTimes).
+
+repeat_par_help(_FunToRepeat, 0, OrgNrOfTimes) ->
+    repeat(fun()-> receive done -> ok end end, OrgNrOfTimes);
+repeat_par_help(FunToRepeat, NrOfTimes, OrgNrOfTimes) ->
+    Parent = self(),
+    case NrOfTimes rem 5 of
+        0 -> timer:sleep(1);
+        _ -> ok
+    end,
+    spawn(fun()->
+                  FunToRepeat(),
+                  Parent ! done
+          end),
+    repeat_par_help(FunToRepeat, NrOfTimes-1, OrgNrOfTimes).
+
 %% Test various duplicate_bags stuff.
 dups(Config) when is_list(Config) ->
     repeat_for_opts(fun dups_do/1).