Deduplicate select* code on ETS hash tables

Refactor existing solution into a common iteration loop parameterized
using stateful callbacks.

2 files changed, 1010 insertions, 975 deletions

diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index 96d74bb050..1b5c8fe422 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -355,8 +355,6 @@ static ERTS_INLINE void SET_SEGTAB(DbTableHash* tb,
 	erts_smp_atomic_set_nob(&tb->segtab, (erts_aint_t) segtab);
 }
 
-
-
 /*
 ** Forward decl's (static functions)
 */
@@ -401,23 +399,21 @@ 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_replace_hash(Process *p, DbTable *tbl,
-                                  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 pattern, Eterm *ret);
 static int db_select_replace_continue_hash(Process *p, DbTable *tbl,
                                            Eterm continuation, Eterm *ret);
 
@@ -1151,816 +1147,1093 @@ static BIF_RETTYPE bif_trap1(Export *bif,
 {
     BIF_TRAP1(bif, p, p1);
 }
-    
+
+
+/*
+ * Match traversal callbacks
+ */
+
+/* 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);
+
 /*
- * Continue collecting select matches, this may happen either due to a trap
- * or when the user calls ets:select/1
+ * Begin hash table match traversal
  */
-static int db_select_continue_hash(Process *p, 
-				   DbTable *tbl, 
-				   Eterm continuation, 
-				   Eterm *ret)
+static int match_traverse(Process* p, DbTableHash* tb, Eterm pattern,
+                          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)
 {
-    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;
+    Sint slot_ix = -1;                /* Slot index */
+    HashDbTerm** current_ptr = NULL;  /* Refers to either the bucket pointer or
+                                       * the 'next' pointer in the previous term
+                                       */
+    HashDbTerm* saved_current = NULL; /* 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;
-    Eterm *tptr;
-    erts_smp_rwmtx_t* lck;
-
-#define RET_TO_BIF(Term, State) do { *ret = (Term); return State; } while(0);
+    Sint got = 0;                     /* Matched terms counter */
+    erts_smp_rwmtx_t* lck = NULL;     /* Slot lock */
+    int ret_value = DB_ERROR_NONE;
+#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);
 
-    /* Decode continuation. We know it's a tuple but not the arity or anything else */
+    *ret = NIL;
 
-    tptr = tuple_val(continuation);
+    if ((ret_value = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
+        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, slot_ix, 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;
-    Eterm match_res;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got = 0;
-    Eterm continuation;
-    int errcode;
-    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)
+    int all_objects = (*mpp)->flags & BIN_FLAG_ALL_OBJECTS;
+    HashDbTerm** current_ptr = NULL;  /* Refers to either the bucket pointer or
+                                       * the 'next' pointer in the previous term
+                                       */
+    HashDbTerm* saved_current = NULL; /* Helper to avoid double skip on match */
+    Eterm match_res = NIL;
+    erts_smp_rwmtx_t* lck = NULL;
+    int ret_value = DB_ERROR_NONE;
+#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);
 
+    *ret = NIL;
 
-    if ((errcode = analyze_pattern(tb, pattern, &mpi)) != DB_ERROR_NONE) {
-	RET_TO_BIF(NIL,errcode);
-    }
+    if (got < 0)
+        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_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);
 
-static int db_select_count_hash(Process *p, 
-				DbTable *tbl,
-                                Eterm tid,
-				Eterm pattern,
-				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 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)
+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
+}
 
-    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 */
-    }
+/*
+ * Common traversal trapping/continuation code;
+ * used by select_count, select_delete and select_replace,
+ * as well as their continuation-handling counterparts.
+ */
 
