Merge PR-1076 from g-andrade/feature/ets_conditional_insert OTP-14319

ETS: Allow for conditional insertions

7 files changed, 1767 insertions, 762 deletions

diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 6f50297fc5..4140938210 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -361,6 +361,7 @@ 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
 
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 2f188b5391..378328856d 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -362,6 +362,7 @@ 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);
 static BIF_RETTYPE ets_select_count_1(BIF_ALIST_1);
+static BIF_RETTYPE ets_select_replace_1(BIF_ALIST_1);
 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);
@@ -376,6 +377,7 @@ static BIF_RETTYPE ets_select3(Process* p, Eterm arg1, Eterm arg2, Eterm arg3);
  */
 Export ets_select_delete_continue_exp;
 Export ets_select_count_continue_exp;
+Export ets_select_replace_continue_exp;
 Export ets_select_continue_exp;
 
 /*
@@ -2922,6 +2924,103 @@ BIF_RETTYPE ets_select_count_2(BIF_ALIST_2)
     return result;
 }
 
+/*
+ ** This is for trapping, cannot be called directly.
+ */
+static BIF_RETTYPE ets_select_replace_1(BIF_ALIST_1)
+{
+    Process *p = BIF_P;
+    Eterm a1 = BIF_ARG_1;
+    BIF_RETTYPE result;
+    DbTable* tb;
+    int cret;
+    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);
+    }
+
+    cret = tb->common.meth->db_select_replace_continue(p,tb,a1,&ret);
+
+    if(!DID_TRAP(p,ret) && ITERATION_SAFETY(p,tb) != ITER_SAFE) {
+        unfix_table_locked(p, tb, &kind);
+    }
+
+    db_unlock(tb, kind);
+
+    switch (cret) {
+    case DB_ERROR_NONE:
+        ERTS_BIF_PREP_RET(result, ret);
+        break;
+    default:
+        ERTS_BIF_PREP_ERROR(result, p, BADARG);
+        break;
+    }
+    erts_match_set_release_result(p);
+
+    return result;
+}
+
+
+BIF_RETTYPE ets_select_replace_2(BIF_ALIST_2)
+{
+    BIF_RETTYPE result;
+    DbTable* tb;
+    int cret;
+    Eterm ret;
+    enum DbIterSafety safety;
+
+    CHECK_TABLES();
+
+    if ((tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC)) == NULL) {
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
+    if (tb->common.status & DB_BAG) {
+        /* Bag implementation presented both semantic consistency
+           and performance issues */
+        db_unlock(tb, LCK_WRITE_REC);
+        BIF_ERROR(BIF_P, BADARG);
+    }
+
+    safety = ITERATION_SAFETY(BIF_P,tb);
+    if (safety == ITER_UNSAFE) {
+        local_fix_table(tb);
+    }
+    cret = tb->common.meth->db_select_replace(BIF_P, tb, BIF_ARG_1, BIF_ARG_2, &ret);
+
+    if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
+        fix_table_locked(BIF_P,tb);
+    }
+    if (safety == ITER_UNSAFE) {
+        local_unfix_table(tb);
+    }
+    db_unlock(tb, LCK_WRITE_REC);
+
+    switch (cret) {
+    case DB_ERROR_NONE:
+        ERTS_BIF_PREP_RET(result, ret);
+        break;
+    case DB_ERROR_SYSRES:
+        ERTS_BIF_PREP_ERROR(result, BIF_P, SYSTEM_LIMIT);
+        break;
+    default:
+        ERTS_BIF_PREP_ERROR(result, BIF_P, BADARG);
+        break;
+    }
+
+    erts_match_set_release_result(BIF_P);
+
+    return result;
+}
+
 
 BIF_RETTYPE ets_select_reverse_3(BIF_ALIST_3)
 {
@@ -3335,6 +3434,11 @@ void init_db(ErtsDbSpinCount db_spin_count)
 			  &ets_select_count_1);
 
     /* Non visual BIF to trap to. */
+    erts_init_trap_export(&ets_select_replace_continue_exp,
+                          am_ets, am_atom_put("replace_trap",11), 1,
+                          &ets_select_replace_1);
+
+    /* Non visual BIF to trap to. */
     erts_init_trap_export(&ets_select_continue_exp,
 			  am_ets, am_atom_put("select_trap",11), 1,
 			  &ets_select_trap_1);
diff --git a/erts/emulator/beam/erl_db.h b/erts/emulator/beam/erl_db.h
index cbf4b9e007..4ff9f224e8 100644
--- a/erts/emulator/beam/erl_db.h
+++ b/erts/emulator/beam/erl_db.h
@@ -122,6 +122,7 @@ extern int erts_ets_realloc_always_moves;  /* set in erl_init */
 extern int erts_ets_always_compress;  /* set in erl_init */
 extern Export ets_select_delete_continue_exp;
 extern Export ets_select_count_continue_exp;
+extern Export ets_select_replace_continue_exp;
 extern Export ets_select_continue_exp;
 extern erts_smp_atomic_t erts_ets_misc_mem_size;
 
diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index 18405342da..7ab27df00c 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -288,6 +288,9 @@ static ERTS_INLINE Sint next_slot_w(DbTableHash* tb, Uint ix,
 #endif
 }
 
+#ifndef MIN
+#define MIN(X, Y) ((X) < (Y) ? (X) : (Y))
+#endif
 
 /*
  * Some special binary flags
@@ -352,7 +355,8 @@ static ERTS_INLINE void SET_SEGTAB(DbTableHash* tb,
 	erts_smp_atomic_set_nob(&tb->segtab, (erts_aint_t) segtab);
 }
 
-
+/* Used by select_replace on analyze_pattern */
+typedef int (*extra_match_validator_t)(int keypos, Eterm match, Eterm guard, Eterm body);
 
 /*
 ** Forward decl's (static functions)
@@ -368,8 +372,9 @@ static void shrink(DbTableHash* tb, int nitems);
 static void grow(DbTableHash* tb, int nitems);
 static Eterm build_term_list(Process* p, HashDbTerm* ptr1, HashDbTerm* ptr2,
 			   Uint sz, DbTableHash*);
-static int analyze_pattern(DbTableHash *tb, Eterm pattern, 
-			   struct mp_info *mpi);
+static int analyze_pattern(DbTableHash *tb, Eterm pattern,
+                           extra_match_validator_t extra_validator, /* Optional callback */
+                           struct mp_info *mpi);
 
 /*
  *  Method interface functions
@@ -398,19 +403,24 @@ static int db_select_chunk_hash(Process *p, DbTable *tbl, Eterm tid,
 				int reverse, Eterm *ret);
 static int db_select_hash(Process *p, DbTable *tbl, Eterm tid,
 			  Eterm pattern, int reverse, Eterm *ret);
-static int db_select_count_hash(Process *p, DbTable *tbl,  Eterm tid,
-				Eterm pattern, Eterm *ret);
-static int db_select_delete_hash(Process *p, DbTable *tbl, Eterm tid,
-				 Eterm pattern, Eterm *ret);
-
-static int db_select_continue_hash(Process *p, DbTable *tbl, 
+static int db_select_continue_hash(Process *p, DbTable *tbl,
 				   Eterm continuation, Eterm *ret);
 
+static int db_select_count_hash(Process *p, DbTable *tbl, Eterm tid,
+				Eterm pattern, Eterm *ret);
 static int db_select_count_continue_hash(Process *p, DbTable *tbl,
 					 Eterm continuation, Eterm *ret);
 
+static int db_select_delete_hash(Process *p, DbTable *tbl, Eterm tid,
+				 Eterm pattern, Eterm *ret);
 static int db_select_delete_continue_hash(Process *p, DbTable *tbl,
 					  Eterm continuation, Eterm *ret);
+
+static int db_select_replace_hash(Process *p, DbTable *tbl, Eterm tid,
+                                  Eterm pattern, Eterm *ret);
+static int db_select_replace_continue_hash(Process *p, DbTable *tbl,
+                                           Eterm continuation, Eterm *ret);
+
 static int db_take_hash(Process *, DbTable *, Eterm, Eterm *);
 static void db_print_hash(fmtfn_t to,
 			  void *to_arg,
@@ -523,6 +533,8 @@ DbTableMethod db_hash =
     db_select_delete_continue_hash,
     db_select_count_hash,
     db_select_count_continue_hash,
+    db_select_replace_hash,
+    db_select_replace_continue_hash,
     db_take_hash,
     db_delete_all_objects_hash,
     db_free_table_hash,
@@ -1139,816 +1151,1104 @@ static BIF_RETTYPE bif_trap1(Export *bif,
 {
     BIF_TRAP1(bif, p, p1);
 }
-    
+
+
 /*
- * Continue collecting select matches, this may happen either due to a trap
- * or when the user calls ets:select/1
+ * Match traversal callbacks
  */
