erts: Make atomic ets:delete_all_objects yield

by using a cooperative strategy that will make
any process accessing the table execute delelete_all_objects_continue
until the table is empty.

This is not an optimal solution as concurrent threads will still
block on the table lock, but at least thread progress is made.

11 files changed, 448 insertions, 251 deletions

diff --git a/bootstrap/lib/stdlib/ebin/ets.beam b/bootstrap/lib/stdlib/ebin/ets.beam
index 3d103b1624..ab4996ef4e 100644
--- a/bootstrap/lib/stdlib/ebin/ets.beam
+++ b/bootstrap/lib/stdlib/ebin/ets.beam
diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 687fd39d58..1276048317 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -341,7 +341,6 @@ bif ets:internal_request_all/0
 bif ets:new/2
 bif ets:delete/1
 bif ets:delete/2
-bif ets:delete_all_objects/1
 bif ets:delete_object/2
 bif ets:first/1
 bif ets:is_compiled_ms/1
@@ -372,7 +371,6 @@ bif ets:select_count/2
 bif ets:select_reverse/1
 bif ets:select_reverse/2
 bif ets:select_reverse/3
-bif ets:select_delete/2
 bif ets:select_replace/2
 bif ets:match_spec_compile/1
 bif ets:match_spec_run_r/3
@@ -693,3 +691,5 @@ bif erts_internal:new_connection/1
 bif erts_internal:abort_connection/2
 bif erts_internal:map_next/3
 bif ets:whereis/1
+bif ets:internal_delete_all/2
+bif ets:internal_select_delete/2
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 89e3d3f43e..3c13991fb6 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -4731,7 +4731,14 @@ BIF_RETTYPE erts_debug_set_internal_state_2(BIF_ALIST_2)
                                             refbin));
             }
         }
-
+        else if (ERTS_IS_ATOM_STR("ets_force_trap", BIF_ARG_1)) {
+#ifdef ETS_DBG_FORCE_TRAP
+            erts_ets_dbg_force_trap = (BIF_ARG_2 == am_true) ? 1 : 0;
+            BIF_RET(am_ok);
+#else
+            BIF_RET(am_notsup);
+#endif
+        }
     }
 
     BIF_ERROR(BIF_P, BADARG);
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 2d46e81b3b..322f5f1505 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -50,6 +50,34 @@ erts_atomic_t erts_ets_misc_mem_size;
 ** Utility macros
 */
 
+#define DB_BIF_GET_TABLE(TB, WHAT, KIND, BIF_IX) \
+        DB_GET_TABLE(TB, BIF_ARG_1, WHAT, KIND, BIF_IX, NULL, BIF_P)
+
+#define DB_TRAP_GET_TABLE(TB, TID, WHAT, KIND, BIF_EXP) \
+        DB_GET_TABLE(TB, TID, WHAT, KIND, 0, BIF_EXP, BIF_P)
+
+#define DB_GET_TABLE(TB, TID, WHAT, KIND, BIF_IX, BIF_EXP, PROC)         \
+do {                                                                     \
+    Uint freason__;                                                      \
+    if (!(TB = db_get_table(PROC, TID, WHAT, KIND, &freason__))) {       \
+        return db_bif_fail(PROC, freason__, BIF_IX, BIF_EXP);            \
+    }                                                                    \
+}while(0)
+
+static BIF_RETTYPE db_bif_fail(Process* p, Uint freason,
+                               Uint bif_ix, Export* bif_exp)
+{
+    if (freason == TRAP) {
+        if (!bif_exp)
+            bif_exp = bif_export[bif_ix];
+        p->arity = bif_exp->info.mfa.arity;
+        p->i = (BeamInstr*) bif_exp->addressv[erts_active_code_ix()];
+    }
+    p->freason = freason;
+    return THE_NON_VALUE;
+}
+
+
 /* Get a key from any table structure and a tagged object */
 #define TERM_GETKEY(tb, obj) db_getkey((tb)->common.keypos, (obj)) 
 