-    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;
-    }
+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)
+{
+    Eterm* hp = NULL;
+    Eterm egot = NIL;
+    Eterm mpb = NIL;
+    Eterm continuation = NIL;
+    int is_first_trap = (prev_continuation_tptr == NULL);
+    size_t base_halloc_sz = (is_first_trap ? PROC_BIN_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 = db_make_mp_binary(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 = NULL;
+    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_binary(tptr[3]) && thing_subtag(*binary_val(tptr[3])) == REFC_BINARY_SUBTAG)
+            || !(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 = ((ProcBin *) binary_val(tptr[3]))->val;
+    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 = NIL;
 
-    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 + PROC_BIN_SIZE);
+                mpb = db_make_mp_binary(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 = NIL;
+    Eterm continuation = NIL;
+    Eterm* hp = NULL;
+
+    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 + PROC_BIN_SIZE);
+        mpb = db_make_mp_binary(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 = {0};
+    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, 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 = NIL;
+    Eterm rest = NIL;
+    Eterm* hp = NULL;
+
+    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 */
+            rest = NIL;
+            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 = NULL;
+    Eterm tid = NIL;
+    Binary* mp = NULL;
+    Sint got = 0;
+    Sint slot_ix = 0;
+    Sint chunk_size = 0;
+    Eterm match_list = NIL;
+    Sint iterations_left = MAX_SELECT_CHUNK_ITERATIONS;
+    *ret = NIL;
 
-    
+    /* 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)
+        return DB_ERROR_BADPARAM;
 
-#undef RET_TO_BIF
+    if (!is_small(tptr[2]) || !is_small(tptr[3]) || !is_binary(tptr[4]) ||
+            !(is_list(tptr[5]) || tptr[5] == NIL) || !is_small(tptr[6]))
+        return DB_ERROR_BADPARAM;
+    if ((chunk_size = signed_val(tptr[3])) < 0)
+        return DB_ERROR_BADPARAM;
+    if (!(thing_subtag(*binary_val(tptr[4])) == REFC_BINARY_SUBTAG))
+        return DB_ERROR_BADPARAM;
 
+    mp = ((ProcBin *) binary_val(tptr[4]))->val;
+    if (!IsMatchProgBinary(mp))
+        return DB_ERROR_BADPARAM;
+
+    if ((got = signed_val(tptr[6])) < 0)
+        return DB_ERROR_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);
 }
-    
+
+#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);
-	}
+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, 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 = NULL;
+    Eterm tid = NIL;
+    Binary* mp = NULL;
+    Sint got = 0;
+    Sint slot_ix = 0;
+    Sint chunk_size = 0;
+    *ret = NIL;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        return DB_ERROR_BADPARAM;
     }
-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);
+
+    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 = NULL;
+    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, 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 = NULL;
+    Eterm tid = NIL;
+    Binary* mp = NULL;
+    Sint got = 0;
+    Sint slot_ix = 0;
+    Sint chunk_size = 0;
+    *ret = NIL;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        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;
+    Eterm key = NIL;
+    HashDbTerm* new = NULL;
+    HashDbTerm* next = NULL;
+    HashValue hval = INVALID_HASH;
+
+    if (is_value(match_res) &&
+            is_value(key = db_getkey(tb->common.keypos, match_res)) &&
+            eq(key, GETKEY(tb, (**current_ptr_ptr)->dbterm.tpl)))
+    {
+        next = (**current_ptr_ptr)->next;
+        hval = (**current_ptr_ptr)->hvalue;
+        new = replace_dbterm(tb, **current_ptr_ptr, match_res);
+        new->next = next;
+        new->hvalue = hval;
+        **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 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(!(tb->common.status & DB_BAG));
+
+    sr_context.p = p;
+    sr_context.tb = &tbl->hash;
+    sr_context.tid = NIL; // TODO
+    sr_context.hp = NULL;
+    sr_context.prev_continuation_tptr = NULL;
+
+    return match_traverse(
+            sr_context.p, sr_context.tb, pattern, 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 = NULL;
+    Eterm tid = NIL;
+    Binary* mp = NULL;
+    Sint got = 0;
+    Sint slot_ix = 0;
+    Sint chunk_size = 0;
+    *ret = NIL;
+
+    if (unpack_simple_mtraversal_continuation(continuation, &tptr, &tid, &slot_ix, &mp, &got)) {
+        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;
@@ -2001,245 +2274,6 @@ static int db_take_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
     return DB_ERROR_NONE;
 }
 
-static int db_select_replace_hash(Process *p,
-                                  DbTable *tbl,
-                                  Eterm pattern,
-                                  Eterm *ret)
-{
-    DbTableHash *tb = &tbl->hash;
-    struct mp_info mpi;
-    Uint slot_ix = 0;
-    HashDbTerm **current = NULL;
-    HashDbTerm *new = NULL, *next = NULL;
-    HashValue hval = INVALID_HASH;
-    unsigned current_list_pos = 0;
-    Eterm key;
-    Eterm match_res;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got = 0;
-    Eterm continuation;
-    int errcode;
-    Eterm mpb;
-    Eterm egot;
-    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)
-
-    /* Bag implementation presented both semantic consistency and performance issues */
-    ASSERT(!(tb->common.status & DB_BAG));
-
-    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                                            */
-        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));
-    }
-
-
-    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 {
-            match_res = db_match_dbterm(&tb->common, p, mpi.mp, 0,
-                                        &(*current)->dbterm, NULL, 0);
-            if (is_value(match_res) &&
-                is_value(key = db_getkey(tb->common.keypos, match_res)) &&
-                eq(key, GETKEY(tb, (*current)->dbterm.tpl)))
-            {
-                next = (*current)->next;
-                hval = (*current)->hvalue;
-                new = replace_dbterm(tb, *current, match_res);
-                new->next = next;
-                new->hvalue = hval;
-                *current = new;
-                ++got;
-            }
-            --num_left;
-            current = &((*current)->next);
-        }
-    }
-done:
-    // the more objects we've replaced, the more reductions we've consumed
-    BUMP_REDS(p, MIN(2000, (1000 - num_left) + (int)got));
-    RET_TO_BIF(erts_make_integer(got,p),DB_ERROR_NONE);
-trap:
-    BUMP_ALL_REDS(p);
-    if (IS_USMALL(0, got)) {
-        hp = HAlloc(p,  PROC_BIN_SIZE + 5);
-        egot = make_small(got);
-    }
-    else {
-        hp = HAlloc(p, BIG_UINT_HEAP_SIZE + PROC_BIN_SIZE + 5);
-        egot = uint_to_big(got, hp);
-        hp += BIG_UINT_HEAP_SIZE;
-    }
-    mpb = db_make_mp_binary(p,mpi.mp,&hp);
-    continuation = TUPLE4(hp, tb->common.id, make_small(slot_ix),
-            mpb,
-            egot);
-    mpi.mp = NULL; /*otherwise the return macro will destroy it */
-    RET_TO_BIF(bif_trap1(&ets_select_replace_continue_exp, p,
-                continuation),
-            DB_ERROR_NONE);
-
-#undef RET_TO_BIF
-
-}
-
-/*
-** This is called when select_replace traps
-*/
-static int db_select_replace_continue_hash(Process *p,
-					   DbTable *tbl,
-					   Eterm continuation,
-					   Eterm *ret)
-{
-    DbTableHash *tb = &tbl->hash;
-    Uint slot_ix;
-    HashDbTerm **current = NULL;
-    HashDbTerm *new = NULL, *next = NULL;
-    HashValue hval = INVALID_HASH;
-    Eterm key;
-    Eterm match_res;
-    Eterm *hp;
-    int num_left = 1000;
-    Uint got, prev_got;
-    Eterm *tptr;
-    Binary *mp;
-    Eterm egot;
-    erts_smp_rwmtx_t* lck;
-
-#define RET_TO_BIF(Term,RetVal) do {		\
-	*ret = (Term);				\
-	return RetVal;				\
-    } while(0)
-
-
-    tptr = tuple_val(continuation);
-    slot_ix = unsigned_val(tptr[2]);
-    mp = ((ProcBin *) binary_val(tptr[3]))->val;
-    if (is_big(tptr[4])) {
-        got = big_to_uint32(tptr[4]);
-    } else {
-        got = unsigned_val(tptr[4]);
-    }
-    prev_got = got;
-
-    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 {
-            match_res = db_match_dbterm(&tb->common, p, mp, 0,
-                                        &(*current)->dbterm, NULL, 0);
-            if (is_value(match_res) &&
-                is_value(key = db_getkey(tb->common.keypos, match_res)) &&
-                eq(key, GETKEY(tb, (*current)->dbterm.tpl)))
-            {
-                next = (*current)->next;
-                hval = (*current)->hvalue;
-                new = replace_dbterm(tb, *current, match_res);
-                new->next = next;
-                new->hvalue = hval;
-                *current = new;
-                ++got;
-            }
-            --num_left;
-            current = &((*current)->next);
-        }
-    }
-done:
-    // the more objects we've replaced, the more reductions we've consumed
-    BUMP_REDS(p, MIN(2000, (1000 - num_left) + (int)(got - prev_got)));
-    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, tb->common.id, make_small(slot_ix),
-            tptr[3],
-            egot);
-    RET_TO_BIF(bif_trap1(&ets_select_replace_continue_exp, p,
-                continuation),
-            DB_ERROR_NONE);
-
-#undef RET_TO_BIF
-
-}
 
 /*
 ** Other interface routines (not directly coupled to one bif)
diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
index af1296367a..83cbf059c5 100644
--- a/erts/emulator/beam/erl_db_tree.c
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -1843,6 +1843,7 @@ static int db_select_replace_tree(Process *p, DbTable *tbl,
                                   Eterm pattern, Eterm *ret)
 {
     DbTableTree *tb = &tbl->tree;
+    Eterm tid = NIL; // TODO
     DbTreeStack* stack;
     struct select_replace_context sc;
     struct mp_info mpi;
@@ -1937,7 +1938,7 @@ static int db_select_replace_tree(Process *p, DbTable *tbl,
 
     continuation = TUPLE5
         (hp,
-         tb->common.id,
+         tid,
          key,
          sc.end_condition, /* From the match program, needn't be copied */
          mpb,