ETS: Allow for matchspec-based replacement

1 files changed, 393 insertions, 14 deletions

diff --git a/erts/emulator/beam/erl_db_tree.c b/erts/emulator/beam/erl_db_tree.c
index 14a7451267..af1296367a 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,21 @@ 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;
+};
+
+/*
 ** Forward declarations 
 */
 static TreeDbTerm *linkout_tree(DbTableTree *tb, Eterm key);
@@ -290,6 +314,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,8 +336,16 @@ 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);
@@ -335,6 +368,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 +420,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 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 +476,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,
@@ -1103,7 +1146,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);
     }
 
@@ -1306,7 +1349,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);
     }
 
@@ -1409,7 +1452,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);
@@ -1650,7 +1693,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 +1745,212 @@ 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];
+    if (!(thing_subtag(*binary_val(tptr[4])) == REFC_BINARY_SUBTAG))
+        RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+    mp = ((ProcBin *) binary_val(tptr[4]))->val;
+    if (!IsMatchProgBinary(mp))
+        RET_TO_BIF(NIL,DB_ERROR_BADPARAM);
+
+    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 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, &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 + PROC_BIN_SIZE + 6);
+        ereplaced = make_small(sc.replaced);
+    }
+    else {
+        hp = HAlloc(p, BIG_UINT_HEAP_SIZE + sz + PROC_BIN_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 = db_make_mp_binary(p,mpi.mp,&hp);
+
+    continuation = TUPLE5
+        (hp,
+         tb->common.id,
+         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;
@@ -2011,7 +2260,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 +2763,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 +2956,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 +3019,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 +3442,37 @@ 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, key;
+
+    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) &&
+        is_value(key = db_getkey(tb->common.keypos, ret)) &&
+        (cmp_key(tb, key, *this) == 0))
+    {
+        *this = replace_dbterm(tb, *this, ret);
+        sc->lastobj = (*this)->dbterm.tpl;
+        ++(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)