@@ -326,8 +354,7 @@ struct meta_name_tab_entry* meta_name_tab_bucket(Eterm name,
 typedef enum {
     LCK_READ=1,     /* read only access */
     LCK_WRITE=2,    /* exclusive table write access */
-    LCK_WRITE_REC=3, /* record write access */
-    LCK_NONE=4
+    LCK_WRITE_REC=3 /* record write access */
 } db_lock_kind_t;
 
 extern DbTableMethod db_hash;
@@ -337,9 +364,6 @@ int user_requested_db_max_tabs;
 int erts_ets_realloc_always_moves;
 int erts_ets_always_compress;
 static int db_max_tabs;
-static Eterm ms_delete_all;
-static Eterm ms_delete_all_buff[8]; /* To compare with for deletion 
-				       of all objects */
 
 /* 
 ** Forward decls, static functions 
@@ -351,6 +375,7 @@ static void set_heir(Process* me, DbTable* tb, Eterm heir, UWord heir_data);
 static void free_heir_data(DbTable*);
 static SWord free_fixations_locked(Process* p, DbTable *tb);
 
+static void delete_all_objects_continue(Process* p, DbTable* tb);
 static SWord free_table_continue(Process *p, DbTable *tb, SWord reds);
 static void print_table(fmtfn_t to, void *to_arg, int show,  DbTable* tb);
 static BIF_RETTYPE ets_select_delete_1(BIF_ALIST_1);
@@ -360,9 +385,9 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1);
 static BIF_RETTYPE ets_delete_trap(BIF_ALIST_1);
 static Eterm table_info(Process* p, DbTable* tb, Eterm What);
 
-static BIF_RETTYPE ets_select1(Process* p, Eterm arg1);
-static BIF_RETTYPE ets_select2(Process* p, Eterm arg1, Eterm arg2);
-static BIF_RETTYPE ets_select3(Process* p, Eterm arg1, Eterm arg2, Eterm arg3);
+static BIF_RETTYPE ets_select1(Process* p, int bif_ix, Eterm arg1);
+static BIF_RETTYPE ets_select2(Process* p, DbTable*, Eterm tid, Eterm ms);
+static BIF_RETTYPE ets_select3(Process* p, DbTable*, Eterm tid, Eterm ms, Sint chunk_size);
 
 
 /* 
@@ -636,15 +661,42 @@ static ERTS_INLINE void db_unlock(DbTable* tb, db_lock_kind_t kind)
     }
 }
 
+static ERTS_INLINE int db_is_exclusive(DbTable* tb, db_lock_kind_t kind)
+{
+    return kind != LCK_READ && tb->common.is_thread_safe;
+}
+
+static DbTable* handle_lacking_permission(Process* p, DbTable* tb,
+                                          db_lock_kind_t kind,
+                                          Uint* freason_p)
+{
+    if (tb->common.status & DB_BUSY) {
+        if (!db_is_exclusive(tb, kind)) {
+            db_unlock(tb, kind);
+            db_lock(tb, LCK_WRITE);
+        }
+        delete_all_objects_continue(p, tb);
+        db_unlock(tb, LCK_WRITE);
+        tb = NULL;
+        *freason_p = TRAP;
+    }
+    else if (p->common.id != tb->common.owner) {
+        db_unlock(tb, kind);
+        tb = NULL;
+        *freason_p = BADARG;
+    }
+    return tb;
+}
+
 static ERTS_INLINE
 DbTable* db_get_table_aux(Process *p,
 			  Eterm id,
 			  int what,
 			  db_lock_kind_t kind,
-			  int meta_already_locked)
+			  int meta_already_locked,
+                          Uint* freason_p)
 {
     DbTable *tb;
-    erts_rwmtx_t *mtl = NULL;
 
     /*
      * IMPORTANT: Only scheduler threads are allowed
@@ -654,13 +706,13 @@ DbTable* db_get_table_aux(Process *p,
     ASSERT(erts_get_scheduler_data());
 
     if (is_atom(id)) {
+        erts_rwmtx_t *mtl;
 	struct meta_name_tab_entry* bucket = meta_name_tab_bucket(id,&mtl);
 	if (!meta_already_locked)
 	    erts_rwmtx_rlock(mtl);
 	else{
 	    ERTS_LC_ASSERT(erts_lc_rwmtx_is_rlocked(mtl)
 			       || erts_lc_rwmtx_is_rwlocked(mtl));
-	    mtl = NULL;
 	}
         tb = NULL;
 	if (bucket->pu.tb != NULL) {
@@ -679,20 +731,29 @@ DbTable* db_get_table_aux(Process *p,
 		}
 	    }
 	}
+        if (!meta_already_locked)
+            erts_rwmtx_runlock(mtl);
     }
     else
         tb = tid2tab(id);
 
     if (tb) {
 	db_lock(tb, kind);
-	if ((tb->common.status & what) == 0
-            && p->common.id != tb->common.owner) {
-	    db_unlock(tb, kind);
-	    tb = NULL;
-	}
+#ifdef ETS_DBG_FORCE_TRAP
+        if (erts_atomic_read_nob(&tb->common.dbg_force_trap) &&
+            erts_atomic_add_read_nob(&tb->common.dbg_force_trap, 2) & 2) {
+            db_unlock(tb, kind);
+            tb = NULL;
+            *freason_p = TRAP;
+        }
+        else
+#endif
+	if (ERTS_UNLIKELY(!(tb->common.status & what)))
+            tb = handle_lacking_permission(p, tb, kind, freason_p);
     }
-    if (mtl)
-	erts_rwmtx_runlock(mtl);
+    else
+        *freason_p = BADARG;
+
     return tb;
 }
 
@@ -700,9 +761,10 @@ static ERTS_INLINE
 DbTable* db_get_table(Process *p,
 		      Eterm id,
 		      int what,
-		      db_lock_kind_t kind)
+		      db_lock_kind_t kind,
+                      Uint* freason_p)
 {
-    return db_get_table_aux(p, id, what, kind, 0);
+    return db_get_table_aux(p, id, what, kind, 0, freason_p);
 }
 
 static int insert_named_tab(Eterm name_atom, DbTable* tb, int have_lock)
@@ -868,9 +930,7 @@ BIF_RETTYPE ets_safe_fixtable_2(BIF_ALIST_2)
 #endif
     kind = (BIF_ARG_2 == am_true) ? LCK_READ : LCK_WRITE_REC; 
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, kind)) == NULL) {
-	    BIF_ERROR(BIF_P, BADARG);
-	}
+    DB_BIF_GET_TABLE(tb, DB_READ, kind, BIF_ets_safe_fixtable_2);
 
     if (BIF_ARG_2 == am_true) {
 	fix_table_locked(BIF_P, tb);
@@ -900,11 +960,7 @@ BIF_RETTYPE ets_first_1(BIF_ALIST_1)
 
     CHECK_TABLES();
 
-    tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ);
-
-    if (!tb) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_first_1);
 
     cret = tb->common.meth->db_first(BIF_P, tb, &ret);
 
@@ -927,11 +983,7 @@ BIF_RETTYPE ets_next_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ);
-
-    if (!tb) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_next_2);
 
     cret = tb->common.meth->db_next(BIF_P, tb, BIF_ARG_2, &ret);
 
@@ -954,11 +1006,7 @@ BIF_RETTYPE ets_last_1(BIF_ALIST_1)
 
     CHECK_TABLES();
 
-    tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ);
-
-    if (!tb) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_last_1);
 
     cret = tb->common.meth->db_last(BIF_P, tb, &ret);
 
@@ -981,11 +1029,7 @@ BIF_RETTYPE ets_prev_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ);
-
-    if (!tb) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_prev_2);
 
     cret = tb->common.meth->db_prev(BIF_P,tb,BIF_ARG_2,&ret);
 
@@ -1003,21 +1047,15 @@ BIF_RETTYPE ets_prev_2(BIF_ALIST_2)
 BIF_RETTYPE ets_take_2(BIF_ALIST_2)
 {
     DbTable* tb;
-#ifdef DEBUG
     int cret;
-#endif
     Eterm ret;
     CHECK_TABLES();
 
-    tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC);
-    if (!tb) {
-        BIF_ERROR(BIF_P, BADARG);
-    }
-#ifdef DEBUG
-    cret =
-#endif
-        tb->common.meth->db_take(BIF_P, tb, BIF_ARG_2, &ret);
-    ASSERT(cret == DB_ERROR_NONE);
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_take_2);
+
+    cret = tb->common.meth->db_take(BIF_P, tb, BIF_ARG_2, &ret);
+
+    ASSERT(cret == DB_ERROR_NONE); (void)cret;
     db_unlock(tb, LCK_WRITE_REC);
     BIF_RET(ret);
 }
@@ -1035,9 +1073,8 @@ BIF_RETTYPE ets_update_element_3(BIF_ALIST_3)
     DeclareTmpHeap(cell,2,BIF_P);
     DbUpdateHandle handle;
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_update_element_3);
+
     UseTmpHeap(2,BIF_P);
     if (!(tb->common.status & (DB_SET | DB_ORDERED_SET))) {
 	goto bail_out;
@@ -1108,9 +1145,9 @@ bail_out:
 }
 
 static BIF_RETTYPE
-do_update_counter(Process *p, Eterm arg1, Eterm arg2, Eterm arg3, Eterm arg4)
+do_update_counter(Process *p, DbTable* tb,
+                  Eterm arg2, Eterm arg3, Eterm arg4)
 {
-    DbTable* tb;
     int cret = DB_ERROR_BADITEM;
     Eterm upop_list;
     int list_size;
@@ -1126,10 +1163,6 @@ do_update_counter(Process *p, Eterm arg1, Eterm arg2, Eterm arg3, Eterm arg4)
     Eterm* hstart;
     Eterm* hend;
 
-    if ((tb = db_get_table(p, arg1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-        BIF_ERROR(p, BADARG);
-    }
-
     UseTmpHeap(5, p);
 
     if (!(tb->common.status & (DB_SET | DB_ORDERED_SET))) {
@@ -1303,7 +1336,11 @@ bail_out:
 */
 BIF_RETTYPE ets_update_counter_3(BIF_ALIST_3)
 {
-    return do_update_counter(BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3, THE_NON_VALUE);
+    DbTable* tb;
+
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_update_counter_3);
+
+    return do_update_counter(BIF_P, tb, BIF_ARG_2, BIF_ARG_3, THE_NON_VALUE);
 }
 
 /*
@@ -1315,10 +1352,14 @@ BIF_RETTYPE ets_update_counter_3(BIF_ALIST_3)
 */
 BIF_RETTYPE ets_update_counter_4(BIF_ALIST_4)
 {
+    DbTable* tb;
+
     if (is_not_tuple(BIF_ARG_4)) {
         BIF_ERROR(BIF_P, BADARG);
     }
-    return do_update_counter(BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3, BIF_ARG_4);
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_update_counter_4);
+
+    return do_update_counter(BIF_P, tb, BIF_ARG_2, BIF_ARG_3, BIF_ARG_4);
 }
 
 