-static int db_select_continue_hash(Process *p, 
-				   DbTable *tbl, 
-				   Eterm continuation, 
-				   Eterm *ret)
-{
-    DbTableHash *tb = &tbl->hash;
-    Sint slot_ix; 
-    Sint save_slot_ix;
-    Sint chunk_size;
-    int all_objects;
-    Binary *mp;
-    int num_left = 1000;
-    HashDbTerm *current = 0;
-    Eterm match_list;
-    Eterm *hp;
-    Eterm match_res;
-    Sint got;
-    Eterm *tptr;
-    erts_smp_rwmtx_t* lck;
 
-#define RET_TO_BIF(Term, State) do { *ret = (Term); return State; } while(0);
+/* Called when no match is possible.
+ *      context_ptr: Pointer to context
+ *      ret: Pointer to traversal function term return.
+ *
+ * Both the direct return value and 'ret' are used as the traversal function return values.
+ */
+typedef int (*mtraversal_on_nothing_can_match_t)(void* context_ptr, Eterm* ret);
+
+/* Called for each match result.
+ *      context_ptr: Pointer to context
+ *      slot_ix: Current slot index
+ *      current_ptr_ptr: Triple pointer to either the bucket or the 'next' pointer in the previous element;
+ *                       can be (carefully) used to adjust iteration when deleting or replacing elements.
+ *      match_res: The result of running the match program against the current term.
+ *
+ * Should return 1 for successful match, 0 otherwise.
+ */
+typedef int (*mtraversal_on_match_res_t)(void* context_ptr, Sint slot_ix, HashDbTerm*** current_ptr_ptr,
+                                         Eterm match_res);
+
+/* Called when either we've matched enough elements in this cycle or EOT was reached.
+ *      context_ptr: Pointer to context
+ *      slot_ix: Current slot index
+ *      got: How many elements have been matched so far
+ *      iterations_left: Number of intended iterations (down from an initial max.) left in this traversal cycle
+ *      mpp: Double pointer to the compiled match program
+ *      ret: Pointer to traversal function term return.
+ *
+ * Both the direct return value and 'ret' are used as the traversal function return values.
+ * If *mpp is set to NULL, it won't be deallocated (useful for trapping.)
+ */
+typedef int (*mtraversal_on_loop_ended_t)(void* context_ptr, Sint slot_ix, Sint got,
+                                          Sint iterations_left, Binary** mpp, Eterm* ret);
+
+/* Called when it's time to trap
+ *      context_ptr: Pointer to context
+ *      slot_ix: Current slot index
+ *      got: How many elements have been matched so far
+ *      mpp: Double pointer to the compiled match program
+ *      ret: Pointer to traversal function term return.
+ *
+ * Both the direct return value and 'ret' are used as the traversal function return values.
+ * If *mpp is set to NULL, it won't be deallocated (useful for trapping.)
+ */
+typedef int (*mtraversal_on_trap_t)(void* context_ptr, Sint slot_ix, Sint got, Binary** mpp, Eterm* ret);
 
-    /* Decode continuation. We know it's a tuple but not the arity or anything else */
+/*
+ * Begin hash table match traversal
+ */
+static int match_traverse(Process* p, DbTableHash* tb,
+                          Eterm pattern,
+                          extra_match_validator_t extra_match_validator, /* Optional */
+                          Sint chunk_size,      /* If 0, no chunking */
+                          Sint iterations_left, /* Nr. of iterations left */
+                          Eterm** hpp,          /* Heap */
+                          int lock_for_write,   /* Set to 1 if we're going to delete or
+                                                   modify existing terms */
+                          mtraversal_on_nothing_can_match_t on_nothing_can_match,
+                          mtraversal_on_match_res_t on_match_res,
+                          mtraversal_on_loop_ended_t on_loop_ended,
+                          mtraversal_on_trap_t on_trap,
+                          void* context_ptr, /* State for callbacks above */
+                          Eterm* ret)
+{
+    Sint slot_ix;                  /* Slot index */
+    HashDbTerm** current_ptr;      /* Refers to either the bucket pointer or
+                                    * the 'next' pointer in the previous term
+                                    */
+    HashDbTerm* saved_current;     /* Helper to avoid double skip on match */
+    struct mp_info mpi;
+    unsigned current_list_pos = 0; /* Prefound buckets list index */
+    Eterm match_res;
+    Sint got = 0;                  /* Matched terms counter */
+    erts_smp_rwmtx_t* lck;         /* Slot lock */
+    int ret_value;
+#ifdef ERTS_SMP
+    erts_smp_rwmtx_t* (*lock_hash_function)(DbTableHash*, HashValue)
+        = (lock_for_write ? WLOCK_HASH : RLOCK_HASH);
+    void (*unlock_hash_function)(erts_smp_rwmtx_t*)
+        = (lock_for_write ? WUNLOCK_HASH : RUNLOCK_HASH);
+#else
+    #define lock_hash_function(tb, hval) NULL
+    #define unlock_hash_function(lck) ((void)lck)
+#endif
+    Sint (*next_slot_function)(DbTableHash*, Uint, erts_smp_rwmtx_t**)
+        = (lock_for_write ? next_slot_w : next_slot);
 
-    tptr = tuple_val(continuation);
+    if ((ret_value = analyze_pattern(tb, pattern, extra_match_validator, &mpi))
+            != DB_ERROR_NONE)
+    {
+        *ret = NIL;
+        goto done;
+    }
 
-    if (arityval(*tptr) != 6)
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
-    
-    if (!is_small(tptr[2]) || !is_small(tptr[3]) || 
-	!(is_list(tptr[5]) || tptr[5] == NIL) || !is_small(tptr[6]))
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
-    if ((chunk_size = signed_val(tptr[3])) < 0)
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
-    mp = erts_db_get_match_prog_binary(tptr[4]);
-    if (!mp)
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
-    all_objects = mp->flags & BIN_FLAG_ALL_OBJECTS;
-    match_list = tptr[5];
-    if ((got = signed_val(tptr[6])) < 0)
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    if (!mpi.something_can_match) {
+        /* Can't possibly match anything */
+        ret_value = on_nothing_can_match(context_ptr, ret);
+        goto done;
+    }
 
-    slot_ix = signed_val(tptr[2]);
-    if (slot_ix < 0 /* EOT */ 
-	|| (chunk_size && got >= chunk_size)) {       
-	goto done; /* Already got all or enough in the match_list */
+    if (mpi.all_objects) {
+        mpi.mp->flags |= BIN_FLAG_ALL_OBJECTS;
     }
 
-    lck = RLOCK_HASH(tb,slot_ix);
-    if (slot_ix >= NACTIVE(tb)) {
-	RUNLOCK_HASH(lck);	
-	RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    /*
+     * Look for initial slot / bucket
+     */
+    if (!mpi.key_given) {
+        /* Run this code if pattern is variable or GETKEY(pattern)  */
+        /* is a variable                                            */
+        slot_ix = 0;
+        lck = lock_hash_function(tb,slot_ix);
+        for (;;) {
+            ASSERT(slot_ix < NACTIVE(tb));
+            if (*(current_ptr = &BUCKET(tb,slot_ix)) != NULL) {
+                break;
+            }
+            slot_ix = next_slot_function(tb,slot_ix,&lck);
+            if (slot_ix == 0) {
+                ret_value = on_loop_ended(context_ptr, slot_ix, got, iterations_left, &mpi.mp, ret);
+                goto done;
+            }
+        }
+    } else {
+        /* We have at least one */
+        slot_ix = mpi.lists[current_list_pos].ix;
+        lck = lock_hash_function(tb, slot_ix);
+        current_ptr = mpi.lists[current_list_pos].bucket;
+        ASSERT(*current_ptr == BUCKET(tb,slot_ix));
+        ++current_list_pos;
     }
 
-    while ((current = BUCKET(tb,slot_ix)) == NULL) {
-	slot_ix = next_slot(tb, slot_ix, &lck);
-	if (slot_ix == 0) {
-	    slot_ix = -1; /* EOT */
-	    goto done;	   
-	}
-    }	  
+    /*
+     * Execute traversal cycle
+     */
     for(;;) {
-	if (current->hvalue != INVALID_HASH && 
-	    (match_res = db_match_dbterm(&tb->common, p, mp, all_objects,
-					 &current->dbterm, &hp, 2),
-	     is_value(match_res))) {
+        if (*current_ptr != NULL) {
+            if ((*current_ptr)->hvalue != INVALID_HASH) {
+                match_res = db_match_dbterm(&tb->common, p, mpi.mp, 0,
+                                            &(*current_ptr)->dbterm, hpp, 2);
+                saved_current = *current_ptr;
+                if (on_match_res(context_ptr, slot_ix, &current_ptr, match_res)) {
+                    ++got;
+                }
+                --iterations_left;
+                if (*current_ptr != saved_current) {
+                    /* Don't advance to next, the callback did it already */
+                    continue;
+                }
+            }
+            current_ptr = &((*current_ptr)->next);
+        }
+        else if (mpi.key_given) {  /* Key is bound */
+            unlock_hash_function(lck);
+            if (current_list_pos == mpi.num_lists) {
+                ret_value = on_loop_ended(context_ptr, -1, got, iterations_left, &mpi.mp, ret);
+                goto done;
+            } else {
+                slot_ix = mpi.lists[current_list_pos].ix;
+                lck = lock_hash_function(tb, slot_ix);
+                current_ptr = mpi.lists[current_list_pos].bucket;
+                ASSERT(mpi.lists[current_list_pos].bucket == &BUCKET(tb,slot_ix));
+                ++current_list_pos;
+            }
+        }
+        else { /* Key is variable */
+            if ((slot_ix = next_slot_function(tb,slot_ix,&lck)) == 0) {
+                slot_ix = -1;
+                break;
+            }
+            if (chunk_size && got >= chunk_size) {
+                unlock_hash_function(lck);
+                break;
+            }
+            if (iterations_left <= 0 || MBUF(p)) {
+                /*
+                 * We have either reached our limit, or just created some heap fragments.
+                 * Since many heap fragments will make the GC slower, trap and GC now.
+                 */
+                unlock_hash_function(lck);
+                ret_value = on_trap(context_ptr, slot_ix, got, &mpi.mp, ret);
+                goto done;
+            }
+            current_ptr = &BUCKET(tb,slot_ix);
+        }
+    }
 
-	    match_list = CONS(hp, match_res, match_list);
-	    ++got;
-	}
+    ret_value = on_loop_ended(context_ptr, slot_ix, got, iterations_left, &mpi.mp, ret);
 
-	--num_left;
-	save_slot_ix = slot_ix;
-	if ((current = next(tb, (Uint*)&slot_ix, &lck, current)) == NULL) {
-	    slot_ix = -1; /* EOT */
-	    break;
-	}
-	if (slot_ix != save_slot_ix) { 
-	    if (chunk_size && got >= chunk_size) {
-		RUNLOCK_HASH(lck);
-		break;
-	    }    
-	    if (num_left <= 0 || MBUF(p)) {
-		/*
-		 * We have either reached our limit, or just created some heap fragments.
-		 * Since many heap fragments will make the GC slower, trap and GC now.
-		 */
-		RUNLOCK_HASH(lck);
-		goto trap;
-	    }
-	}
-    }
 done:
-    BUMP_REDS(p, 1000 - num_left);
-    if (chunk_size) {
-	Eterm continuation;
-	Eterm rest = NIL;
-	Sint rest_size = 0;
-
-	if (got > chunk_size) { /* Cannot write destructively here, 
-				   the list may have 
-				   been in user space */
-	    rest = NIL;
-	    hp = HAlloc(p, (got - chunk_size) * 2); 
-	    while (got-- > chunk_size) {
-		rest = CONS(hp, CAR(list_val(match_list)), rest);
-		hp += 2;
-		match_list = CDR(list_val(match_list));
-		++rest_size;
-	    }
-	}
-	if (rest != NIL || slot_ix >= 0) {
-	    hp = HAlloc(p,3+7);
-	    continuation = TUPLE6(hp, tptr[1], make_small(slot_ix), 
-				  tptr[3], tptr[4], rest, 
-				  make_small(rest_size));
-	    hp += 7;
-	    RET_TO_BIF(TUPLE2(hp, match_list, continuation),DB_ERROR_NONE);
-	} else {
-	    if (match_list != NIL) {
-		hp = HAlloc(p, 3);
-		RET_TO_BIF(TUPLE2(hp, match_list, am_EOT),DB_ERROR_NONE);
-	    } else {
-		RET_TO_BIF(am_EOT, DB_ERROR_NONE);
-	    }
-	}
+    /* We should only jump directly to this label if
+     * we've already called on_nothing_can_match / on_loop_ended / on_trap
+     */
+    if (mpi.mp != NULL) {
+        erts_bin_free(mpi.mp);
     }
-    RET_TO_BIF(match_list,DB_ERROR_NONE);
-
-trap:
-    BUMP_ALL_REDS(p);
-
-    hp = HAlloc(p,7);
-    continuation = TUPLE6(hp, tptr[1], make_small(slot_ix), tptr[3],
-			  tptr[4], match_list, make_small(got));
-    RET_TO_BIF(bif_trap1(&ets_select_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
-
-#undef RET_TO_BIF
-
-}
+    if (mpi.lists != mpi.dlists) {
+        erts_free(ERTS_ALC_T_DB_SEL_LIST,
+                (void *) mpi.lists);
+    }
+    return ret_value;
 
-static int db_select_hash(Process *p, DbTable *tbl, Eterm tid,
-			  Eterm pattern, int reverse,
-			  Eterm *ret)
-{
-    return db_select_chunk_hash(p, tbl, tid, pattern, 0, reverse, ret);
+#ifndef SMP
+#undef lock_hash_function
+#undef unlock_hash_function
+#endif
 }
 
-static int db_select_chunk_hash(Process *p, DbTable *tbl, Eterm tid,
-				Eterm pattern, Sint chunk_size, 
-				int reverse, /* not used */
-				Eterm *ret)
+/*
+ * Continue hash table match traversal
+ */
+static int match_traverse_continue(Process* p, DbTableHash* tb,
+                                   Sint chunk_size,      /* If 0, no chunking */
+                                   Sint iterations_left, /* Nr. of iterations left */
+                                   Eterm** hpp,          /* Heap */
+                                   Sint slot_ix,         /* Slot index to resume traversal from */
+                                   Sint got,             /* Matched terms counter */
+                                   Binary** mpp,         /* Existing match program */
+                                   int lock_for_write,   /* Set to 1 if we're going to delete or
+                                                            modify existing terms */
+                                   mtraversal_on_match_res_t on_match_res,
+                                   mtraversal_on_loop_ended_t on_loop_ended,
+                                   mtraversal_on_trap_t on_trap,
+                                   void* context_ptr, /* For callbacks */
+                                   Eterm* ret)
 {
-    DbTableHash *tb = &tbl->hash;
-    struct mp_info mpi;
-    Sint slot_ix;
-    HashDbTerm *current = 0;
-    unsigned current_list_pos = 0;
-    Eterm match_list;
+    int all_objects = (*mpp)->flags & BIN_FLAG_ALL_OBJECTS;
+    HashDbTerm** current_ptr;  /* Refers to either the bucket pointer or
+                                       * the 'next' pointer in the previous term
+                                       */
+    HashDbTerm* saved_current; /* Helper to avoid double skip on match */
     Eterm match_res;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got = 0;
-    Eterm continuation;
-    int errcode;
-    Eterm mpb;
     erts_smp_rwmtx_t* lck;
+    int ret_value;
+#ifdef ERTS_SMP
+    erts_smp_rwmtx_t* (*lock_hash_function)(DbTableHash*, HashValue)
+        = (lock_for_write ? WLOCK_HASH : RLOCK_HASH);
+    void (*unlock_hash_function)(erts_smp_rwmtx_t*)
+        = (lock_for_write ? WUNLOCK_HASH : RUNLOCK_HASH);
+#else
+    #define lock_hash_function(tb, hval) NULL
+    #define unlock_hash_function(lck) ((void)lck)
+#endif
+    Sint (*next_slot_function)(DbTableHash* tb, Uint ix, erts_smp_rwmtx_t** lck_ptr)
+        = (lock_for_write ? next_slot_w : next_slot);
 
-
-#define RET_TO_BIF(Term,RetVal) do {		\
-	if (mpi.mp != NULL) {			\
-	    erts_bin_free(mpi.mp);		\
-	}					\
-	if (mpi.lists != mpi.dlists) {		\
-	    erts_free(ERTS_ALC_T_DB_SEL_LIST,	\
-		      (void *) mpi.lists);	\
-	}					\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
-
-
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
-	RET_TO_BIF(NIL,errcode);
+    if (got < 0) {
+        *ret = NIL;
+        return DB_ERROR_BADPARAM;
     }
 
-    if (!mpi.something_can_match) {
-	if (chunk_size) {
-	    RET_TO_BIF(am_EOT, DB_ERROR_NONE); /* We're done */
-	}  
-	RET_TO_BIF(NIL, DB_ERROR_NONE);
-	/* can't possibly match anything */
+    if (slot_ix < 0 /* EOT */
+       || (chunk_size && got >= chunk_size))
+    {
+        /* Already got all or enough in the match_list */
+        ret_value = on_loop_ended(context_ptr, slot_ix, got, iterations_left, mpp, ret);
+        goto done;
     }
 
-    if (!mpi.key_given) {
-    /* Run this code if pattern is variable or GETKEY(pattern)  */
-    /* is a variable                                            */
-	slot_ix = 0;
-	lck = RLOCK_HASH(tb,slot_ix);
-	for (;;) { 
-	    ASSERT(slot_ix < NACTIVE(tb));
-	    if ((current = BUCKET(tb,slot_ix)) != NULL) {
-		break;
-	    }
-	    slot_ix = next_slot(tb,slot_ix,&lck);
-	    if (slot_ix == 0) {
-		if (chunk_size) {
-		    RET_TO_BIF(am_EOT, DB_ERROR_NONE); /* We're done */
-		}  
-		RET_TO_BIF(NIL,DB_ERROR_NONE); 
-	    }
-	}
-    } else {
-	/* We have at least one */
-	slot_ix = mpi.lists[current_list_pos].ix;
-	lck = RLOCK_HASH(tb, slot_ix);
-	current = *(mpi.lists[current_list_pos].bucket);
-	ASSERT(current == BUCKET(tb,slot_ix));
-	++current_list_pos;
+    lck = lock_hash_function(tb, slot_ix);
+    if (slot_ix >= NACTIVE(tb)) { /* Is this possible? */
+        unlock_hash_function(lck);
+        *ret = NIL;
+        ret_value = DB_ERROR_BADPARAM;
+        goto done;
     }
 
-    match_list = NIL;
-
+    /*
+     * Resume traversal cycle from where we left
+     */
+    current_ptr = &BUCKET(tb,slot_ix);
     for(;;) {
-	if (current != NULL) {
-	    if (current->hvalue != INVALID_HASH) {
-		match_res = db_match_dbterm(&tb->common, p, mpi.mp, 0,
-					    &current->dbterm, &hp, 2);
-		if (is_value(match_res)) {
-		    match_list = CONS(hp, match_res, match_list);
-		    ++got;
-		}
-                --num_left;
-	    }
-	    current = current->next;
-	}	
-	else if (mpi.key_given) {  /* Key is bound */
-	    RUNLOCK_HASH(lck);
-	    if (current_list_pos == mpi.num_lists) {
-		slot_ix = -1; /* EOT */
-		goto done;
-	    } else {
-		slot_ix = mpi.lists[current_list_pos].ix;
-		lck = RLOCK_HASH(tb, slot_ix);
-		current = *(mpi.lists[current_list_pos].bucket);
-		ASSERT(mpi.lists[current_list_pos].bucket == &BUCKET(tb,slot_ix));
-		++current_list_pos;		       
-	    }
-	}
-	else { /* Key is variable */
-
-	    if ((slot_ix=next_slot(tb,slot_ix,&lck)) == 0) {
-		slot_ix = -1;
-		break;
-	    }
-	    if (chunk_size && got >= chunk_size) {
-		RUNLOCK_HASH(lck);
-		break;
-	    }    
-	    if (num_left <= 0 || MBUF(p)) {
-		/*
-		 * We have either reached our limit, or just created some heap fragments.
-		 * Since many heap fragments will make the GC slower, trap and GC now.
-		 */
-		RUNLOCK_HASH(lck);
-		goto trap;
-	    }
-	    current = BUCKET(tb,slot_ix);
+        if (*current_ptr != NULL) {
+            if ((*current_ptr)->hvalue != INVALID_HASH) {
+                match_res = db_match_dbterm(&tb->common, p, *mpp, all_objects,
+                                            &(*current_ptr)->dbterm, hpp, 2);
+                saved_current = *current_ptr;
+                if (on_match_res(context_ptr, slot_ix, &current_ptr, match_res)) {
+                    ++got;
+                }
+                --iterations_left;
+                if (*current_ptr != saved_current) {
+                    /* Don't advance to next, the callback did it already */
+                    continue;
+                }
+            }
+            current_ptr = &((*current_ptr)->next);
+        }
+        else {
+            if ((slot_ix=next_slot_function(tb,slot_ix,&lck)) == 0) {
+                slot_ix = -1;
+                break;
+            }
+            if (chunk_size && got >= chunk_size) {
+                unlock_hash_function(lck);
+                break;
+            }
+            if (iterations_left <= 0 || MBUF(p)) {
+                /*
+                 * We have either reached our limit, or just created some heap fragments.
+                 * Since many heap fragments will make the GC slower, trap and GC now.
+                 */
+                unlock_hash_function(lck);
+                ret_value = on_trap(context_ptr, slot_ix, got, mpp, ret);
+                goto done;
+            }
+            current_ptr = &BUCKET(tb,slot_ix);
         }
     }
-done:
-    BUMP_REDS(p, 1000 - num_left);
-    if (chunk_size) {
-	Eterm continuation;
-	Eterm rest = NIL;
-	Sint rest_size = 0;
-
-	if (mpi.all_objects)
-	    (mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
-	if (got > chunk_size) { /* Split list in return value and 'rest' */
-	    Eterm tmp = match_list;
-	    rest = match_list;
-	    while (got-- > chunk_size + 1) { 
-		tmp = CDR(list_val(tmp));
-		++rest_size;
-	    }
-	    ++rest_size;
-	    match_list = CDR(list_val(tmp));
-	    CDR(list_val(tmp)) = NIL; /* Destructive, the list has never 
-					 been in 'user space' */ 
-	}
-	if (rest != NIL || slot_ix >= 0) { /* Need more calls */
-	    hp = HAlloc(p,3+7+ERTS_MAGIC_REF_THING_SIZE);
-	    mpb = erts_db_make_match_prog_ref(p,(mpi.mp),&hp);
-	    if (mpi.all_objects)
-		(mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
-	    continuation = TUPLE6(hp, tid, make_small(slot_ix),
-				  make_small(chunk_size),  
-				  mpb, rest, 
-				  make_small(rest_size));
-	    mpi.mp = NULL; /*otherwise the return macro will destroy it */
-	    hp += 7;
-	    RET_TO_BIF(TUPLE2(hp, match_list, continuation),DB_ERROR_NONE);
-	} else { /* All data is exhausted */
-	    if (match_list != NIL) { /* No more data to search but still a
-					result to return to the caller */
-		hp = HAlloc(p, 3);
-		RET_TO_BIF(TUPLE2(hp, match_list, am_EOT),DB_ERROR_NONE);
-	    } else { /* Reached the end of the ttable with no data to return */
-		RET_TO_BIF(am_EOT, DB_ERROR_NONE);
-	    }
-	}
-    }
-    RET_TO_BIF(match_list,DB_ERROR_NONE);
-trap:
-    BUMP_ALL_REDS(p);
-    if (mpi.all_objects)
-	(mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
-    hp = HAlloc(p,7+ERTS_MAGIC_REF_THING_SIZE);
-    mpb = erts_db_make_match_prog_ref(p,(mpi.mp),&hp);
-    continuation = TUPLE6(hp, tid, make_small(slot_ix),
-			  make_small(chunk_size), 
-			  mpb, match_list, 
-			  make_small(got));
-    mpi.mp = NULL; /*otherwise the return macro will destroy it */
-    RET_TO_BIF(bif_trap1(&ets_select_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
 
-#undef RET_TO_BIF
+    ret_value = on_loop_ended(context_ptr, slot_ix, got, iterations_left, mpp, ret);
+
+done:
+    /* We should only jump directly to this label if
+     * we've already called on_loop_ended / on_trap
+     */
+    return ret_value;
 
+#ifndef SMP
+#undef lock_hash_function
+#undef unlock_hash_function
+#endif
 }
 
-static int db_select_count_hash(Process *p, 
-				DbTable *tbl,
-                                Eterm tid,
-				Eterm pattern,
-				Eterm *ret)
+
+/*
+ * Common traversal trapping/continuation code;
+ * used by select_count, select_delete and select_replace,
+ * as well as their continuation-handling counterparts.
+ */
+
+static ERTS_INLINE int on_mtraversal_simple_trap(Export* trap_function,
+                                                 Process* p,
+                                                 DbTableHash* tb,
+                                                 Eterm tid,
+                                                 Eterm* prev_continuation_tptr,
+                                                 Sint slot_ix,
+                                                 Sint got,
+                                                 Binary** mpp,
+                                                 Eterm* ret)
 {
-    DbTableHash *tb = &tbl->hash;
-    struct mp_info mpi;
-    Uint slot_ix = 0;
-    HashDbTerm* current = NULL;
-    unsigned current_list_pos = 0;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got = 0;
-    Eterm continuation;
-    int errcode;
+    Eterm* hp;
     Eterm egot;
     Eterm mpb;
-    erts_smp_rwmtx_t* lck;
-
-#define RET_TO_BIF(Term,RetVal) do {		\
-	if (mpi.mp != NULL) {			\
-	    erts_bin_free(mpi.mp);		\
-	}					\
-	if (mpi.lists != mpi.dlists) {		\
-	    erts_free(ERTS_ALC_T_DB_SEL_LIST,	\
-		      (void *) mpi.lists);	\
-	}					\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
-
-
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
-	RET_TO_BIF(NIL,errcode);
-    }
-
-    if (!mpi.something_can_match) {
-	RET_TO_BIF(make_small(0), DB_ERROR_NONE);
-	/* can't possibly match anything */
-    }
-
-    if (!mpi.key_given) {
-    /* Run this code if pattern is variable or GETKEY(pattern)  */
-    /* is a variable                                            */      
-	slot_ix = 0;
-	lck = RLOCK_HASH(tb,slot_ix);
-	current = BUCKET(tb,slot_ix);
-    } else {
-	/* We have at least one */
-	slot_ix = mpi.lists[current_list_pos].ix;
-	lck = RLOCK_HASH(tb, slot_ix);
-	current = *(mpi.lists[current_list_pos].bucket);
-	ASSERT(current == BUCKET(tb,slot_ix));
-	++current_list_pos;
-    }
+    Eterm continuation;
+    int is_first_trap = (prev_continuation_tptr == NULL);
+    size_t base_halloc_sz = (is_first_trap ? ERTS_MAGIC_REF_THING_SIZE : 0);
 
-    for(;;) {
-	if (current != NULL) {
-	    if (current->hvalue != INVALID_HASH) {
-		if (db_match_dbterm(&tb->common, p, mpi.mp, 0,
-				    &current->dbterm, NULL,0) == am_true) {
-		    ++got;
-		}
-		--num_left;
-	    }
-	    current = current->next;
-	}
-	else { /* next bucket */
-	    if (mpi.key_given) {  /* Key is bound */
-		RUNLOCK_HASH(lck);
-		if (current_list_pos == mpi.num_lists) {
-		    goto done;
-		} else {
-		    slot_ix = mpi.lists[current_list_pos].ix;
-		    lck = RLOCK_HASH(tb, slot_ix);
-		    current = *(mpi.lists[current_list_pos].bucket);
-		    ASSERT(mpi.lists[current_list_pos].bucket == &BUCKET(tb,slot_ix));
-		    ++current_list_pos;		       
-		}
-	    }
-	    else {
-		if ((slot_ix=next_slot(tb,slot_ix,&lck)) == 0) {
-		    goto done;
-		}
-		if (num_left <= 0) {
-		    RUNLOCK_HASH(lck);
-		    goto trap;
-		}		
-		current = BUCKET(tb,slot_ix);
-	    }
-	}
-    }
-done:
-    BUMP_REDS(p, 1000 - num_left);
-    RET_TO_BIF(erts_make_integer(got,p),DB_ERROR_NONE);
-trap:
     BUMP_ALL_REDS(p);
     if (IS_USMALL(0, got)) {
-	hp = HAlloc(p,  ERTS_MAGIC_REF_THING_SIZE + 5);
+	hp = HAlloc(p,  base_halloc_sz + 5);
 	egot = make_small(got);
     }
     else {
-	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + ERTS_MAGIC_REF_THING_SIZE + 5);
+	hp = HAlloc(p, base_halloc_sz + BIG_UINT_HEAP_SIZE + 5);
 	egot = uint_to_big(got, hp);
 	hp += BIG_UINT_HEAP_SIZE;
     }
-    mpb = erts_db_make_match_prog_ref(p,mpi.mp,&hp);
-    continuation = TUPLE4(hp, tid, make_small(slot_ix),
-			  mpb, 
-			  egot);
-    mpi.mp = NULL; /*otherwise the return macro will destroy it */
-    RET_TO_BIF(bif_trap1(&ets_select_count_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
 
-#undef RET_TO_BIF
+    if (is_first_trap) {
+        mpb = erts_db_make_match_prog_ref(p, *mpp, &hp);
+        *mpp = NULL; /* otherwise the caller will destroy it */
+    }
+    else {
+        mpb = prev_continuation_tptr[3];
+    }
+
+    continuation = TUPLE4(
+            hp,
+            tid,
+            make_small(slot_ix),
+            mpb,
+            egot);
+    *ret = bif_trap1(trap_function, p, continuation);
+    return DB_ERROR_NONE;
 }
 
-static int db_select_delete_hash(Process *p,
-				 DbTable *tbl,
-                                 Eterm tid,
-				 Eterm pattern,
-				 Eterm *ret)
+static ERTS_INLINE int unpack_simple_mtraversal_continuation(Eterm continuation,
+                                                             Eterm** tptr_ptr,
+                                                             Eterm* tid_ptr,
+                                                             Sint* slot_ix_p,
+                                                             Binary** mpp,
+                                                             Sint* got_p)
 {
-    DbTableHash *tb = &tbl->hash;
-    struct mp_info mpi;
-    Uint slot_ix = 0;
-    HashDbTerm **current = NULL;
-    unsigned current_list_pos = 0;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got = 0;
-    Eterm continuation;
-    int errcode;
-    Uint last_pseudo_delete = (Uint)-1;
-    Eterm mpb;
-    Eterm egot;
-#ifdef ERTS_SMP
-    erts_aint_t fixated_by_me = tb->common.is_thread_safe ? 0 : 1; /* ToDo: something nicer */
-#else
-    erts_aint_t fixated_by_me = 0;
-#endif
-    erts_smp_rwmtx_t* lck;
-
-#define RET_TO_BIF(Term,RetVal) do {		\
-	if (mpi.mp != NULL) {			\
-	    erts_bin_free(mpi.mp);		\
-	}					\
-	if (mpi.lists != mpi.dlists) {		\
-	    erts_free(ERTS_ALC_T_DB_SEL_LIST,	\
-		      (void *) mpi.lists);	\
-	}					\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
-
+    Eterm* tptr;
+    ASSERT(is_tuple(continuation));
+    tptr = tuple_val(continuation);
+    if (arityval(*tptr) != 4)
+        return 1;
 
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
-	RET_TO_BIF(NIL,errcode);
+    if (! is_small(tptr[2]) || !(is_big(tptr[4]) || is_small(tptr[4]))) {
+        return 1;
     }
 
-    if (!mpi.something_can_match) {
-	RET_TO_BIF(make_small(0), DB_ERROR_NONE);
-	/* can't possibly match anything */
+    *tptr_ptr = tptr;
+    *tid_ptr = tptr[1];
+    *slot_ix_p = unsigned_val(tptr[2]);
+    *mpp = erts_db_get_match_prog_binary_unchecked(tptr[3]);
+    if (is_big(tptr[4])) {
+        *got_p = big_to_uint32(tptr[4]);
+    }
+    else {
+        *got_p = unsigned_val(tptr[4]);
     }
+    return 0;
+}
 
-    if (!mpi.key_given) {
-	/* Run this code if pattern is variable or GETKEY(pattern)  */
-	/* is a variable                                            */
-	lck = WLOCK_HASH(tb,slot_ix);
-	current = &BUCKET(tb,slot_ix);
-    } else {
-	/* We have at least one */
-	slot_ix = mpi.lists[current_list_pos].ix;
-	lck = WLOCK_HASH(tb, slot_ix);
-	current = mpi.lists[current_list_pos++].bucket; 
-	ASSERT(*current == BUCKET(tb,slot_ix));
+
+/*
+ *
+ * select / select_chunk match traversal
+ *
+ */
+
+#define MAX_SELECT_CHUNK_ITERATIONS 1000
+
+typedef struct {
+    Process* p;
+    DbTableHash* tb;
+    Eterm tid;
+    Eterm* hp;
+    Sint chunk_size;
+    Eterm match_list;
+    Eterm* prev_continuation_tptr;
+} mtraversal_select_chunk_context_t;
+
+static int mtraversal_select_chunk_on_nothing_can_match(void* context_ptr, Eterm* ret) {
+    mtraversal_select_chunk_context_t* sc_context_ptr = (mtraversal_select_chunk_context_t*) context_ptr;
+    *ret = (sc_context_ptr->chunk_size > 0 ? am_EOT : NIL);
+    return DB_ERROR_NONE;
+}
+
+static int mtraversal_select_chunk_on_match_res(void* context_ptr, Sint slot_ix,
+                                                HashDbTerm*** current_ptr_ptr,
+                                                Eterm match_res)
+{
+    mtraversal_select_chunk_context_t* sc_context_ptr = (mtraversal_select_chunk_context_t*) context_ptr;
+    if (is_value(match_res)) {
+        sc_context_ptr->match_list = CONS(sc_context_ptr->hp, match_res, sc_context_ptr->match_list);
+        return 1;
     }
+    return 0;
+}
 
+static int mtraversal_select_chunk_on_loop_ended(void* context_ptr, Sint slot_ix, Sint got,
+                                                 Sint iterations_left, Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_chunk_context_t* sc_context_ptr = (mtraversal_select_chunk_context_t*) context_ptr;
+    Eterm mpb;
 
-    for(;;) {
-	if ((*current) == NULL) {
-	    if (mpi.key_given) {  /* Key is bound */
-		WUNLOCK_HASH(lck);
-		if (current_list_pos == mpi.num_lists) {
-		    goto done;
-		} else {
-		    slot_ix = mpi.lists[current_list_pos].ix;
-		    lck = WLOCK_HASH(tb, slot_ix);
-		    current = mpi.lists[current_list_pos].bucket;
-		    ASSERT(mpi.lists[current_list_pos].bucket == &BUCKET(tb,slot_ix));
-		    ++current_list_pos;
-		}
-	    } else {
-		if ((slot_ix=next_slot_w(tb,slot_ix,&lck)) == 0) {
-		    goto done;
-		}
-		if (num_left <= 0) {
-		    WUNLOCK_HASH(lck);
-		    goto trap;
-		}		
-		current = &BUCKET(tb,slot_ix);
-	    } 
-	}
-	else if ((*current)->hvalue == INVALID_HASH) {
-	    current = &((*current)->next);
-	} 
-	else {
-	    int did_erase = 0;
-	    if (db_match_dbterm(&tb->common, p, mpi.mp, 0,
-				&(*current)->dbterm, NULL, 0) == am_true) {
-                HashDbTerm *del;
-		if (NFIXED(tb) > fixated_by_me) { /* fixated by others? */
-		    if (slot_ix != last_pseudo_delete) {
-                        if (!add_fixed_deletion(tb, slot_ix, fixated_by_me))
-                            goto do_erase;
-                        last_pseudo_delete = slot_ix;
-		    }
-		    (*current)->hvalue = INVALID_HASH;
-		} else {
-                do_erase:
-		    del = *current;
-		    *current = (*current)->next;
-		    free_term(tb, del);
-		    did_erase = 1;
-		}
-		erts_smp_atomic_dec_nob(&tb->common.nitems);
-		++got;
-	    }	    
-	    --num_left;
-	    if (!did_erase) {
-		current = &((*current)->next);
-	    }
-	}
+    if (iterations_left == MAX_SELECT_CHUNK_ITERATIONS) {
+        /* We didn't get to iterate a single time, which means EOT */
+        ASSERT(sc_context_ptr->match_list == NIL);
+        *ret = (sc_context_ptr->chunk_size > 0 ? am_EOT : NIL);
+        return DB_ERROR_NONE;
     }
-done:
-    BUMP_REDS(p, 1000 - num_left);
-    if (got) {
-	try_shrink(tb);
+    else {
+        ASSERT(iterations_left < MAX_SELECT_CHUNK_ITERATIONS);
+        BUMP_REDS(sc_context_ptr->p, MAX_SELECT_CHUNK_ITERATIONS - iterations_left);
+        if (sc_context_ptr->chunk_size) {
+            Eterm continuation;
+            Eterm rest = NIL;
+            Sint rest_size = 0;
+
+            if (got > sc_context_ptr->chunk_size) { /* Split list in return value and 'rest' */
+                Eterm tmp = sc_context_ptr->match_list;
+                rest = sc_context_ptr->match_list;
+                while (got-- > sc_context_ptr->chunk_size + 1) {
+                    tmp = CDR(list_val(tmp));
+                    ++rest_size;
+                }
+                ++rest_size;
+                sc_context_ptr->match_list = CDR(list_val(tmp));
+                CDR(list_val(tmp)) = NIL; /* Destructive, the list has never
+                                             been in 'user space' */
+            }
+            if (rest != NIL || slot_ix >= 0) { /* Need more calls */
+                sc_context_ptr->hp = HAlloc(sc_context_ptr->p, 3 + 7 + ERTS_MAGIC_REF_THING_SIZE);
+                mpb = erts_db_make_match_prog_ref(sc_context_ptr->p, *mpp, &sc_context_ptr->hp);
+                continuation = TUPLE6(
+                        sc_context_ptr->hp,
+                        sc_context_ptr->tid,
+                        make_small(slot_ix),
+                        make_small(sc_context_ptr->chunk_size),
+                        mpb, rest,
+                        make_small(rest_size));
+                *mpp = NULL; /* Otherwise the caller will destroy it */
+                sc_context_ptr->hp += 7;
+                *ret = TUPLE2(sc_context_ptr->hp, sc_context_ptr->match_list, continuation);
+                return DB_ERROR_NONE;
+            } else { /* All data is exhausted */
+                if (sc_context_ptr->match_list != NIL) { /* No more data to search but still a
+                                                            result to return to the caller */
+                    sc_context_ptr->hp = HAlloc(sc_context_ptr->p, 3);
+                    *ret = TUPLE2(sc_context_ptr->hp, sc_context_ptr->match_list, am_EOT);
+                    return DB_ERROR_NONE;
+                } else { /* Reached the end of the ttable with no data to return */
+                    *ret = am_EOT;
+                    return DB_ERROR_NONE;
+                }
+            }
+        }
+        *ret = sc_context_ptr->match_list;
+        return DB_ERROR_NONE;
     }
-    RET_TO_BIF(erts_make_integer(got,p),DB_ERROR_NONE);
-trap:
-    BUMP_ALL_REDS(p);
-    if (IS_USMALL(0, got)) {
-	hp = HAlloc(p,  ERTS_MAGIC_REF_THING_SIZE + 5);
-	egot = make_small(got);
+}
+
+static int mtraversal_select_chunk_on_trap(void* context_ptr, Sint slot_ix, Sint got,
+                                           Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_chunk_context_t* sc_context_ptr = (mtraversal_select_chunk_context_t*) context_ptr;
+    Eterm mpb;
+    Eterm continuation;
+    Eterm* hp;
+
+    BUMP_ALL_REDS(sc_context_ptr->p);
+
+    if (sc_context_ptr->prev_continuation_tptr == NULL) {
+        /* First time we're trapping */
+        hp = HAlloc(sc_context_ptr->p, 7 + ERTS_MAGIC_REF_THING_SIZE);
+        mpb = erts_db_make_match_prog_ref(sc_context_ptr->p, *mpp, &hp);
+        continuation = TUPLE6(
+                hp,
+                sc_context_ptr->tid,
+                make_small(slot_ix),
+                make_small(sc_context_ptr->chunk_size),
+                mpb,
+                sc_context_ptr->match_list,
+                make_small(got));
+        *mpp = NULL; /* otherwise the caller will destroy it */
     }
     else {
-	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + ERTS_MAGIC_REF_THING_SIZE + 5);
-	egot = uint_to_big(got, hp);
-	hp += BIG_UINT_HEAP_SIZE;
-    }
-    mpb = erts_db_make_match_prog_ref(p,mpi.mp,&hp);
-    continuation = TUPLE4(hp, tid, make_small(slot_ix),
-			  mpb, 
-			  egot);
-    mpi.mp = NULL; /*otherwise the return macro will destroy it */
-    RET_TO_BIF(bif_trap1(&ets_select_delete_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
+        /* Not the first time we're trapping; reuse continuation terms */
+        hp = HAlloc(sc_context_ptr->p, 7);
+        continuation = TUPLE6(
+                hp,
+                sc_context_ptr->prev_continuation_tptr[1],
+                make_small(slot_ix),
+                sc_context_ptr->prev_continuation_tptr[3],
+                sc_context_ptr->prev_continuation_tptr[4],
+                sc_context_ptr->match_list,
+                make_small(got));
+    }
+    *ret = bif_trap1(&ets_select_continue_exp, sc_context_ptr->p, continuation);
+    return DB_ERROR_NONE;
+}
 
-#undef RET_TO_BIF
+static int db_select_hash(Process *p, DbTable *tbl, Eterm tid, Eterm pattern, int reverse, Eterm *ret) {
+    return db_select_chunk_hash(p, tbl, tid, pattern, 0, reverse, ret);
+}
 
+static int db_select_chunk_hash(Process *p, DbTable *tbl, Eterm tid, Eterm pattern, Sint chunk_size,
+                                int reverse, Eterm *ret)
+{
+    mtraversal_select_chunk_context_t sc_context;
+    sc_context.p = p;
+    sc_context.tb = &tbl->hash;
+    sc_context.tid = tid;
+    sc_context.hp = NULL;
+    sc_context.chunk_size = chunk_size;
+    sc_context.match_list = NIL;
+    sc_context.prev_continuation_tptr = NULL;
+
+    return match_traverse(
+            sc_context.p, sc_context.tb,
+            pattern, NULL,
+            sc_context.chunk_size,
+            MAX_SELECT_CHUNK_ITERATIONS,
+            &sc_context.hp, 0,
+            mtraversal_select_chunk_on_nothing_can_match,
+            mtraversal_select_chunk_on_match_res,
+            mtraversal_select_chunk_on_loop_ended,
+            mtraversal_select_chunk_on_trap,
+            &sc_context, ret);
 }
+
 /*
-** This is called when select_delete traps
-*/
-static int db_select_delete_continue_hash(Process *p, 
-					  DbTable *tbl,
-					  Eterm continuation,
-					  Eterm *ret)
+ *
+ * select_continue match traversal
+ *
+ */
+
+static int mtraversal_select_chunk_continue_on_loop_ended(void* context_ptr, Sint slot_ix, Sint got,
+                                                          Sint iterations_left, Binary** mpp, Eterm* ret)
 {
-    DbTableHash *tb = &tbl->hash;
-    Uint slot_ix;
-    Uint last_pseudo_delete = (Uint)-1;
-    HashDbTerm **current = NULL;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got;
-    Eterm *tptr;
-    Binary *mp;
-    Eterm egot;
-    int fixated_by_me = ONLY_WRITER(p,tb) ? 0 : 1; /* ToDo: something nicer */
-    erts_smp_rwmtx_t* lck;
+    mtraversal_select_chunk_context_t* sc_context_ptr = (mtraversal_select_chunk_context_t*) context_ptr;
+    Eterm continuation;
+    Eterm rest = NIL;
+    Eterm* hp;
+
+    ASSERT(iterations_left <= MAX_SELECT_CHUNK_ITERATIONS);
+    BUMP_REDS(sc_context_ptr->p, MAX_SELECT_CHUNK_ITERATIONS - iterations_left);
+    if (sc_context_ptr->chunk_size) {
+        Sint rest_size = 0;
+        if (got > sc_context_ptr->chunk_size) {
+            /* Cannot write destructively here,
+               the list may have
+               been in user space */
+            hp = HAlloc(sc_context_ptr->p, (got - sc_context_ptr->chunk_size) * 2);
+            while (got-- > sc_context_ptr->chunk_size) {
+                rest = CONS(hp, CAR(list_val(sc_context_ptr->match_list)), rest);
+                hp += 2;
+                sc_context_ptr->match_list = CDR(list_val(sc_context_ptr->match_list));
+                ++rest_size;
+            }
+        }
+        if (rest != NIL || slot_ix >= 0) {
+            hp = HAlloc(sc_context_ptr->p, 3 + 7);
+            continuation = TUPLE6(
+                    hp,
+                    sc_context_ptr->prev_continuation_tptr[1],
+                    make_small(slot_ix),
+                    sc_context_ptr->prev_continuation_tptr[3],
+                    sc_context_ptr->prev_continuation_tptr[4],
+                    rest,
+                    make_small(rest_size));
+            hp += 7;
+            *ret = TUPLE2(hp, sc_context_ptr->match_list, continuation);
+            return DB_ERROR_NONE;
+        } else {
+            if (sc_context_ptr->match_list != NIL) {
+                hp = HAlloc(sc_context_ptr->p, 3);
+                *ret = TUPLE2(hp, sc_context_ptr->match_list, am_EOT);
+                return DB_ERROR_NONE;
+            } else {
+                *ret = am_EOT;
+                return DB_ERROR_NONE;
+            }
+        }
+    }
+    *ret = sc_context_ptr->match_list;
+    return DB_ERROR_NONE;
+}
 
-#define RET_TO_BIF(Term,RetVal) do {		\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
+/*
+ * This is called when select traps
+ */
+static int db_select_continue_hash(Process* p, DbTable* tbl, Eterm continuation, Eterm* ret) {
+    mtraversal_select_chunk_context_t sc_context = {0};
+    Eterm* tptr;
+    Eterm tid;
+    Binary* mp;
+    Sint got;
+    Sint slot_ix;
+    Sint chunk_size;
+    Eterm match_list;
+    Sint iterations_left = MAX_SELECT_CHUNK_ITERATIONS;
 
-    
+    /* Decode continuation. We know it's a tuple but not the arity or anything else */
+    ASSERT(is_tuple(continuation));
     tptr = tuple_val(continuation);
-    slot_ix = unsigned_val(tptr[2]);
-    mp = erts_db_get_match_prog_binary_unchecked(tptr[3]);
-    if (is_big(tptr[4])) {
-	got = big_to_uint32(tptr[4]);
-    } else {
-	got = unsigned_val(tptr[4]);
-    }
-    
-    lck = WLOCK_HASH(tb,slot_ix);
-    if (slot_ix >= NACTIVE(tb)) {
-	WUNLOCK_HASH(lck);
-	goto done;
-    } 
-    current = &BUCKET(tb,slot_ix);
 
-    for(;;) {
-	if ((*current) == NULL) {
-	    if ((slot_ix=next_slot_w(tb,slot_ix,&lck)) == 0) {
-		goto done;
-	    }
-	    if (num_left <= 0) {
-		WUNLOCK_HASH(lck);
-		goto trap;
-	    }		
-	    current = &BUCKET(tb,slot_ix);
-	}
-	else if ((*current)->hvalue == INVALID_HASH) {
-	    current = &((*current)->next);
-	} 
-	else {
-	    int did_erase = 0;
-	    if (db_match_dbterm(&tb->common, p, mp, 0,
-				&(*current)->dbterm, NULL, 0) == am_true) {
-                HashDbTerm *del;
-		if (NFIXED(tb) > fixated_by_me) { /* fixated by others? */
-		    if (slot_ix != last_pseudo_delete) {
-                        if (!add_fixed_deletion(tb, slot_ix, fixated_by_me))
-                            goto do_erase;
-			last_pseudo_delete = slot_ix;
-		    }
-		    (*current)->hvalue = INVALID_HASH;
-		} else {
-                do_erase:
-		    del = *current;
-		    *current = (*current)->next;
-		    free_term(tb, del);
-		    did_erase = 1;
-		}
-		erts_smp_atomic_dec_nob(&tb->common.nitems);
-		++got;
-	    }
-	    
-	    --num_left;
-	    if (!did_erase) {
-		current = &((*current)->next);
-	    }
-	}
-    }
-done:
-    BUMP_REDS(p, 1000 - num_left);
-    if (got) {
-	try_shrink(tb);
-    }
-    RET_TO_BIF(erts_make_integer(got,p),DB_ERROR_NONE);
-trap:
-    BUMP_ALL_REDS(p);
-    if (IS_USMALL(0, got)) {
-	hp = HAlloc(p,  5);
-	egot = make_small(got);
-    }
-    else {
-	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + 5);
-	egot = uint_to_big(got, hp);
-	hp += BIG_UINT_HEAP_SIZE;
-    }
-    continuation = TUPLE4(hp, tptr[1], make_small(slot_ix),
-			  tptr[3], 
-			  egot);
-    RET_TO_BIF(bif_trap1(&ets_select_delete_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
+    if (arityval(*tptr) != 6)
+        goto badparam;
+
+    if (!is_small(tptr[2]) || !is_small(tptr[3]) ||
+            !(is_list(tptr[5]) || tptr[5] == NIL) || !is_small(tptr[6]))
+        goto badparam;
+    if ((chunk_size = signed_val(tptr[3])) < 0)
+        goto badparam;
+
+    mp = erts_db_get_match_prog_binary(tptr[4]);
+    if (mp == NULL)
+        goto badparam;
 
-#undef RET_TO_BIF
+    if ((got = signed_val(tptr[6])) < 0)
+        goto badparam;
+
+    tid = tptr[1];
+    slot_ix = signed_val(tptr[2]);
+    match_list = tptr[5];
 
+    /* Proceed */
+    sc_context.p = p;
+    sc_context.tb = &tbl->hash;
+    sc_context.tid = tid;
+    sc_context.hp = NULL;
+    sc_context.chunk_size = chunk_size;
+    sc_context.match_list = match_list;
+    sc_context.prev_continuation_tptr = tptr;
+
+    return match_traverse_continue(
+            sc_context.p, sc_context.tb, sc_context.chunk_size,
+            iterations_left, &sc_context.hp, slot_ix, got, &mp, 0,
+            mtraversal_select_chunk_on_match_res, /* Reuse callback */
+            mtraversal_select_chunk_continue_on_loop_ended,
+            mtraversal_select_chunk_on_trap,      /* Reuse callback */
+            &sc_context, ret);
+
+badparam:
+    *ret = NIL;
+    return DB_ERROR_BADPARAM;
 }
-    
+
+#undef MAX_SELECT_CHUNK_ITERATIONS
+
+
 /*
-** This is called when select_count traps
-*/
-static int db_select_count_continue_hash(Process *p, 
-					 DbTable *tbl,
-					 Eterm continuation,
-					 Eterm *ret)
+ *
+ * select_count match traversal
+ *
+ */
+
+#define MAX_SELECT_COUNT_ITERATIONS 1000
+
+typedef struct {
+    Process* p;
+    DbTableHash* tb;
+    Eterm tid;
+    Eterm* hp;
+    Eterm* prev_continuation_tptr;
+} mtraversal_select_count_context_t;
+
+static int mtraversal_select_count_on_nothing_can_match(void* context_ptr, Eterm* ret) {
+    *ret = make_small(0);
+    return DB_ERROR_NONE;
+}
+
+static int mtraversal_select_count_on_match_res(void* context_ptr, Sint slot_ix,
+                                                HashDbTerm*** current_ptr_ptr,
+                                                Eterm match_res)
 {
-    DbTableHash *tb = &tbl->hash;
-    Uint slot_ix;
-    HashDbTerm* current;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got;
-    Eterm *tptr;
-    Binary *mp;
-    Eterm egot;
-    erts_smp_rwmtx_t* lck;
+    return (match_res == am_true);
+}
 
-#define RET_TO_BIF(Term,RetVal) do {		\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
+static int mtraversal_select_count_on_loop_ended(void* context_ptr, Sint slot_ix, Sint got,
+                                                 Sint iterations_left, Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_count_context_t* scnt_context_ptr = (mtraversal_select_count_context_t*) context_ptr;
+    ASSERT(iterations_left <= MAX_SELECT_COUNT_ITERATIONS);
+    BUMP_REDS(scnt_context_ptr->p, MAX_SELECT_COUNT_ITERATIONS - iterations_left);
+    *ret = erts_make_integer(got, scnt_context_ptr->p);
+    return DB_ERROR_NONE;
+}
 
-    
-    tptr = tuple_val(continuation);
-    slot_ix = unsigned_val(tptr[2]);
-    mp = erts_db_get_match_prog_binary_unchecked(tptr[3]);
-    if (is_big(tptr[4])) {
-	got = big_to_uint32(tptr[4]);
-    } else {
-	got = unsigned_val(tptr[4]);
-    }
-    
+static int mtraversal_select_count_on_trap(void* context_ptr, Sint slot_ix, Sint got,
+                                           Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_count_context_t* scnt_context_ptr = (mtraversal_select_count_context_t*) context_ptr;
+    return on_mtraversal_simple_trap(
+            &ets_select_count_continue_exp,
+            scnt_context_ptr->p,
+            scnt_context_ptr->tb,
+            scnt_context_ptr->tid,
+            scnt_context_ptr->prev_continuation_tptr,
+            slot_ix, got, mpp, ret);
+}
 
-    lck = RLOCK_HASH(tb, slot_ix);
-    if (slot_ix >= NACTIVE(tb)) { /* Is this posible? */
-	RUNLOCK_HASH(lck);
-	goto done;
-    }     
-    current = BUCKET(tb,slot_ix);
-    
-    for(;;) {
-	if (current != NULL) {
-	    if (current->hvalue == INVALID_HASH) {
-		current = current->next;
-		continue;
-	    }
-	    if (db_match_dbterm(&tb->common, p, mp, 0, &current->dbterm,
-				NULL, 0) == am_true) {
-		++got;
-	    }
-	    --num_left;
-	    current = current->next;
-	}
-	else { /* next bucket */
-            if ((slot_ix = next_slot(tb,slot_ix,&lck)) == 0) {
-		goto done;
-	    }
-	    if (num_left <= 0) {
-		RUNLOCK_HASH(lck);
-		goto trap;
-	    }
-	    current = BUCKET(tb,slot_ix);
-	}
-    }
-done:
-    BUMP_REDS(p, 1000 - num_left);
-    RET_TO_BIF(erts_make_integer(got,p),DB_ERROR_NONE);
-trap:
-    BUMP_ALL_REDS(p);
-    if (IS_USMALL(0, got)) {
-	hp = HAlloc(p, 5);
-	egot = make_small(got);
+static int db_select_count_hash(Process *p, DbTable *tbl, Eterm tid, Eterm pattern, Eterm *ret) {
+    mtraversal_select_count_context_t scnt_context = {0};
+    Sint iterations_left = MAX_SELECT_COUNT_ITERATIONS;
+    Sint chunk_size = 0;
+
+    scnt_context.p = p;
+    scnt_context.tb = &tbl->hash;
+    scnt_context.tid = tid;
+    scnt_context.hp = NULL;
+    scnt_context.prev_continuation_tptr = NULL;
+
+    return match_traverse(
+            scnt_context.p, scnt_context.tb,
+            pattern, NULL,
+            chunk_size, iterations_left, NULL, 0,
+            mtraversal_select_count_on_nothing_can_match,
+            mtraversal_select_count_on_match_res,
+            mtraversal_select_count_on_loop_ended,
+            mtraversal_select_count_on_trap,
+            &scnt_context, ret);
+}
+
+/*
+ * This is called when select_count traps
+ */
+static int db_select_count_continue_hash(Process* p, DbTable* tbl, Eterm continuation, Eterm* ret) {
+    mtraversal_select_count_context_t scnt_context = {0};
+    Eterm* tptr;
+    Eterm tid;
+    Binary* mp;
+    Sint got;
+    Sint slot_ix;
+    Sint chunk_size = 0;
+    *ret = NIL;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        *ret = NIL;
+        return DB_ERROR_BADPARAM;
+    }
+
+    scnt_context.p = p;
+    scnt_context.tb = &tbl->hash;
+    scnt_context.tid = tid;
+    scnt_context.hp = NULL;
+    scnt_context.prev_continuation_tptr = tptr;
+
+    return match_traverse_continue(
+            scnt_context.p, scnt_context.tb, chunk_size,
+            MAX_SELECT_COUNT_ITERATIONS,
+            NULL, slot_ix, got, &mp, 0,
+            mtraversal_select_count_on_match_res,  /* Reuse callback */
+            mtraversal_select_count_on_loop_ended, /* Reuse callback */
+            mtraversal_select_count_on_trap,       /* Reuse callback */
+            &scnt_context, ret);
+}
+
+#undef MAX_SELECT_COUNT_ITERATIONS
+
+
+/*
+ *
+ * select_delete match traversal
+ *
+ */
+
+#define MAX_SELECT_DELETE_ITERATIONS 1000
+
+typedef struct {
+    Process* p;
+    DbTableHash* tb;
+    Eterm tid;
+    Eterm* hp;
+    Eterm* prev_continuation_tptr;
+    erts_aint_t fixated_by_me;
+    Uint last_pseudo_delete;
+} mtraversal_select_delete_context_t;
+
+static int mtraversal_select_delete_on_nothing_can_match(void* context_ptr, Eterm* ret) {
+    *ret = make_small(0);
+    return DB_ERROR_NONE;
+}
+
+static int mtraversal_select_delete_on_match_res(void* context_ptr, Sint slot_ix,
+                                                 HashDbTerm*** current_ptr_ptr,
+                                                 Eterm match_res)
+{
+    HashDbTerm** current_ptr = *current_ptr_ptr;
+    mtraversal_select_delete_context_t* sd_context_ptr = (mtraversal_select_delete_context_t*) context_ptr;
+    HashDbTerm* del;
+    if (match_res != am_true)
+        return 0;
+
+    if (NFIXED(sd_context_ptr->tb) > sd_context_ptr->fixated_by_me) { /* fixated by others? */
+        if (slot_ix != sd_context_ptr->last_pseudo_delete) {
+            if (!add_fixed_deletion(sd_context_ptr->tb, slot_ix, sd_context_ptr->fixated_by_me))
+                goto do_erase;
+            sd_context_ptr->last_pseudo_delete = slot_ix;
+        }
+        (*current_ptr)->hvalue = INVALID_HASH;
     }
     else {
-	hp = HAlloc(p, BIG_UINT_HEAP_SIZE + 5);
-	egot = uint_to_big(got, hp);
-	hp += BIG_UINT_HEAP_SIZE;
+    do_erase:
+        del = *current_ptr;
+        *current_ptr = (*current_ptr)->next; // replace pointer to term using next
+        free_term(sd_context_ptr->tb, del);
     }
-    continuation = TUPLE4(hp, tptr[1], make_small(slot_ix),
-			  tptr[3], 
-			  egot);
-    RET_TO_BIF(bif_trap1(&ets_select_count_continue_exp, p, 
-			 continuation), 
-	       DB_ERROR_NONE);
+    erts_smp_atomic_dec_nob(&sd_context_ptr->tb->common.nitems);
 
-#undef RET_TO_BIF
+    return 1;
+}
 
+static int mtraversal_select_delete_on_loop_ended(void* context_ptr, Sint slot_ix, Sint got,
+                                                  Sint iterations_left, Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_delete_context_t* sd_context_ptr = (mtraversal_select_delete_context_t*) context_ptr;
+    ASSERT(iterations_left <= MAX_SELECT_DELETE_ITERATIONS);
+    BUMP_REDS(sd_context_ptr->p, MAX_SELECT_DELETE_ITERATIONS - iterations_left);
+    if (got) {
+	try_shrink(sd_context_ptr->tb);
+    }
+    *ret = erts_make_integer(got, sd_context_ptr->p);
+    return DB_ERROR_NONE;
 }
-    
+
+static int mtraversal_select_delete_on_trap(void* context_ptr, Sint slot_ix, Sint got,
+                                            Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_delete_context_t* sd_context_ptr = (mtraversal_select_delete_context_t*) context_ptr;
+    return on_mtraversal_simple_trap(
+            &ets_select_delete_continue_exp,
+            sd_context_ptr->p,
+            sd_context_ptr->tb,
+            sd_context_ptr->tid,
+            sd_context_ptr->prev_continuation_tptr,
+            slot_ix, got, mpp, ret);
+}
+
+static int db_select_delete_hash(Process *p, DbTable *tbl, Eterm tid, Eterm pattern, Eterm *ret) {
+    mtraversal_select_delete_context_t sd_context = {0};
+    Sint chunk_size = 0;
+
+    sd_context.p = p;
+    sd_context.tb = &tbl->hash;
+    sd_context.tid = tid;
+    sd_context.hp = NULL;
+    sd_context.prev_continuation_tptr = NULL;
+#ifdef ERTS_SMP
+    sd_context.fixated_by_me = sd_context.tb->common.is_thread_safe ? 0 : 1; /* TODO: something nicer */
+#else
+    sd_context.fixated_by_me = 0;
+#endif
+    sd_context.last_pseudo_delete = (Uint) -1;
+
+    return match_traverse(
+            sd_context.p, sd_context.tb,
+            pattern, NULL,
+            chunk_size,
+            MAX_SELECT_DELETE_ITERATIONS, NULL, 1,
+            mtraversal_select_delete_on_nothing_can_match,
+            mtraversal_select_delete_on_match_res,
+            mtraversal_select_delete_on_loop_ended,
+            mtraversal_select_delete_on_trap,
+            &sd_context, ret);
+}
+
+/*
+ * This is called when select_delete traps
+ */
+static int db_select_delete_continue_hash(Process* p, DbTable* tbl, Eterm continuation, Eterm* ret) {
+    mtraversal_select_delete_context_t sd_context = {0};
+    Eterm* tptr;
+    Eterm tid;
+    Binary* mp;
+    Sint got;
+    Sint slot_ix;
+    Sint chunk_size = 0;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        *ret = NIL;
+        return DB_ERROR_BADPARAM;
+    }
+
+    sd_context.p = p;
+    sd_context.tb = &tbl->hash;
+    sd_context.tid = tid;
+    sd_context.hp = NULL;
+    sd_context.prev_continuation_tptr = tptr;
+    sd_context.fixated_by_me = ONLY_WRITER(p, sd_context.tb) ? 0 : 1; /* TODO: something nicer */
+    sd_context.last_pseudo_delete = (Uint) -1;
+
+    return match_traverse_continue(
+            sd_context.p, sd_context.tb, chunk_size,
+            MAX_SELECT_DELETE_ITERATIONS,
+            NULL, slot_ix, got, &mp, 1,
+            mtraversal_select_delete_on_match_res,  /* Reuse callback */
+            mtraversal_select_delete_on_loop_ended, /* Reuse callback */
+            mtraversal_select_delete_on_trap,       /* Reuse callback */
+            &sd_context, ret);
+}
+
+#undef MAX_SELECT_DELETE_ITERATIONS
+
+
+/*
+ *
+ * select_replace match traversal
+ *
+ */
+
+#define MAX_SELECT_REPLACE_ITERATIONS 1000
+
+typedef struct {
+    Process* p;
+    DbTableHash* tb;
+    Eterm tid;
+    Eterm* hp;
+    Eterm* prev_continuation_tptr;
+} mtraversal_select_replace_context_t;
+
+static int mtraversal_select_replace_on_nothing_can_match(void* context_ptr, Eterm* ret) {
+    *ret = make_small(0);
+    return DB_ERROR_NONE;
+}
+
+static int mtraversal_select_replace_on_match_res(void* context_ptr, Sint slot_ix,
+                                                  HashDbTerm*** current_ptr_ptr,
+                                                  Eterm match_res)
+{
+    mtraversal_select_replace_context_t* sr_context_ptr = (mtraversal_select_replace_context_t*) context_ptr;
+    DbTableHash* tb = sr_context_ptr->tb;
+    HashDbTerm* new;
+    HashDbTerm* next;
+    HashValue hval;
+
+    if (is_value(match_res)) {
+#ifdef DEBUG
+        Eterm key = db_getkey(tb->common.keypos, match_res);
+        ASSERT(is_value(key));
+        ASSERT(eq(key, GETKEY(tb, (**current_ptr_ptr)->dbterm.tpl)));
+#endif
+        next = (**current_ptr_ptr)->next;
+        hval = (**current_ptr_ptr)->hvalue;
+        new = new_dbterm(tb, match_res);
+        new->next = next;
+        new->hvalue = hval;
+        free_term(tb, **current_ptr_ptr);
+        **current_ptr_ptr = new; /* replace 'next' pointer in previous object */
+        *current_ptr_ptr = &((**current_ptr_ptr)->next); /* advance to next object */
+        return 1;
+    }
+    return 0;
+}
+
+static int mtraversal_select_replace_on_loop_ended(void* context_ptr, Sint slot_ix, Sint got,
+                                                   Sint iterations_left, Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_replace_context_t* sr_context_ptr = (mtraversal_select_replace_context_t*) context_ptr;
+    ASSERT(iterations_left <= MAX_SELECT_REPLACE_ITERATIONS);
+    /* the more objects we've replaced, the more reductions we've consumed */
+    BUMP_REDS(sr_context_ptr->p,
+              MIN(MAX_SELECT_REPLACE_ITERATIONS * 2,
+                  (MAX_SELECT_REPLACE_ITERATIONS - iterations_left) + (int)got));
+    *ret = erts_make_integer(got, sr_context_ptr->p);
+    return DB_ERROR_NONE;
+}
+
+static int mtraversal_select_replace_on_trap(void* context_ptr, Sint slot_ix, Sint got,
+                                             Binary** mpp, Eterm* ret)
+{
+    mtraversal_select_replace_context_t* sr_context_ptr = (mtraversal_select_replace_context_t*) context_ptr;
+    return on_mtraversal_simple_trap(
+            &ets_select_replace_continue_exp,
+            sr_context_ptr->p,
+            sr_context_ptr->tb,
+            sr_context_ptr->tid,
+            sr_context_ptr->prev_continuation_tptr,
+            slot_ix, got, mpp, ret);
+}
+
+static int db_select_replace_hash(Process *p, DbTable *tbl, Eterm tid, Eterm pattern, Eterm *ret)
+{
+    mtraversal_select_replace_context_t sr_context = {0};
+    Sint chunk_size = 0;
+
+    /* Bag implementation presented both semantic consistency and performance issues,
+     * unsupported for now
+     */
+    ASSERT(!(tbl->hash.common.status & DB_BAG));
+
+    sr_context.p = p;
+    sr_context.tb = &tbl->hash;
+    sr_context.tid = tid;
+    sr_context.hp = NULL;
+    sr_context.prev_continuation_tptr = NULL;
+
+    return match_traverse(
+            sr_context.p, sr_context.tb,
+            pattern, db_match_keeps_key,
+            chunk_size,
+            MAX_SELECT_REPLACE_ITERATIONS, NULL, 1,
+            mtraversal_select_replace_on_nothing_can_match,
+            mtraversal_select_replace_on_match_res,
+            mtraversal_select_replace_on_loop_ended,
+            mtraversal_select_replace_on_trap,
+            &sr_context, ret);
+}
+
+/*
+ * This is called when select_replace traps
+ */
+static int db_select_replace_continue_hash(Process* p, DbTable* tbl, Eterm continuation, Eterm* ret)
+{
+    mtraversal_select_replace_context_t sr_context = {0};
+    Eterm* tptr;
+    Eterm tid ;
+    Binary* mp;
+    Sint got;
+    Sint slot_ix;
+    Sint chunk_size = 0;
+    *ret = NIL;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        *ret = NIL;
+        return DB_ERROR_BADPARAM;
+    }
+
+    /* Proceed */
+    sr_context.p = p;
+    sr_context.tb = &tbl->hash;
+    sr_context.tid = tid;
+    sr_context.hp = NULL;
+    sr_context.prev_continuation_tptr = tptr;
+
+    return match_traverse_continue(
+            sr_context.p, sr_context.tb, chunk_size,
+            MAX_SELECT_REPLACE_ITERATIONS,
+            NULL, slot_ix, got, &mp, 1,
+            mtraversal_select_replace_on_match_res,  /* Reuse callback */
+            mtraversal_select_replace_on_loop_ended, /* Reuse callback */
+            mtraversal_select_replace_on_trap,       /* Reuse callback */
+            &sr_context, ret);
+}
+
+
 static int db_take_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
 {
     DbTableHash *tb = &tbl->hash;
@@ -1989,6 +2289,7 @@ static int db_take_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
     return DB_ERROR_NONE;
 }
 
+
 /*
 ** Other interface routines (not directly coupled to one bif)
 */
@@ -2147,7 +2448,8 @@ static SWord db_free_table_continue_hash(DbTable *tbl, SWord reds)
 ** slots should be searched. Also compiles the match program
 */
 static int analyze_pattern(DbTableHash *tb, Eterm pattern, 
-			   struct mp_info *mpi)
+                           extra_match_validator_t extra_validator, /* Optional callback */
+                           struct mp_info *mpi)
 {
     Eterm *ptpl;
     Eterm lst, tpl, ttpl;
@@ -2185,7 +2487,10 @@ static int analyze_pattern(DbTableHash *tb, Eterm pattern,
 
     i = 0;
     for(lst = pattern; is_list(lst); lst = CDR(list_val(lst))) {
-	Eterm body;
+        Eterm match;
+        Eterm guard;
+        Eterm body;
+
 	ttpl = CAR(list_val(lst));
 	if (!is_tuple(ttpl)) {
 	    if (buff != sbuff) { 
@@ -2200,9 +2505,17 @@ static int analyze_pattern(DbTableHash *tb, Eterm pattern,
 	    }
 	    return DB_ERROR_BADPARAM;
 	}
-	matches[i] = tpl = ptpl[1];
-	guards[i] = ptpl[2];
+	matches[i] = match = tpl = ptpl[1];
+	guards[i] = guard = ptpl[2];
 	bodies[i] = body = ptpl[3];
+
+        if(extra_validator != NULL && !extra_validator(tb->common.keypos, match, guard, body)) {
+	    if (buff != sbuff) {
+		erts_free(ERTS_ALC_T_DB_TMP, buff);
+	    }
+            return DB_ERROR_BADPARAM;
+        }
+
 	if (!is_list(body) || CDR(list_val(body)) != NIL ||
 	    CAR(list_val(body)) != am_DollarUnderscore) {
 	    mpi->all_objects = 0;
diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
index 14a7451267..ab8da6ccf6 100644
--- a/erts/emulator/beam/erl_db_tree.c
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -76,9 +76,18 @@
      ((Dtt->pos) ? 			\
       (Dtt)->array[(Dtt)->pos - 1] : NULL)
 
-#define EMPTY_NODE(Dtt) (TOP_NODE(Dtt) == NULL)
+#define TOPN_NODE(Dtt, Pos)                   \
+     (((Pos) < Dtt->pos) ? 			\
+      (Dtt)->array[(Dtt)->pos - ((Pos) + 1)] : NULL)
+
+#define REPLACE_TOP_NODE(Dtt, Node)          \
+     if ((Dtt)->pos) (Dtt)->array[(Dtt)->pos - 1] = (Node)
 
+#define EMPTY_NODE(Dtt) (TOP_NODE(Dtt) == NULL)
 
+#ifndef MIN
+#define MIN(X, Y) ((X) < (Y) ? (X) : (Y))
+#endif
 
 /* Obtain table static stack if available. NULL if not.
 ** Must be released with release_stack()
@@ -225,9 +234,9 @@ struct mp_info {
     Eterm most;                 /* The highest matching key (possibly 
 				 * partially bound expression) */
 
-    TreeDbTerm *save_term;      /* If the key is completely bound, this
-				 * will be the Tree node we're searching
-				 * for, otherwise it will be useless */
+    TreeDbTerm **save_term;      /* If the key is completely bound, this
+	 			  * will be the Tree node we're searching
+				  * for, otherwise it will be useless */
     Binary *mp;                 /* The compiled match program */
 };
 
@@ -277,6 +286,24 @@ struct select_delete_context {
 };
 
 /*
+ * Used by doit_select_replace
+ */
+struct select_replace_context {
+    Process *p;
+    DbTableTree *tb;
+    Binary *mp;
+    Eterm end_condition;
+    Eterm *lastobj;
+    Sint32 max;
+    int keypos;
+    int all_objects;
+    Sint replaced;
+};
+
+/* Used by select_replace on analyze_pattern */
+typedef int (*extra_match_validator_t)(int keypos, Eterm match, Eterm guard, Eterm body);
+
+/*
 ** Forward declarations 
 */
 static TreeDbTerm *linkout_tree(DbTableTree *tb, Eterm key);
@@ -290,6 +317,7 @@ static int delsub(TreeDbTerm **this);
 static TreeDbTerm *slot_search(Process *p, DbTableTree *tb, Sint slot);
 static TreeDbTerm *find_node(DbTableTree *tb, Eterm key);
 static TreeDbTerm **find_node2(DbTableTree *tb, Eterm key);
+static TreeDbTerm **find_ptr(DbTableTree *tb, DbTreeStack*, TreeDbTerm *this);
 static TreeDbTerm *find_next(DbTableTree *tb, DbTreeStack*, Eterm key);
 static TreeDbTerm *find_prev(DbTableTree *tb, DbTreeStack*, Eterm key);
 static TreeDbTerm *find_next_from_pb_key(DbTableTree *tb, DbTreeStack*,
@@ -311,14 +339,23 @@ static void traverse_forward(DbTableTree *tb,
 					 TreeDbTerm *,
 					 void *,
 					 int),
-			     void *context); 
-static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm **ret,
+			     void *context);
+static void traverse_update_backwards(DbTableTree *tb,
+                                      DbTreeStack*,
+                                      Eterm lastkey,
+                                      int (*doit)(DbTableTree *tb,
+                                                  TreeDbTerm **, // out
+                                                  void *,
+                                                  int),
+                                      void *context);
+static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm ***ret,
 		     Eterm *partly_bound_key);
 static Sint cmp_partly_bound(Eterm partly_bound_key, Eterm bound_key);
 static Sint do_cmp_partly_bound(Eterm a, Eterm b, int *done);
 
 static int analyze_pattern(DbTableTree *tb, Eterm pattern, 
-			   struct mp_info *mpi);
+                           extra_match_validator_t extra_validator, /* Optional callback */
+                           struct mp_info *mpi);
 static int doit_select(DbTableTree *tb,
 		       TreeDbTerm *this,
 		       void *ptr,
@@ -335,6 +372,10 @@ static int doit_select_delete(DbTableTree *tb,
 			      TreeDbTerm *this,
 			      void *ptr,
 			      int forward);
+static int doit_select_replace(DbTableTree *tb,
+                               TreeDbTerm **this_ptr,
+                               void *ptr,
+                               int forward);
 
 static int partly_bound_can_match_lesser(Eterm partly_bound_1, 
 					 Eterm partly_bound_2);
@@ -383,6 +424,10 @@ static int db_select_delete_tree(Process *p, DbTable *tbl, Eterm tid,
 				 Eterm pattern,  Eterm *ret);
 static int db_select_delete_continue_tree(Process *p, DbTable *tbl, 
 					  Eterm continuation, Eterm *ret);
+static int db_select_replace_tree(Process *p, DbTable *tbl, Eterm tid,
+                                  Eterm pattern, Eterm *ret);
+static int db_select_replace_continue_tree(Process *p, DbTable *tbl,
+                                           Eterm continuation, Eterm *ret);
 static int db_take_tree(Process *, DbTable *, Eterm, Eterm *);
 static void db_print_tree(fmtfn_t to, void *to_arg,
 			  int show, DbTable *tbl);
@@ -435,6 +480,8 @@ DbTableMethod db_tree =
     db_select_delete_continue_tree,
     db_select_count_tree,
     db_select_count_continue_tree,
+    db_select_replace_tree,
+    db_select_replace_continue_tree,
     db_take_tree,
     db_delete_all_objects_tree,
     db_free_table_tree,
@@ -1089,7 +1136,7 @@ static int db_select_tree(Process *p, DbTable *tbl, Eterm tid,
     sc.got = 0;
     sc.chunk_size = 0;
 
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+    if ((errcode = analyze_pattern(tb, pattern, NULL, &mpi)) != DB_ERROR_NONE) {
 	RET_TO_BIF(NIL,errcode);
     }
 
@@ -1103,7 +1150,7 @@ static int db_select_tree(Process *p, DbTable *tbl, Eterm tid,
 
     if (!mpi.got_partial && mpi.some_limitation && 
 	CMP_EQ(mpi.least,mpi.most)) {
-	doit_select(tb,mpi.save_term,&sc,0 /* direction doesn't matter */);
+	doit_select(tb,*(mpi.save_term),&sc,0 /* direction doesn't matter */);
 	RET_TO_BIF(sc.accum,DB_ERROR_NONE);
     }
 
@@ -1292,7 +1339,7 @@ static int db_select_count_tree(Process *p, DbTable *tbl, Eterm tid,
     sc.keypos = tb->common.keypos;
     sc.got = 0;
 
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+    if ((errcode = analyze_pattern(tb, pattern, NULL, &mpi)) != DB_ERROR_NONE) {
 	RET_TO_BIF(NIL,errcode);
     }
 
@@ -1306,7 +1353,7 @@ static int db_select_count_tree(Process *p, DbTable *tbl, Eterm tid,
 
     if (!mpi.got_partial && mpi.some_limitation && 
 	CMP_EQ(mpi.least,mpi.most)) {
-	doit_select_count(tb,mpi.save_term,&sc,0 /* dummy */);
+	doit_select_count(tb,*(mpi.save_term),&sc,0 /* dummy */);
 	RET_TO_BIF(erts_make_integer(sc.got,p),DB_ERROR_NONE);
     }
 
@@ -1395,7 +1442,7 @@ static int db_select_chunk_tree(Process *p, DbTable *tbl, Eterm tid,
     sc.got = 0;
     sc.chunk_size = chunk_size;
 
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+    if ((errcode = analyze_pattern(tb, pattern, NULL, &mpi)) != DB_ERROR_NONE) {
 	RET_TO_BIF(NIL,errcode);
     }
 
@@ -1409,7 +1456,7 @@ static int db_select_chunk_tree(Process *p, DbTable *tbl, Eterm tid,
 
     if (!mpi.got_partial && mpi.some_limitation && 
 	CMP_EQ(mpi.least,mpi.most)) {
-	doit_select(tb,mpi.save_term,&sc, 0 /* direction doesn't matter */);
+	doit_select(tb,*(mpi.save_term),&sc, 0 /* direction doesn't matter */);
 	if (sc.accum != NIL) {
 	    hp=HAlloc(p, 3);
 	    RET_TO_BIF(TUPLE2(hp,sc.accum,am_EOT),DB_ERROR_NONE);
@@ -1637,7 +1684,7 @@ static int db_select_delete_tree(Process *p, DbTable *tbl, Eterm tid,
     sc.keypos = tb->common.keypos;
     sc.tb = tb;
     
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+    if ((errcode = analyze_pattern(tb, pattern, NULL, &mpi)) != DB_ERROR_NONE) {
 	RET_TO_BIF(0,errcode);
     }
 
@@ -1650,7 +1697,7 @@ static int db_select_delete_tree(Process *p, DbTable *tbl, Eterm tid,
 
     if (!mpi.got_partial && mpi.some_limitation && 
 	CMP_EQ(mpi.least,mpi.most)) {
-	doit_select_delete(tb,mpi.save_term,&sc, 0 /* direction doesn't 
+	doit_select_delete(tb,*(mpi.save_term),&sc, 0 /* direction doesn't
 						      matter */);
 	RET_TO_BIF(erts_make_integer(sc.accum,p),DB_ERROR_NONE);
     }
@@ -1702,6 +1749,208 @@ static int db_select_delete_tree(Process *p, DbTable *tbl, Eterm tid,
 
 }
 
+static int db_select_replace_continue_tree(Process *p,
+                                           DbTable *tbl,
+                                           Eterm continuation,
+                                           Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_replace_context sc;
+    unsigned sz;
+    Eterm *hp;
+    Eterm lastkey;
+    Eterm end_condition;
+    Binary *mp;
+    Eterm key;
+    Eterm *tptr;
+    Eterm ereplaced;
+    Sint prev_replaced;
+
+
+#define RET_TO_BIF(Term, State) do { *ret = (Term); return State; } while(0);
+
+    /* Decode continuation. We know it's a tuple and everything else as
+       this is only called by ourselves */
+
+    /* continuation:
+       {Table, Lastkey, EndCondition, MatchProgBin, HowManyReplaced}*/
+
+    tptr = tuple_val(continuation);
+
+    if (arityval(*tptr) != 5)
+        erts_exit(ERTS_ERROR_EXIT,"Internal error in ets:select_replace/1");
+
+    lastkey = tptr[2];
+    end_condition = tptr[3];
+    mp = erts_db_get_match_prog_binary_unchecked(tptr[4]);
+
+    sc.p = p;
+    sc.mp = mp;
+    sc.end_condition = NIL;
+    sc.lastobj = NULL;
+    sc.max = 1000;
+    sc.keypos = tb->common.keypos;
+    if (is_big(tptr[5])) {
+        sc.replaced = big_to_uint32(tptr[5]);
+    } else {
+        sc.replaced = unsigned_val(tptr[5]);
+    }
+    prev_replaced = sc.replaced;
+
+    stack = get_any_stack(tb);
+    traverse_update_backwards(tb, stack, lastkey, &doit_select_replace, &sc);
+    release_stack(tb,stack);
+
+    // the more objects we've replaced, the more reductions we've consumed
+    BUMP_REDS(p, MIN(2000, (1000 - sc.max) + (sc.replaced - prev_replaced)));
+
+    if (sc.max > 0) {
+        RET_TO_BIF(erts_make_integer(sc.replaced,p), DB_ERROR_NONE);
+    }
+    key = GETKEY(tb, sc.lastobj);
+    if (end_condition != NIL &&
+            (cmp_partly_bound(end_condition,key) > 0)) {
+        /* done anyway */
+        RET_TO_BIF(make_small(sc.replaced),DB_ERROR_NONE);
+    }
+    /* Not done yet, let's trap. */
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.replaced)) {
+        hp = HAlloc(p, sz + 6);
+        ereplaced = make_small(sc.replaced);
+    }
+    else {
+        hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + 6);
+        ereplaced = uint_to_big(sc.replaced, hp);
+        hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    continuation = TUPLE5
+        (hp,
+         tptr[1],
+         key,
+         tptr[3],
+         tptr[4],
+         ereplaced);
+    RET_TO_BIF(bif_trap1(&ets_select_replace_continue_exp, p, continuation),
+            DB_ERROR_NONE);
+
+#undef RET_TO_BIF
+}
+
+static int db_select_replace_tree(Process *p, DbTable *tbl, Eterm tid,
+                                  Eterm pattern, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    DbTreeStack* stack;
+    struct select_replace_context sc;
+    struct mp_info mpi;
+    Eterm lastkey = THE_NON_VALUE;
+    Eterm key;
+    Eterm continuation;
+    unsigned sz;
+    Eterm *hp;
+    TreeDbTerm *this;
+    int errcode;
+    Eterm ereplaced;
+    Eterm mpb;
+
+
+#define RET_TO_BIF(Term,RetVal) do { 	       	\
+	if (mpi.mp != NULL) {			\
+	    erts_bin_free(mpi.mp);       	\
+	}					\
+	*ret = (Term); 				\
+	return RetVal; 			        \
+    } while(0)
+
+    mpi.mp = NULL;
+
+    sc.lastobj = NULL;
+    sc.p = p;
+    sc.tb = tb;
+    sc.max = 1000;
+    sc.end_condition = NIL;
+    sc.keypos = tb->common.keypos;
+    sc.replaced = 0;
+
+    if ((errcode = analyze_pattern(tb, pattern, db_match_keeps_key, &mpi)) != DB_ERROR_NONE) {
+        RET_TO_BIF(NIL,errcode);
+    }
+
+    if (!mpi.something_can_match) {
+        RET_TO_BIF(make_small(0),DB_ERROR_NONE);
+        /* can't possibly match anything */
+    }
+
+    sc.mp = mpi.mp;
+    sc.all_objects = mpi.all_objects;
+
+    stack = get_static_stack(tb);
+    if (!mpi.got_partial && mpi.some_limitation &&
+            CMP_EQ(mpi.least,mpi.most)) {
+        TreeDbTerm* term = *(mpi.save_term);
+        doit_select_replace(tb,mpi.save_term,&sc,0 /* dummy */);
+        if (stack != NULL) {
+            if (TOP_NODE(stack) == term)
+                // throw away potentially invalid reference
+                REPLACE_TOP_NODE(stack, *(mpi.save_term));
+            release_stack(tb, stack);
+        }
+        RET_TO_BIF(erts_make_integer(sc.replaced,p),DB_ERROR_NONE);
+    }
+
+    if (stack == NULL)
+        stack = get_any_stack(tb);
+
+    if (mpi.some_limitation) {
+        if ((this = find_next_from_pb_key(tb, stack, mpi.most)) != NULL) {
+            lastkey = GETKEY(tb, this->dbterm.tpl);
+        }
+        sc.end_condition = mpi.least;
+    }
+
+    traverse_update_backwards(tb, stack, lastkey, &doit_select_replace, &sc);
+    release_stack(tb,stack);
+    // the more objects we've replaced, the more reductions we've consumed
+    BUMP_REDS(p, MIN(2000, (1000 - sc.max) + sc.replaced));
+    if (sc.max > 0) {
+        RET_TO_BIF(erts_make_integer(sc.replaced,p),DB_ERROR_NONE);
+    }
+
+    key = GETKEY(tb, sc.lastobj);
+    sz = size_object(key);
+    if (IS_USMALL(0, sc.replaced)) {
+        hp = HAlloc(p, sz + ERTS_MAGIC_REF_THING_SIZE + 6);
+        ereplaced = make_small(sc.replaced);
+    }
+    else {
+        hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + ERTS_MAGIC_REF_THING_SIZE + 6);
+        ereplaced = uint_to_big(sc.replaced, hp);
+        hp += BIG_UINT_HEAP_SIZE;
+    }
+    key = copy_struct(key, sz, &hp, &MSO(p));
+    if (mpi.all_objects)
+        (mpi.mp)->flags |= BIN_FLAG_ALL_OBJECTS;
+    mpb = erts_db_make_match_prog_ref(p,mpi.mp,&hp);
+
+    continuation = TUPLE5
+        (hp,
+         tid,
+         key,
+         sc.end_condition, /* From the match program, needn't be copied */
+         mpb,
+         ereplaced);
+
+    /* Don't free mpi.mp, so don't use macro */
+    *ret = bif_trap1(&ets_select_replace_continue_exp, p, continuation);
+    return DB_ERROR_NONE;
+
+#undef RET_TO_BIF
+
+}
+
 static int db_take_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
 {
     DbTableTree *tb = &tbl->tree;
@@ -1947,8 +2196,9 @@ static TreeDbTerm *linkout_object_tree(DbTableTree *tb,
 ** For the select functions, analyzes the pattern and determines which
 ** part of the tree should be searched. Also compiles the match program
 */
-static int analyze_pattern(DbTableTree *tb, Eterm pattern, 
-			   struct mp_info *mpi)
+static int analyze_pattern(DbTableTree *tb, Eterm pattern,
+                           extra_match_validator_t extra_validator, /* Optional callback */
+                           struct mp_info *mpi)
 {
     Eterm lst, tpl, ttpl;
     Eterm *matches,*guards, *bodies;
@@ -1986,7 +2236,10 @@ static int analyze_pattern(DbTableTree *tb, Eterm pattern,
 
     i = 0;
     for(lst = pattern; is_list(lst); lst = CDR(list_val(lst))) {
-	Eterm body;
+        Eterm match;
+        Eterm guard;
+        Eterm body;
+
 	ttpl = CAR(list_val(lst));
 	if (!is_tuple(ttpl)) {
 	    if (buff != sbuff) { 
@@ -2001,9 +2254,17 @@ static int analyze_pattern(DbTableTree *tb, Eterm pattern,
 	    }
 	    return DB_ERROR_BADPARAM;
 	}
-	matches[i] = tpl = ptpl[1];
-	guards[i] = ptpl[2];
+	matches[i] = match = tpl = ptpl[1];
+	guards[i] = guard = ptpl[2];
 	bodies[i] = body = ptpl[3];
+
+        if(extra_validator != NULL && !extra_validator(tb->common.keypos, match, guard, body)) {
+	    if (buff != sbuff) {
+		erts_free(ERTS_ALC_T_DB_TMP, buff);
+	    }
+            return DB_ERROR_BADPARAM;
+        }
+
 	if (!is_list(body) || CDR(list_val(body)) != NIL ||
 	    CAR(list_val(body)) != am_DollarUnderscore) {
 	    mpi->all_objects = 0;
@@ -2011,7 +2272,7 @@ static int analyze_pattern(DbTableTree *tb, Eterm pattern,
 	++i;
 
 	partly_bound = NIL;
-	res = key_given(tb, tpl, &mpi->save_term, &partly_bound);
+	res = key_given(tb, tpl, &(mpi->save_term), &partly_bound);
 	if ( res >= 0 ) {   /* Can match something */
 	    key = 0;
 	    mpi->something_can_match = 1;
@@ -2514,6 +2775,58 @@ static TreeDbTerm **find_node2(DbTableTree *tb, Eterm key)
     return this;
 }
 
+/*
+ * Find node and return the address of the node pointer (NULL if not found)
+ * Tries to reuse the existing stack for performance.
+ */
+
+static TreeDbTerm **find_ptr(DbTableTree *tb, DbTreeStack *stack, TreeDbTerm *this) {
+    Eterm key = GETKEY(tb, this->dbterm.tpl);
+    TreeDbTerm *tmp;
+    TreeDbTerm *parent;
+    Sint c;
+
+    if(( tmp = TOP_NODE(stack)) != NULL) {
+	if (!cmp_key_eq(tb,key,tmp)) {
+	    /* Start from the beginning */
+	    stack->pos = stack->slot = 0;
+	}
+    }
+    if (EMPTY_NODE(stack)) { /* Have to rebuild the stack */
+	if (( tmp = tb->root ) == NULL)
+	    return NULL;
+	for (;;) {
+	    PUSH_NODE(stack, tmp);
+	    if (( c = cmp_key(tb,key,tmp) ) < 0) {
+		if (tmp->left == NULL) /* We are at the next
+					   and the element does
+					   not exist */
+		    break;
+		else
+		    tmp = tmp->left;
+	    } else if (c > 0) {
+		if (tmp->right == NULL) /* Done */
+		    return NULL;
+		else
+		    tmp = tmp->right;
+	    } else
+		break;
+	}
+    }
+
+    if (TOP_NODE(stack) != this)
+        return NULL;
+
+    parent = TOPN_NODE(stack, 1);
+    if (parent == NULL)
+        return ((this != tb->root) ? NULL : &(tb->root));
+    if (parent->left == this)
+        return &(parent->left);
+    if (parent->right == this)
+        return &(parent->right);
+    return NULL;
+}
+
 static int
 db_lookup_dbterm_tree(Process *p, DbTable *tbl, Eterm key, Eterm obj,
                       DbUpdateHandle* handle)
@@ -2655,13 +2968,60 @@ static void traverse_forward(DbTableTree *tb,
 }
 
 /*
+ * Traverse the tree with an update callback function, used by db_select_replace
+ */
+static void traverse_update_backwards(DbTableTree *tb,
+                                      DbTreeStack* stack,
+                                      Eterm lastkey,
+                                      int (*doit)(DbTableTree*,
+                                                  TreeDbTerm**,
+                                                  void*,
+                                                  int),
+                                      void* context)
+{
+    int res;
+    TreeDbTerm *this, *next, **this_ptr;
+
+    if (lastkey == THE_NON_VALUE) {
+        stack->pos = stack->slot = 0;
+        if (( this = tb->root ) == NULL) {
+            return;
+        }
+        while (this != NULL) {
+            PUSH_NODE(stack, this);
+            this = this->right;
+        }
+        this = TOP_NODE(stack);
+        this_ptr = find_ptr(tb, stack, this);
+        ASSERT(this_ptr != NULL);
+        res = (*doit)(tb, this_ptr, context, 0);
+        REPLACE_TOP_NODE(stack, *this_ptr);
+        next = find_prev(tb, stack, GETKEY(tb, (*this_ptr)->dbterm.tpl));
+        if (!res)
+            return;
+    } else {
+        next = find_prev(tb, stack, lastkey);
+    }
+
+    while ((this = next) != NULL) {
+        this_ptr = find_ptr(tb, stack, this);
+        ASSERT(this_ptr != NULL);
+        res = (*doit)(tb, this_ptr, context, 0);
+        REPLACE_TOP_NODE(stack, *this_ptr);
+        next = find_prev(tb, stack, GETKEY(tb, (*this_ptr)->dbterm.tpl));
+        if (!res)
+            return;
+    }
+}
+
+/*
  * Returns 0 if not given 1 if given and -1 on no possible match
  * if key is given; *ret is set to point to the object concerned.
  */
-static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm **ret, 
+static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm ***ret,
 		     Eterm *partly_bound)
 {
-    TreeDbTerm *this;
+    TreeDbTerm **this;
     Eterm key;
 
     ASSERT(ret != NULL);
@@ -2671,7 +3031,7 @@ static int key_given(DbTableTree *tb, Eterm pattern, TreeDbTerm **ret,
     if (is_non_value(key))
 	return -1;  /* can't possibly match anything */
     if (!db_has_variable(key)) {   /* Bound key */
-	if (( this = find_node(tb, key) ) == NULL) {
+	if (( this = find_node2(tb, key) ) == NULL) {
 	    return -1;
 	}
 	*ret = this;
@@ -3094,6 +3454,46 @@ static int doit_select_delete(DbTableTree *tb, TreeDbTerm *this, void *ptr,
     return 1;
 }
 
+static int doit_select_replace(DbTableTree *tb, TreeDbTerm **this, void *ptr,
+                               int forward)
+{
+    struct select_replace_context *sc = (struct select_replace_context *) ptr;
+    Eterm ret;
+
+    sc->lastobj = (*this)->dbterm.tpl;
+
+    /* Always backwards traversing */
+    if (sc->end_condition != NIL &&
+	(cmp_partly_bound(sc->end_condition,
+			  GETKEY_WITH_POS(sc->keypos, (*this)->dbterm.tpl)) > 0)) {
+	return 0;
+    }
+    ret = db_match_dbterm(&tb->common, sc->p, sc->mp, 0,
+			  &(*this)->dbterm, NULL, 0);
+
+    if (is_value(ret)) {
+        TreeDbTerm* new;
+        TreeDbTerm* old = *this;
+#ifdef DEBUG
+        Eterm key = db_getkey(tb->common.keypos, ret);
+        ASSERT(is_value(key));
+        ASSERT(cmp_key(tb, key, old) == 0);
+#endif
+        new = new_dbterm(tb, ret);
+        new->left = old->left;
+        new->right = old->right;
+        new->balance = old->balance;
+        sc->lastobj = new->dbterm.tpl;
+        *this = new;
+        free_term(tb, old);
+        ++(sc->replaced);
+    }
+    if (--(sc->max) <= 0) {
+	return 0;
+    }
+    return 1;
+}
+
 #ifdef TREE_DEBUG
 static void do_dump_tree2(DbTableTree* tb, int to, void *to_arg, int show,
 			  TreeDbTerm *t, int offset)
diff --git a/erts/emulator/beam/erl_db_util.c b/erts/emulator/beam/erl_db_util.c
index 6f30b1d3dd..03cc11bdc4 100644
--- a/erts/emulator/beam/erl_db_util.c
+++ b/erts/emulator/beam/erl_db_util.c
@@ -1119,6 +1119,177 @@ error:
     return NULL;
 }
 
+/*
+ * Compare a matching term 'a' with a constructing term 'b' for equality.
+ *
+ * Returns true if 'b' is guaranteed to always construct
+ * the same term as 'a' has matched.
+ */
+static int db_match_eq_body(Eterm a, Eterm b)
+{
+    DECLARE_ESTACK(s);
+    Uint arity;
+    Eterm *ap, *bp;
+    int const_mode = 0;
+    const Eterm CONST_MODE_OFF = THE_NON_VALUE;
+
+    while (1) {
+        switch(b & _TAG_PRIMARY_MASK) {
+        case TAG_PRIMARY_LIST:
+            if (!is_list(a))
+                return 0;
+            ESTACK_PUSH2(s, CDR(list_val(a)), CDR(list_val(b)));
+            a = CAR(list_val(a));
+            b = CAR(list_val(b));
+            continue; /* loop without pop */
+
+        case TAG_PRIMARY_BOXED:
+            if (is_tuple(b)) {
+                bp = tuple_val(b);
+                if (!const_mode) {
+                    if (bp[0] == make_arityval(1) && is_tuple(bp[1])) {
+                        b = bp[1]; /* double-tuple syntax */
+                    }
+                    else if (bp[0] == make_arityval(2) && bp[1] == am_const) {
+                        ESTACK_PUSH(s, CONST_MODE_OFF);
+                        const_mode = 1;   /* {const, term()} syntax */
+                        b = bp[2];
+                        continue; /* loop without pop */
+                    }
+                    else
+                        return 0; /* function call or invalid tuple syntax */
+                }
+                if (!is_tuple(a))
+                    return 0;
+
+                ap = tuple_val(a);
+                bp = tuple_val(b);
+                if (ap[0] != bp[0])
+                    return 0;
+                arity = arityval(ap[0]);
+                if (arity > 0) {
+                    a = *(++ap);
+                    b = *(++bp);
+                    while(--arity) {
+                        ESTACK_PUSH2(s, *(++ap), *(++bp));
+                    }
+                    continue; /* loop without pop */
+                }
+            }
+            else if (is_map(b)) {
+                /* We don't know what other pairs the matched map may contain */
+                return 0;
+            }
+            else if (!eq(a,b)) /* other boxed */
+                return 0;
+            break;
+
+        case TAG_PRIMARY_IMMED1:
+            if (a != b || a == am_Underscore || a == am_DollarDollar
+                || a == am_DollarUnderscore
+                || (const_mode && db_is_variable(a) >= 0)) {
+
+                return 0;
+            }
+            break;
+        default:
+            erts_exit(ERTS_ABORT_EXIT, "db_compare: "
+                      "Bad object on ESTACK: 0x%bex\n", b);
+        }
+
+pop_next:
+        if (ESTACK_ISEMPTY(s))
+            break; /* done */
+
+        b = ESTACK_POP(s);
+        if (b == CONST_MODE_OFF) {
+            ASSERT(const_mode);
+            const_mode = 0;
+            goto pop_next;
+        }
+        a = ESTACK_POP(s);
+    }
+
+    DESTROY_ESTACK(s);
+    return 1;
+}
+
+/* This is used by select_replace */
+int db_match_keeps_key(int keypos, Eterm match, Eterm guard, Eterm body)
+{
+    Eterm match_key;
+    Eterm* body_list;
+    Eterm single_body_term;
+    Eterm* single_body_term_tpl;
+    Eterm single_body_subterm;
+    Eterm single_body_subterm_key;
+    Eterm* single_body_subterm_key_tpl;
+
+    if (!is_list(body)) {
+        return 0;
+    }
+
+    body_list = list_val(body);
+    if (CDR(body_list) != NIL) {
+        return 0;
+    }
+
+    single_body_term = CAR(body_list);
+    if (single_body_term == am_DollarUnderscore) {
+        /* same tuple is returned */
+        return 1;
+    }
+
+    if (!is_tuple(single_body_term)) {
+        return 0;
+    }
+
+    single_body_term_tpl = tuple_val(single_body_term);
+    if (arityval(*single_body_term_tpl) != 1) {
+        // not the 1-element tuple we're expecting
+        return 0;
+    }
+
+    match_key = db_getkey(keypos, match);
+    if (!is_value(match_key)) {
+        // can't get key out of match
+        return 0;
+    }
+
+    single_body_subterm = single_body_term_tpl[1];
+    single_body_subterm_key = db_getkey(keypos, single_body_subterm);
+    if (!is_value(single_body_subterm_key)) {
+        // can't get key out of single body subterm
+        return 0;
+    }
+
+    if (db_match_eq_body(match_key, single_body_subterm_key)) {
+        /* tuple with same key is returned */
+        return 1;
+    }
+
+    if (!is_tuple(single_body_subterm_key)) {
+        /* can't possibly be an element instruction */
+        return 0;
+    }
+
+    single_body_subterm_key_tpl = tuple_val(single_body_subterm_key);
+    if (arityval(*single_body_subterm_key_tpl) != 3) {
+        /* can't possibly be an element instruction */
+        return 0;
+    }
+
+    if (single_body_subterm_key_tpl[1] == am_element &&
+        single_body_subterm_key_tpl[3] == am_DollarUnderscore &&
+        single_body_subterm_key_tpl[2] == make_small(keypos))
+    {
+        /* {element, KeyPos, '$_'} */
+        return 1;
+    }
+
+    return 0;
+}
+
 /* This is used when tracing */
 Eterm erts_match_set_lint(Process *p, Eterm matchexpr) {
     return db_match_set_lint(p, matchexpr, DCOMP_TRACE);
@@ -2172,7 +2343,7 @@ restart:
 		}
 	    }
 	    else {
-		*esp = term;
+		*esp++ = term;
 	    }
 	    break;
 	case matchPushArrayAsList:
@@ -5161,6 +5332,7 @@ void db_free_tmp_uncompressed(DbTerm* obj)
 Eterm db_match_dbterm(DbTableCommon* tb, Process* c_p, Binary* bprog,
 			     int all, DbTerm* obj, Eterm** hpp, Uint extra)
 {
+    enum erts_pam_run_flags flags;
     Uint32 dummy;
     Eterm res;
 
@@ -5168,9 +5340,13 @@ Eterm db_match_dbterm(DbTableCommon* tb, Process* c_p, Binary* bprog,
 	obj = db_alloc_tmp_uncompressed(tb, obj);
     }
 
+    flags = (hpp ?
+             ERTS_PAM_COPY_RESULT | ERTS_PAM_CONTIGUOUS_TUPLE :
+             ERTS_PAM_TMP_RESULT  | ERTS_PAM_CONTIGUOUS_TUPLE);
+
     res = db_prog_match(c_p, c_p,
                         bprog, make_tuple(obj->tpl), NULL, 0,
-			ERTS_PAM_COPY_RESULT|ERTS_PAM_CONTIGUOUS_TUPLE, &dummy);
+			flags, &dummy);
 
     if (is_value(res) && hpp!=NULL) {
 	*hpp = HAlloc(c_p, extra);
diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h
index 72298842d7..9be77fcefa 100644
--- a/erts/emulator/beam/erl_db_util.h
+++ b/erts/emulator/beam/erl_db_util.h
@@ -169,6 +169,15 @@ typedef struct db_table_method
 				    DbTable* tb, /* [in out] */ 
 				    Eterm continuation, 
 				    Eterm* ret);
+    int (*db_select_replace)(Process* p,
+            DbTable* tb, /* [in out] */
+            Eterm tid,
+            Eterm pattern,
+            Eterm* ret);
+    int (*db_select_replace_continue)(Process* p,
+            DbTable* tb, /* [in out] */
+            Eterm continuation,
+            Eterm* ret);
     int (*db_take)(Process *, DbTable *, Eterm, Eterm *);
 
     int (*db_delete_all_objects)(Process* p,
@@ -374,6 +383,7 @@ Eterm db_add_counter(Eterm** hpp, Wterm counter, Eterm incr);
 Eterm db_match_set_lint(Process *p, Eterm matchexpr, Uint flags);
 Binary *db_match_set_compile(Process *p, Eterm matchexpr, 
 			     Uint flags);
+int db_match_keeps_key(int keypos, Eterm match, Eterm guard, Eterm body);
 int erts_db_match_prog_destructor(Binary *);
 
 typedef struct match_prog {