Implement ets:take/2

This new ETS BIF returns and deletes objects from tables.

5 files changed, 120 insertions, 15 deletions

diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index fbdddf09db..fe25b2d27c 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -596,6 +596,8 @@ bif maps:values/1
 
 bif erts_internal:cmp_term/2
 
+bif ets:take/2
+
 #
 # Obsolete
 #
diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index 8f246ffa07..fba1e32230 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -753,6 +753,31 @@ BIF_RETTYPE ets_prev_2(BIF_ALIST_2)
     BIF_RET(ret);
 }
 
+/*
+** take(Tab, Key)
+*/
+BIF_RETTYPE ets_take_2(BIF_ALIST_2)
+{
+    DbTable* tb;
+#ifdef DEBUG
+    int cret;
+#endif
+    Eterm ret;
+    CHECK_TABLES();
+
+    tb = db_get_table(BIF_P, BIF_ARG_1, DB_WRITE, LCK_WRITE_REC);
+    if (!tb) {
+        BIF_ERROR(BIF_P, BADARG);
+    }
+#ifdef DEBUG
+    cret =
+#endif
+        tb->common.meth->db_take(BIF_P, tb, BIF_ARG_2, &ret);
+    ASSERT(cret == DB_ERROR_NONE);
+    db_unlock(tb, LCK_WRITE_REC);
+    BIF_RET(ret);
+}
+
 /* 
 ** update_element(Tab, Key, {Pos, Value})
 ** update_element(Tab, Key, [{Pos, Value}])
diff --git a/erts/emulator/beam/erl_db_hash.c b/erts/emulator/beam/erl_db_hash.c
index 06dac8f161..633a2272ad 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -426,6 +426,7 @@ static int db_select_count_continue_hash(Process *p, DbTable *tbl,
 
 static int db_select_delete_continue_hash(Process *p, DbTable *tbl,
 					  Eterm continuation, Eterm *ret);
+static int db_take_hash(Process *, DbTable *, Eterm, Eterm *);
 static void db_print_hash(int to,
 			  void *to_arg,
 			  int show,
@@ -536,6 +537,7 @@ DbTableMethod db_hash =
     db_select_delete_continue_hash,
     db_select_count_hash,
     db_select_count_continue_hash,
+    db_take_hash,
     db_delete_all_objects_hash,
     db_free_table_hash,
     db_free_table_continue_hash,
@@ -879,34 +881,45 @@ Ldone:
     return ret;
 }
 
+static Eterm
+get_term_list(Process *p, DbTableHash *tb, Eterm key, HashValue hval,
+              HashDbTerm *b1, HashDbTerm **bend)
+{
+    HashDbTerm* b2 = b1->next;
+    Eterm copy;
+
+    if (tb->common.status & (DB_BAG | DB_DUPLICATE_BAG)) {
+        while (b2 && has_key(tb, b2, key, hval)) {
+            b2 = b2->next;
+        }
+    }
+    copy = build_term_list(p, b1, b2, tb);
+    CHECK_TABLES();
+    if (bend) {
+        *bend = b2;
+    }
+    return copy;
+}
+
 int db_get_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
 {
     DbTableHash *tb = &tbl->hash;
     HashValue hval;
     int ix;
-    HashDbTerm* b1;
+    HashDbTerm* b;
     erts_smp_rwmtx_t* lck;
 
     hval = MAKE_HASH(key);
     lck = RLOCK_HASH(tb,hval);
     ix = hash_to_ix(tb, hval);
-    b1 = BUCKET(tb, ix);
+    b = BUCKET(tb, ix);
 
-    while(b1 != 0) {
-	if (has_live_key(tb,b1,key,hval)) {
-	    HashDbTerm* b2 = b1->next;
-	    Eterm copy;
-
-	    if (tb->common.status & (DB_BAG | DB_DUPLICATE_BAG)) {
-		while(b2 != NULL && has_key(tb,b2,key,hval))
-		    b2 = b2->next;
-	    }
-	    copy = build_term_list(p, b1, b2, tb);
-	    CHECK_TABLES();
-	    *ret = copy;
+    while(b != 0) {
+        if (has_live_key(tb, b, key, hval)) {
+            *ret = get_term_list(p, tb, key, hval, b, NULL);
 	    goto done;
 	}
-	b1 = b1->next;
+        b = b->next;
     }
     *ret = NIL;
 done:
@@ -2069,6 +2082,46 @@ trap:
 
 }
     
+static int db_take_hash(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableHash *tb = &tbl->hash;
+    HashDbTerm **bp, *b;
+    HashValue hval = MAKE_HASH(key);
+    erts_smp_rwmtx_t *lck = WLOCK_HASH(tb, hval);
+    int ix = hash_to_ix(tb, hval);
+    int nitems_diff = 0;
+
+    *ret = NIL;
+    for (bp = &BUCKET(tb, ix), b = *bp; b; bp = &b->next, b = b->next) {
+        if (has_live_key(tb, b, key, hval)) {
+            HashDbTerm *bend;
+
+            *ret = get_term_list(p, tb, key, hval, b, &bend);
+            while (b != bend) {
+                --nitems_diff;
+                if (nitems_diff == -1 && IS_FIXED(tb)) {
+                    /* Pseudo remove (no need to keep several of same key) */
+                    add_fixed_deletion(tb, ix);
+                    bp = &b->next;
+                    b->hvalue = INVALID_HASH;
+                    b = b->next;
+                } else {
+                    *bp = b->next;
+                    free_term(tb, b);
+                    b = *bp;
+                }
+            }
+            break;
+        }
+    }
+    WUNLOCK_HASH(lck);
+    if (nitems_diff) {
+        erts_smp_atomic_add_nob(&tb->common.nitems, nitems_diff);
+        try_shrink(tb);
+    }
+    return DB_ERROR_NONE;
+}
+
 /*
 ** 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 a62a83a928..720c0659c3 100644
--- a/erts/emulator/beam/erl_db_tree.c
+++ b/erts/emulator/beam/erl_db_tree.c
@@ -383,6 +383,7 @@ static int db_select_delete_tree(Process *p, DbTable *tbl,
 				 Eterm pattern,  Eterm *ret);
 static int db_select_delete_continue_tree(Process *p, DbTable *tbl, 
 					  Eterm continuation, Eterm *ret);
+static int db_take_tree(Process *, DbTable *, Eterm, Eterm *);
 static void db_print_tree(int to, void *to_arg,
 			  int show, DbTable *tbl);
 static int db_free_table_tree(DbTable *tbl);
@@ -431,6 +432,7 @@ DbTableMethod db_tree =
     db_select_delete_continue_tree,
     db_select_count_tree,
     db_select_count_continue_tree,
+    db_take_tree,
     db_delete_all_objects_tree,
     db_free_table_tree,
     db_free_table_continue_tree,
@@ -1722,6 +1724,28 @@ static int db_select_delete_tree(Process *p, DbTable *tbl,
 
 }
 
+static int db_take_tree(Process *p, DbTable *tbl, Eterm key, Eterm *ret)
+{
+    DbTableTree *tb = &tbl->tree;
+    TreeDbTerm *this;
+
+    *ret = NIL;
+    this = linkout_tree(tb, key, NULL);
+    if (this) {
+        Eterm copy, *hp, *hend;
+
+        hp = HAlloc(p, this->dbterm.size + 2);
+        hend = hp + this->dbterm.size + 2;
+        copy = db_copy_object_from_ets(&tb->common,
+                                       &this->dbterm, &hp, &MSO(p));
+        *ret = CONS(hp, copy, NIL);
+        hp += 2;
+        HRelease(p, hend, hp);
+        free_term(tb, this);
+    }
+    return DB_ERROR_NONE;
+}
+
 /*
 ** Other interface routines (not directly coupled to one bif)
 */
diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h
index 328b19dfc9..5ace93c8ed 100644
--- a/erts/emulator/beam/erl_db_util.h
+++ b/erts/emulator/beam/erl_db_util.h
@@ -165,6 +165,7 @@ typedef struct db_table_method
 				    DbTable* tb, /* [in out] */ 
 				    Eterm continuation, 
 				    Eterm* ret);
+    int (*db_take)(Process *, DbTable *, Eterm, Eterm *);
 
     int (*db_delete_all_objects)(Process* p,
 				 DbTable* db /* [in out] */ );