@@ -1339,9 +1380,8 @@ BIF_RETTYPE ets_insert_2(BIF_ALIST_2)
     kind = ((is_list(BIF_ARG_2) && CDR(list_val(BIF_ARG_2)) != NIL)
 	    ? LCK_WRITE : LCK_WRITE_REC);
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, kind)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, kind, BIF_ets_insert_2);
+
     if (BIF_ARG_2 == NIL) {
 	db_unlock(tb, kind);
 	BIF_RET(am_true);
@@ -1407,11 +1447,9 @@ BIF_RETTYPE ets_insert_new_2(BIF_ALIST_2)
 
 	    /* More than one object, use LCK_WRITE to keep atomicity */
 	    kind = LCK_WRITE;
-	    tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, kind);
-	    if (tb == NULL) {
-		BIF_ERROR(BIF_P, BADARG);
-	    }
-	    meth = tb->common.meth;
+            DB_BIF_GET_TABLE(tb, DB_WRITE, kind, BIF_ets_insert_new_2);
+
+            meth = tb->common.meth;
 	    for (lst = BIF_ARG_2; is_list(lst); lst = CDR(list_val(lst))) {
 		if (is_not_tuple(CAR(list_val(lst)))
 		    || (arityval(*tuple_val(CAR(list_val(lst))))
@@ -1446,9 +1484,8 @@ BIF_RETTYPE ets_insert_new_2(BIF_ALIST_2)
     /* Only one object (or NIL) 
     */
     kind = LCK_WRITE_REC;
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, kind)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, kind, BIF_ets_insert_new_2);
+
     if (BIF_ARG_2 == NIL) {
 	db_unlock(tb, kind);
 	BIF_RET(am_true);
@@ -1487,6 +1524,7 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2)
     Eterm ret;
     Eterm old_name;
     erts_rwmtx_t *lck1, *lck2;
+    Uint freason;
 
 #ifdef HARDDEBUG
     erts_fprintf(stderr,
@@ -1531,9 +1569,9 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2)
     if (lck2)
 	erts_rwmtx_rwlock(lck2);
 
-    tb = db_get_table_aux(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE, 1);
+    tb = db_get_table_aux(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE, 1, &freason);
     if (!tb)
-	goto badarg;
+	goto fail;
 
     if (is_table_named(tb)) {
         if (!insert_named_tab(BIF_ARG_2, tb, 1))
@@ -1553,13 +1591,18 @@ BIF_RETTYPE ets_rename_2(BIF_ALIST_2)
     if (lck2)
 	erts_rwmtx_rwunlock(lck2);
     BIF_RET(ret);
- badarg:
+
+badarg:
+    freason = BADARG;
+
+fail:
     if (tb)
 	db_unlock(tb, LCK_WRITE);
     erts_rwmtx_rwunlock(lck1);
     if (lck2)
 	erts_rwmtx_rwunlock(lck2);
-    BIF_ERROR(BIF_P, BADARG);    
+
+    return db_bif_fail(BIF_P, freason, BIF_ets_rename_2, NULL);
 }
 
 
@@ -1706,7 +1749,7 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
     tb->common.meth = meth;
     tb->common.the_name = BIF_ARG_1;
     tb->common.status = status;    
-    tb->common.type = status & ERTS_ETS_TABLE_TYPES;
+    tb->common.type = status;
     /* Note, 'type' is *read only* from now on... */
     erts_refc_init(&tb->common.fix_count, 0);
     db_init_lock(tb, status & (DB_FINE_LOCKED|DB_FREQ_READ));
@@ -1718,6 +1761,9 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
 
     tb->common.fixing_procs = NULL;
     tb->common.compress = is_compressed;
+#ifdef ETS_DBG_FORCE_TRAP
+    erts_atomic_init_nob(&tb->common.dbg_force_trap, erts_ets_dbg_force_trap);
+#endif
 
     cret = meth->db_create(BIF_P, tb);
     ASSERT(cret == DB_ERROR_NONE); (void)cret;
@@ -1762,13 +1808,19 @@ BIF_RETTYPE ets_whereis_1(BIF_ALIST_1)
 {
     DbTable* tb;
     Eterm res;
+    Uint freason;
 
     if (is_not_atom(BIF_ARG_1)) {
         BIF_ERROR(BIF_P, BADARG);
     }
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ)) == NULL) {
-        BIF_RET(am_undefined);
+    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ, &freason)) == NULL) {
+        if (freason == BADARG)
+            BIF_RET(am_undefined);
+        else {
+            //ToDo: Could we avoid this
+            return db_bif_fail(BIF_P, freason, BIF_ets_whereis_1, NULL);
+        }
     }
 
     res = make_tid(BIF_P, tb);
@@ -1788,9 +1840,7 @@ BIF_RETTYPE ets_lookup_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_lookup_2);
 
     cret = tb->common.meth->db_get(BIF_P, tb, BIF_ARG_2, &ret);
 
@@ -1818,9 +1868,7 @@ BIF_RETTYPE ets_member_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_member_2);
 
     cret = tb->common.meth->db_member(tb, BIF_ARG_2, &ret);
 
@@ -1851,9 +1899,7 @@ BIF_RETTYPE ets_lookup_element_3(BIF_ALIST_3)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_lookup_element_3);
 
     if (is_not_small(BIF_ARG_3) || ((index = signed_val(BIF_ARG_3)) < 1)) {
 	db_unlock(tb, LCK_READ);
@@ -1891,9 +1937,7 @@ BIF_RETTYPE ets_delete_1(BIF_ALIST_1)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE, BIF_ets_delete_1);
     
     /*
      * Clear all access bits to prevent any ets operation to access the
@@ -1966,6 +2010,7 @@ BIF_RETTYPE ets_give_away_3(BIF_ALIST_3)
     Eterm to_pid = BIF_ARG_2;
     Eterm from_pid;
     DbTable* tb = NULL;
+    Uint freason;
 
     if (!is_internal_pid(to_pid)) {
 	goto badarg;
@@ -1975,10 +2020,11 @@ BIF_RETTYPE ets_give_away_3(BIF_ALIST_3)
 	goto badarg;
     }
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL
-	|| tb->common.owner != BIF_P->common.id) {
-	goto badarg;
-    }
+    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE, &freason)) == NULL)
+        goto fail;
+    if (tb->common.owner != BIF_P->common.id)
+        goto badarg;
+
     from_pid = tb->common.owner;
     if (to_pid == from_pid) {
 	goto badarg;  /* or should we be idempotent? return false maybe */
@@ -1997,9 +2043,12 @@ BIF_RETTYPE ets_give_away_3(BIF_ALIST_3)
     BIF_RET(am_true);
 
 badarg:
+    freason = BADARG;
+fail:
     if (to_proc != NULL && to_proc != BIF_P) erts_proc_unlock(to_proc, to_locks);
     if (tb != NULL) db_unlock(tb, LCK_WRITE);
-    BIF_ERROR(BIF_P, BADARG);
+
+    return db_bif_fail(BIF_P, freason, BIF_ets_give_away_3, NULL);
 }
 
 BIF_RETTYPE ets_setopts_2(BIF_ALIST_2)
@@ -2050,11 +2099,13 @@ BIF_RETTYPE ets_setopts_2(BIF_ALIST_2)
 	}
     }
 
-    if (tail != NIL
-	|| (tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL
-	|| tb->common.owner != BIF_P->common.id) {
+    if (tail != NIL)
+        goto badarg;
+
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE, BIF_ets_setopts_2);
+
+    if (tb->common.owner != BIF_P->common.id)
 	goto badarg;
-    }
 
     if (heir_data != THE_NON_VALUE) {
 	free_heir_data(tb);
@@ -2078,23 +2129,84 @@ badarg:
 }
 
 /* 
-** BIF to erase a whole table and release all memory it holds 
-*/
-BIF_RETTYPE ets_delete_all_objects_1(BIF_ALIST_1)
+ * Common for delete_all_objects and select_delete(DeleteAll).
+ */
+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;
     DbTable* tb;
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    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);
 
-    tb->common.meth->db_delete_all_objects(BIF_P, tb);
+        ASSERT(!(tb->common.status & DB_BUSY));
+
+        if (reds < 0) {
+            /*
+             * Oboy, need to trap AND need to be atomic.
+             * Solved by cooperative trapping where every process trying to
+             * access this table (including this process) will "fail" to lookup
+             * the table and instead pitch in deleting objects
+             * (in delete_all_objects_continue) and then trap to self.
+             */
+            ASSERT((tb->common.status & (DB_PRIVATE|DB_PROTECTED|DB_PUBLIC))
+                   ==
+                   (tb->common.type & (DB_PRIVATE|DB_PROTECTED|DB_PUBLIC)));
+            tb->common.status &= ~(DB_PRIVATE|DB_PROTECTED|DB_PUBLIC);
+            tb->common.status |= DB_BUSY;
+            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);
+        }
+        else {
+            /* Done, no trapping needed */
+            BUMP_REDS(BIF_P, (initial_reds - reds));
+        }
+
+    }
+    else {
+        /*
+         * The table lookup succeeded and second argument is nitems
+         * and not 'undefined', which means we have trapped at least once
+         * and are now done.
+         */
+        nitems = BIF_ARG_2;
+    }
 
     db_unlock(tb, LCK_WRITE);
+    BIF_RET(nitems);
+}
 
-    BIF_RET(am_true);
+static void delete_all_objects_continue(Process* p, DbTable* tb)
+{
+    SWord initial_reds = ERTS_BIF_REDS_LEFT(p);
+    SWord reds = initial_reds;
+
+    ERTS_LC_ASSERT(erts_lc_rwmtx_is_rwlocked(&tb->common.rwlock));
+
+    if ((tb->common.status & (DB_DELETE|DB_BUSY)) != DB_BUSY)
+        return;
+
+    reds = tb->common.meth->db_delete_all_objects(p, tb, reds);
+    
+    if (reds < 0) {
+        BUMP_ALL_REDS(p);
+    }
+    else {
+        tb->common.status |= tb->common.type & (DB_PRIVATE|DB_PROTECTED|DB_PUBLIC);
+        tb->common.status &= ~DB_BUSY;
+        BUMP_REDS(p, (initial_reds - reds));
+    }
 }
 
 /* 
@@ -2110,9 +2222,7 @@ BIF_RETTYPE ets_delete_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_delete_2);
 
     cret = tb->common.meth->db_erase(tb,BIF_ARG_2,&ret);
 
@@ -2139,9 +2249,8 @@ BIF_RETTYPE ets_delete_object_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_delete_object_2);
+
     if (is_not_tuple(BIF_ARG_2) || 
 	(arityval(*tuple_val(BIF_ARG_2)) < tb->common.keypos)) {
 	db_unlock(tb, LCK_WRITE_REC);
@@ -2174,15 +2283,14 @@ static BIF_RETTYPE ets_select_delete_1(BIF_ALIST_1)
     Eterm ret;
     Eterm *tptr;
     db_lock_kind_t kind = LCK_WRITE_REC;
-    
+
     CHECK_TABLES();
     ASSERT(is_tuple(a1));
     tptr = tuple_val(a1);
     ASSERT(arityval(*tptr) >= 1);
     
-    if ((tb = db_get_table(p, tptr[1], DB_WRITE, kind)) == NULL) {
-	BIF_ERROR(p,BADARG);
-    }
+    DB_TRAP_GET_TABLE(tb, tptr[1], DB_WRITE, kind,
+                      &ets_select_delete_continue_exp);
 
     cret = tb->common.meth->db_select_delete_continue(p,tb,a1,&ret);
 
@@ -2206,7 +2314,10 @@ static BIF_RETTYPE ets_select_delete_1(BIF_ALIST_1)
 }
     
 
-BIF_RETTYPE ets_select_delete_2(BIF_ALIST_2)
+/*
+ * ets:select_delete/2 without special case for "delete-all".
+ */
+BIF_RETTYPE ets_internal_select_delete_2(BIF_ALIST_2)
 {
     BIF_RETTYPE result;
     DbTable* tb;
@@ -2216,20 +2327,8 @@ BIF_RETTYPE ets_select_delete_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if(eq(BIF_ARG_2, ms_delete_all)) {
-	int nitems;
-	if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE)) == NULL) {
-	    BIF_ERROR(BIF_P, BADARG);
-	}
-	nitems = erts_atomic_read_nob(&tb->common.nitems);
-	tb->common.meth->db_delete_all_objects(BIF_P, tb);
-	db_unlock(tb, LCK_WRITE);
-	BIF_RET(erts_make_integer(nitems,BIF_P));
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_internal_select_delete_2);
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
     safety = ITERATION_SAFETY(BIF_P,tb);
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
@@ -2521,9 +2620,8 @@ BIF_RETTYPE ets_slot_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_slot_2);
+
     /* The slot number is checked in table specific code. */
     cret = tb->common.meth->db_slot(BIF_P, tb, BIF_ARG_2, &ret);
     db_unlock(tb, LCK_READ);
@@ -2543,41 +2641,53 @@ BIF_RETTYPE ets_slot_2(BIF_ALIST_2)
 
 BIF_RETTYPE ets_match_1(BIF_ALIST_1)
 {
-    return ets_select1(BIF_P, BIF_ARG_1);
+    return ets_select1(BIF_P, BIF_ets_match_1, BIF_ARG_1);
 }
 
 BIF_RETTYPE ets_match_2(BIF_ALIST_2)
 {
+    DbTable* tb;
     Eterm ms;
     DeclareTmpHeap(buff,8,BIF_P);
     Eterm *hp = buff;
     Eterm res;
 
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_match_2);
+
     UseTmpHeap(8,BIF_P);
     ms = CONS(hp, am_DollarDollar, NIL);
     hp += 2;
     ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
     hp += 4;
     ms = CONS(hp, ms, NIL);
-    res = ets_select2(BIF_P, BIF_ARG_1, ms);
+    res = ets_select2(BIF_P, tb, BIF_ARG_1, ms);
     UnUseTmpHeap(8,BIF_P);
     return res;
 }
 
 BIF_RETTYPE ets_match_3(BIF_ALIST_3)
 {
+    DbTable* tb;
     Eterm ms;
+    Sint chunk_size;
     DeclareTmpHeap(buff,8,BIF_P);
     Eterm *hp = buff;
     Eterm res;
 
+    /* Chunk size strictly greater than 0 */
+    if (is_not_small(BIF_ARG_3) || (chunk_size = signed_val(BIF_ARG_3)) <= 0) {
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_match_3);
+
     UseTmpHeap(8,BIF_P);
     ms = CONS(hp, am_DollarDollar, NIL);
     hp += 2;
     ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
     hp += 4;
     ms = CONS(hp, ms, NIL);
-    res = ets_select3(BIF_P, BIF_ARG_1, ms, BIF_ARG_3);
+    res = ets_select3(BIF_P, tb, BIF_ARG_1, ms, chunk_size);
     UnUseTmpHeap(8,BIF_P);
     return res;
 }
@@ -2585,34 +2695,35 @@ BIF_RETTYPE ets_match_3(BIF_ALIST_3)
 
 BIF_RETTYPE ets_select_3(BIF_ALIST_3)
 {
-    return ets_select3(BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3);
+    DbTable* tb;
+    Sint chunk_size;
+
+    /* Chunk size strictly greater than 0 */
+    if (is_not_small(BIF_ARG_3) || (chunk_size = signed_val(BIF_ARG_3)) <= 0) {
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_3);
+
+    return ets_select3(BIF_P, tb, BIF_ARG_1, BIF_ARG_2, chunk_size);
 }
 
 static BIF_RETTYPE
-ets_select3(Process* p, Eterm arg1, Eterm arg2, Eterm arg3)
+ets_select3(Process* p, DbTable* tb, Eterm tid, Eterm ms, Sint chunk_size)
 {
     BIF_RETTYPE result;
-    DbTable* tb;
     int cret;
     Eterm ret;
-    Sint chunk_size;
     enum DbIterSafety safety;
 
     CHECK_TABLES();
 
-    /* Chunk size strictly greater than 0 */
-    if (is_not_small(arg3) || (chunk_size = signed_val(arg3)) <= 0) {
-	BIF_ERROR(p, BADARG);
-    }
-    if ((tb = db_get_table(p, arg1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(p, BADARG);
-    }
     safety = ITERATION_SAFETY(p,tb);
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
     }
-    cret = tb->common.meth->db_select_chunk(p, tb, arg1,
-					    arg2, chunk_size,
+    cret = tb->common.meth->db_select_chunk(p, tb, tid,
+					    ms, chunk_size,
 					    0 /* not reversed */,
 					    &ret);
     if (DID_TRAP(p,ret) && safety != ITER_SAFE) {
@@ -2658,9 +2769,8 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
     tptr = tuple_val(a1);
     ASSERT(arityval(*tptr) >= 1);
 
-    if ((tb = db_get_table(p, tptr[1], DB_READ, kind)) == NULL) {
-	BIF_ERROR(p, BADARG);
-    }
+    DB_TRAP_GET_TABLE(tb, tptr[1], DB_READ, kind,
+                      &ets_select_continue_exp);
 
     cret = tb->common.meth->db_select_continue(p, tb, a1,
 					       &ret);
@@ -2690,10 +2800,10 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
 
 BIF_RETTYPE ets_select_1(BIF_ALIST_1)
 {
-    return ets_select1(BIF_P, BIF_ARG_1);
+    return ets_select1(BIF_P, BIF_ets_select_1, BIF_ARG_1);
 }
 
-static BIF_RETTYPE ets_select1(Process *p, Eterm arg1)
+static BIF_RETTYPE ets_select1(Process *p, int bif_ix, Eterm arg1)
 {
     BIF_RETTYPE result;
     DbTable* tb;
@@ -2715,10 +2825,10 @@ static BIF_RETTYPE ets_select1(Process *p, Eterm arg1)
 	BIF_ERROR(p, BADARG);
     }
     tptr = tuple_val(arg1);
-    if (arityval(*tptr) < 1 ||
-	(tb = db_get_table(p, tptr[1], DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(p, BADARG);
-    }
+    if (arityval(*tptr) < 1)
+        BIF_ERROR(p, BADARG);
+
+    DB_GET_TABLE(tb, tptr[1], DB_READ, LCK_READ, bif_ix, NULL, p);
 
     safety = ITERATION_SAFETY(p,tb);
     if (safety == ITER_UNSAFE) {
@@ -2754,33 +2864,27 @@ static BIF_RETTYPE ets_select1(Process *p, Eterm arg1)
 
 BIF_RETTYPE ets_select_2(BIF_ALIST_2)
 {
-    return ets_select2(BIF_P, BIF_ARG_1, BIF_ARG_2);
+    DbTable* tb;
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_2);
+    return ets_select2(BIF_P, tb, BIF_ARG_1, BIF_ARG_2);
 }
 
 static BIF_RETTYPE
-ets_select2(Process* p, Eterm arg1, Eterm arg2)
+ets_select2(Process* p, DbTable* tb, Eterm tid, Eterm ms)
 {
     BIF_RETTYPE result;
-    DbTable* tb;
     int cret;
     enum DbIterSafety safety;
     Eterm ret;
 
     CHECK_TABLES();
 
-    /*
-     * Make sure that the table exists.
-     */
-
-    if ((tb = db_get_table(p, arg1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(p, BADARG);
-    }
     safety = ITERATION_SAFETY(p,tb);
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
     }
 
-    cret = tb->common.meth->db_select(p, tb, arg1, arg2, 0, &ret);
+    cret = tb->common.meth->db_select(p, tb, tid, ms, 0, &ret);
 
     if (DID_TRAP(p,ret) && safety != ITER_SAFE) {
 	fix_table_locked(p, tb);
@@ -2823,9 +2927,9 @@ static BIF_RETTYPE ets_select_count_1(BIF_ALIST_1)
 
     tptr = tuple_val(a1);
     ASSERT(arityval(*tptr) >= 1);
-    if ((tb = db_get_table(p, tptr[1], DB_READ, kind)) == NULL) {
-	BIF_ERROR(p, BADARG);
-    }
+
+    DB_TRAP_GET_TABLE(tb, tptr[1], DB_READ, kind,
+                      &ets_select_count_continue_exp);
 
     cret = tb->common.meth->db_select_count_continue(p, tb, a1, &ret);
 
@@ -2860,13 +2964,9 @@ BIF_RETTYPE ets_select_count_2(BIF_ALIST_2)
     Eterm ret;
 
     CHECK_TABLES();
-    /*
-     * Make sure that the table exists.
-     */
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_count_2);
+
     safety = ITERATION_SAFETY(BIF_P,tb);
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
@@ -2916,9 +3016,8 @@ static BIF_RETTYPE ets_select_replace_1(BIF_ALIST_1)
     tptr = tuple_val(a1);
     ASSERT(arityval(*tptr) >= 1);
 
-    if ((tb = db_get_table(p, tptr[1], DB_WRITE, kind)) == NULL) {
-        BIF_ERROR(p,BADARG);
-    }
+    DB_TRAP_GET_TABLE(tb, tptr[1], DB_WRITE, kind,
+                      &ets_select_replace_continue_exp);
 
     cret = tb->common.meth->db_select_replace_continue(p,tb,a1,&ret);
 
@@ -2952,9 +3051,7 @@ BIF_RETTYPE ets_select_replace_2(BIF_ALIST_2)
 
     CHECK_TABLES();
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
-        BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_select_replace_2);
 
     if (tb->common.status & DB_BAG) {
         /* Bag implementation presented both semantic consistency
@@ -3005,13 +3102,8 @@ BIF_RETTYPE ets_select_reverse_3(BIF_ALIST_3)
     Sint chunk_size;
 
     CHECK_TABLES();
-    /*
-     * Make sure that the table exists.
-     */
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_reverse_3);
     
     /* Chunk size strictly greater than 0 */
     if (is_not_small(BIF_ARG_3) || (chunk_size = signed_val(BIF_ARG_3)) <= 0) {
@@ -3049,7 +3141,7 @@ BIF_RETTYPE ets_select_reverse_3(BIF_ALIST_3)
 
 BIF_RETTYPE ets_select_reverse_1(BIF_ALIST_1)
 {
-    return ets_select1(BIF_P, BIF_ARG_1);
+    return ets_select1(BIF_P, BIF_ets_select_reverse_1, BIF_ARG_1);
 }
 
 BIF_RETTYPE ets_select_reverse_2(BIF_ALIST_2)
@@ -3061,13 +3153,9 @@ BIF_RETTYPE ets_select_reverse_2(BIF_ALIST_2)
     Eterm ret;
 
     CHECK_TABLES();
-    /*
-     * Make sure that the table exists.
-     */
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_READ, LCK_READ)) == NULL) {
-	BIF_ERROR(BIF_P, BADARG);
-    }
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_reverse_2);
+
     safety = ITERATION_SAFETY(BIF_P,tb);
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
@@ -3099,45 +3187,63 @@ BIF_RETTYPE ets_select_reverse_2(BIF_ALIST_2)
 
 
 /* 
-** ets:match_object(Continuation), ets:match_object(Table, Pattern), ets:match_object(Table,Pattern,ChunkSize) 
+** ets:match_object(Continuation)
 */
 BIF_RETTYPE ets_match_object_1(BIF_ALIST_1)
 {
-    return ets_select1(BIF_P, BIF_ARG_1);
+    return ets_select1(BIF_P, BIF_ets_match_object_1, BIF_ARG_1);
 }
 
+/*
+** ets:match_object(Table, Pattern)
+*/
 BIF_RETTYPE ets_match_object_2(BIF_ALIST_2)
 {
+    DbTable* tb;
     Eterm ms;
     DeclareTmpHeap(buff,8,BIF_P);
     Eterm *hp = buff;
     Eterm res;
 
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_match_object_2);
+
     UseTmpHeap(8,BIF_P);
     ms = CONS(hp, am_DollarUnderscore, NIL);
     hp += 2;
     ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
     hp += 4;
     ms = CONS(hp, ms, NIL);
-    res = ets_select2(BIF_P, BIF_ARG_1, ms);
+    res = ets_select2(BIF_P, tb, BIF_ARG_1, ms);
     UnUseTmpHeap(8,BIF_P);
     return res;
 }
 
+/*
+** ets:match_object(Table,Pattern,ChunkSize)
+*/
 BIF_RETTYPE ets_match_object_3(BIF_ALIST_3)
 {
+    DbTable* tb;
+    Sint chunk_size;
     Eterm ms;
     DeclareTmpHeap(buff,8,BIF_P);
     Eterm *hp = buff;
     Eterm res;
 
+    /* Chunk size strictly greater than 0 */
+    if (is_not_small(BIF_ARG_3) || (chunk_size = signed_val(BIF_ARG_3)) <= 0) {
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
+    DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_match_object_3);
+
     UseTmpHeap(8,BIF_P);
     ms = CONS(hp, am_DollarUnderscore, NIL);
     hp += 2;
     ms = TUPLE3(hp, BIF_ARG_2, NIL, ms); 
     hp += 4;
     ms = CONS(hp, ms, NIL);
-    res = ets_select3(BIF_P, BIF_ARG_1, ms, BIF_ARG_3);
+    res = ets_select3(BIF_P, tb, BIF_ARG_1, ms, chunk_size);
     UnUseTmpHeap(8,BIF_P);
     return res;
 }
@@ -3158,16 +3264,17 @@ BIF_RETTYPE ets_info_1(BIF_ALIST_1)
     Eterm res;
     int i;
     Eterm* hp;
+    Uint freason;
     /*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)) == NULL) {
-	if (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)) {
+    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);
-	}
-	BIF_ERROR(BIF_P, BADARG);
+	else
+            return db_bif_fail(BIF_P, freason, BIF_ets_info_1, NULL);
     }
 
     /* If/when we implement lockless private tables:
@@ -3224,12 +3331,13 @@ BIF_RETTYPE ets_info_2(BIF_ALIST_2)
 {
     DbTable* tb;
     Eterm ret = THE_NON_VALUE;
+    Uint freason;
 
-    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_INFO, LCK_READ)) == NULL) {
-	if (is_atom(BIF_ARG_1) || is_ref(BIF_ARG_1)) {
+    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);
-	}
-	BIF_ERROR(BIF_P, BADARG);
+        else
+            return db_bif_fail(BIF_P, freason, BIF_ets_info_2, NULL);
     }
     ret = table_info(BIF_P, tb, BIF_ARG_2);
     db_unlock(tb, LCK_READ);
@@ -3317,7 +3425,6 @@ int erts_ets_rwmtx_spin_count = -1;
 void init_db(ErtsDbSpinCount db_spin_count)
 {
     int i;
-    Eterm *hp;
     unsigned bits;
     size_t size;
 
@@ -3420,13 +3527,6 @@ void init_db(ErtsDbSpinCount db_spin_count)
     erts_init_trap_export(&ets_delete_continue_exp,
 			  am_ets, am_atom_put("delete_trap",11), 1,
 			  &ets_delete_trap);
-
-    hp = ms_delete_all_buff;
-    ms_delete_all = CONS(hp, am_true, NIL);
-    hp += 2;
-    ms_delete_all = TUPLE3(hp,am_Underscore,NIL,ms_delete_all);
-    hp +=4;
-    ms_delete_all = CONS(hp, ms_delete_all,NIL);
 }
 
 void
@@ -3788,7 +3888,8 @@ unlocked:
 	    erts_rwmtx_runlock(&tb->common.rwlock);
 	    erts_rwmtx_rwlock(&tb->common.rwlock);
 	    *kind_p = LCK_WRITE;
-	    if (tb->common.status & DB_DELETE) return;
+	    if (tb->common.status & (DB_DELETE|DB_BUSY))
+                return;
 	}
 	db_unfix_table_hash(&(tb->hash));
     }
@@ -4318,5 +4419,8 @@ void erts_lcnt_update_db_locks(int enable) {
     erts_schedule_multi_misc_aux_work(0, erts_no_schedulers,
         &lcnt_update_db_locks_per_sched, (void*)(UWord)enable);
 }
-
 #endif /* ERTS_ENABLE_LOCK_COUNT */
+
+#ifdef ETS_DBG_FORCE_TRAP
+erts_aint_t erts_ets_dbg_force_trap = 0;
+#endif
diff --git a/erts/emulator/beam/erl_db.h b/erts/emulator/beam/erl_db.h
index eb6da2c9fb..db86c81914 100644
--- a/erts/emulator/beam/erl_db.h
+++ b/erts/emulator/beam/erl_db.h
@@ -135,6 +135,10 @@ void erts_lcnt_enable_db_lock_count(DbTable *tb, int enable);
 void erts_lcnt_update_db_locks(int enable);
 #endif
 
+#ifdef ETS_DBG_FORCE_TRAP
+extern erts_aint_t erts_ets_dbg_force_trap;
+#endif
+
 #endif /* ERL_DB_H__ */
 
 #if defined(ERTS_WANT_DB_INTERNAL__) && !defined(ERTS_HAVE_DB_INTERNAL__)
diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index cb5c496e90..8bbd0cd9a3 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -411,7 +411,7 @@ static void db_foreach_offheap_hash(DbTable *,
 				    void (*)(ErlOffHeap *, void *),
 				    void *);
 
-static int db_delete_all_objects_hash(Process* p, DbTable* tbl);
+static SWord db_delete_all_objects_hash(Process* p, DbTable* tbl, SWord reds);
 #ifdef HARDDEBUG
 static void db_check_table_hash(DbTableHash *tb);
 #endif
@@ -2255,7 +2255,7 @@ void db_initialize_hash(void)
 }
 
 
-int db_mark_all_deleted_hash(DbTable *tbl)
+static SWord db_mark_all_deleted_hash(DbTable *tbl, SWord reds)
 {
     DbTableHash *tb = &tbl->hash;
     HashDbTerm* list;
@@ -2270,10 +2270,11 @@ int db_mark_all_deleted_hash(DbTable *tbl)
 		list->hvalue = INVALID_HASH;
 		list = list->next;
 	    }while(list != NULL);
+            reds--;
 	}
     }
     erts_atomic_set_nob(&tb->common.nitems, 0);
-    return DB_ERROR_NONE;
+    return reds < 0 ? 0 : reds;   /* ToDo: Yield! */
 }
 
 
@@ -3073,16 +3074,20 @@ db_finalize_dbterm_hash(int cret, DbUpdateHandle* handle)
     return;
 }
 
-static int db_delete_all_objects_hash(Process* p, DbTable* tbl)
+static SWord db_delete_all_objects_hash(Process* p, DbTable* tbl, SWord reds)
 {
     if (IS_FIXED(tbl)) {
-	db_mark_all_deleted_hash(tbl);
+        /* ToDo: Yield! */
+	reds = db_mark_all_deleted_hash(tbl, reds);
     } else {
-	db_free_table_hash(tbl);
+        reds = db_free_table_continue_hash(tbl, reds);
+        if (reds < 0)
+            return reds;
+
 	db_create_hash(p, tbl);
 	erts_atomic_set_nob(&tbl->hash.common.nitems, 0);
     }
-    return 0;
+    return reds;
 }
 
 void db_foreach_offheap_hash(DbTable *tbl,
diff --git a/erts/emulator/beam/erl_db_hash.h b/erts/emulator/beam/erl_db_hash.h
index 7d27609825..5f88170ced 100644
--- a/erts/emulator/beam/erl_db_hash.h
+++ b/erts/emulator/beam/erl_db_hash.h
@@ -86,9 +86,6 @@ int db_get_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret);
 
 int db_erase_hash(DbTable *tbl, Eterm key, Eterm *ret);
 
-/* not yet in method table */
-int db_mark_all_deleted_hash(DbTable *tbl);
-
 typedef struct {
     float avg_chain_len;
     float std_dev_chain_len;
diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
index 5a276b9d88..5ef5451289 100644
--- a/erts/emulator/beam/erl_db_tree.c
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -436,7 +436,7 @@ static void db_foreach_offheap_tree(DbTable *,
 				    void (*)(ErlOffHeap *, void *),
 				    void *);
 
-static int db_delete_all_objects_tree(Process* p, DbTable* tbl);
+static SWord db_delete_all_objects_tree(Process* p, DbTable* tbl, SWord reds);
 
 #ifdef HARDDEBUG
 static void db_check_table_tree(DbTable *tbl);
@@ -2023,12 +2023,14 @@ static SWord db_free_table_continue_tree(DbTable *tbl, SWord reds)
     return reds;
 }
 
-static int db_delete_all_objects_tree(Process* p, DbTable* tbl)
+static SWord db_delete_all_objects_tree(Process* p, DbTable* tbl, SWord reds)
 {
-    db_free_table_tree(tbl);
+    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);
-    return 0;
+    return reds;
 }
 
 static void do_db_tree_foreach_offheap(TreeDbTerm *,
diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h
index 6b126f35d6..db8dfa5be4 100644
--- a/erts/emulator/beam/erl_db_util.h
+++ b/erts/emulator/beam/erl_db_util.h
@@ -32,6 +32,7 @@
 ** DMC_DEBUG does NOT need DEBUG, but DEBUG needs DMC_DEBUG
 */
 #define DMC_DEBUG 1
+#define ETS_DBG_FORCE_TRAP 1
 #endif
 
 /*
@@ -180,8 +181,7 @@ typedef struct db_table_method
             Eterm* ret);
     int (*db_take)(Process *, DbTable *, Eterm, Eterm *);
 
-    int (*db_delete_all_objects)(Process* p,
-				 DbTable* db /* [in out] */ );
+    SWord (*db_delete_all_objects)(Process* p, DbTable* db, SWord reds);
 
     int (*db_free_table)(DbTable* db /* [in out] */ );
     SWord (*db_free_table_continue)(DbTable* db, SWord reds);
@@ -267,6 +267,10 @@ typedef struct db_table_common {
     Uint32 status;            /* bit masks defined  below */
     int keypos;               /* defaults to 1 */
     int compress;
+
+#ifdef ETS_DBG_FORCE_TRAP
+    erts_atomic_t dbg_force_trap;  /* &1 force enabled, &2 trap this call */
+#endif
 } DbTableCommon;
 
 /* These are status bit patterns */
@@ -281,9 +285,7 @@ typedef struct db_table_common {
 #define DB_FINE_LOCKED   (1 << 8) /* write_concurrency */
 #define DB_FREQ_READ     (1 << 9) /* read_concurrency */
 #define DB_NAMED_TABLE   (1 << 10)
-
-#define ERTS_ETS_TABLE_TYPES (DB_BAG|DB_SET|DB_DUPLICATE_BAG|DB_ORDERED_SET\
-                              |DB_FINE_LOCKED|DB_FREQ_READ|DB_NAMED_TABLE)
+#define DB_BUSY          (1 << 11)
 
 #define IS_HASH_TABLE(Status) (!!((Status) & \
 				  (DB_BAG | DB_SET | DB_DUPLICATE_BAG)))
diff --git a/lib/stdlib/src/ets.erl b/lib/stdlib/src/ets.erl
index 6a559f0be5..a35f79c0d9 100644
--- a/lib/stdlib/src/ets.erl
+++ b/lib/stdlib/src/ets.erl
@@ -77,7 +77,9 @@
          whereis/1]).
 
 %% internal exports
--export([internal_request_all/0]).
+-export([internal_request_all/0,
+         internal_delete_all/2,
+         internal_select_delete/2]).
 
 -spec all() -> [Tab] when
       Tab :: tab().
@@ -116,7 +118,15 @@ delete(_, _) ->
 -spec delete_all_objects(Tab) -> true when
       Tab :: tab().
 
-delete_all_objects(_) ->
+delete_all_objects(Tab) ->
+    _ = ets:internal_delete_all(Tab, undefined),
+    true.
+
+-spec internal_delete_all(Tab, undefined) -> NumDeleted when
+      Tab :: tab(),
+      NumDeleted :: non_neg_integer().
+
+internal_delete_all(_, _) ->
     erlang:nif_error(undef).
 
 -spec delete_object(Tab, Object) -> true when
@@ -378,7 +388,17 @@ select_count(_, _) ->
       MatchSpec :: match_spec(),
       NumDeleted :: non_neg_integer().
 
-select_delete(_, _) ->
+select_delete(Tab, [{'_',[],[true]}]) ->
+    ets:internal_delete_all(Tab, undefined);
+select_delete(Tab, MatchSpec) ->
+    ets:internal_select_delete(Tab, MatchSpec).
+
+-spec internal_select_delete(Tab, MatchSpec) -> NumDeleted when
+      Tab :: tab(),
+      MatchSpec :: match_spec(),
+      NumDeleted :: non_neg_integer().
+
+internal_select_delete(_, _) ->
     erlang:nif_error(undef).
 
 -spec select_replace(Tab, MatchSpec) -> NumReplaced when
diff --git a/lib/stdlib/test/ets_SUITE.erl b/lib/stdlib/test/ets_SUITE.erl
index 02211fa8df..574aac96c8 100644
--- a/lib/stdlib/test/ets_SUITE.erl
+++ b/lib/stdlib/test/ets_SUITE.erl
@@ -87,6 +87,7 @@
 
 -export([t_select_reverse/1]).
 
+-include_lib("stdlib/include/ms_transform.hrl"). % ets:fun2ms
 -include_lib("common_test/include/ct.hrl").
 
 -define(m(A,B), assert_eq(A,B)).
@@ -173,10 +174,12 @@ groups() ->
 
 init_per_suite(Config) ->
     erts_debug:set_internal_state(available_internal_state, true),
+    erts_debug:set_internal_state(ets_force_trap, true),
     Config.
 
 end_per_suite(_Config) ->
     stop_spawn_logger(),
+    erts_debug:set_internal_state(ets_force_trap, false),
     catch erts_debug:set_internal_state(available_internal_state, false),
     ok.
 
@@ -812,7 +815,60 @@ t_delete_all_objects_do(Opts) ->
     4000 = ets:info(T,size),
     true = ets:delete_all_objects(T),
     0 = ets:info(T,size),
-    ets:delete(T).
+    ets:delete(T),
+
+    %% Test delete_all_objects is atomic
+    T2 = ets:new(t_delete_all_objects, [public | Opts]),
+    Self = self(),
+    Inserters = [spawn_link(fun() -> inserter(T2, 100*1000, 1, Self) end) || _ <- [1,2,3,4]],
+    [receive {Ipid, running} -> ok end || Ipid <- Inserters],
+    
+    ets:delete_all_objects(T2),
+    erlang:yield(),
+    [Ipid ! stop || Ipid <- Inserters],
+    Result = [receive {Ipid, stopped, Highest} -> {Ipid,Highest} end || Ipid <- Inserters],
+    
+    %% Verify unbroken sequences of objects inserted _after_ ets:delete_all_objects.
+    Sum = lists:foldl(fun({Ipid, Highest}, AccSum) ->
+                              %% ets:fun2ms(fun({{K,Ipid}}) when K =< Highest -> true end),
+                              AliveMS = [{{{'$1',Ipid}},[{'=<','$1',{const,Highest}}],[true]}],
+                              Alive = ets:select_count(T2, AliveMS),
+                              Lowest = Highest - (Alive-1),
+
+                              %% ets:fun2ms(fun({{K,Ipid}}) when K < Lowest -> true end)
+                              DeletedMS = [{{{'$1',Ipid}},[{'<','$1',{const,Lowest}}],[true]}],
+                              0 = ets:select_count(T2, DeletedMS),
+                              AccSum + Alive
+                      end,
+                      0,
+                      Result),
+    ok = case ets:info(T2, size) of
+             Sum -> ok;
+             Size ->
+                 io:format("Sum = ~p\nSize = ~p\n", [Sum, Size]),
+                 {Sum,Size}
+         end,
+
+    ets:delete(T2).
+
+inserter(_, 0, _, _) ->
+    ok;
+inserter(T, N, Next, Papa) ->
+    case Next of
+        10*1000 ->
+            Papa ! {self(), running};
+        _ ->
+            ok
+    end,
+                
+    ets:insert(T, {{Next, self()}}),
+    receive
+        stop ->
+            Papa ! {self(), stopped, Next},
+            ok
+    after 0 ->
+            inserter(T, N-1, Next+1, Papa)
+    end.
 
 
 %% Test ets:delete_object/2.