Merge branch 'sverk/ets-take-2/OTP-12309'

* sverk/ets-take-2/OTP-12309:
  erts: Optimize ets:lookup and ets:take for bags
  Implement ets:take/2

5 files changed, 135 insertions, 26 deletions

diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab
index 55ac778475..aea3224342 100644
--- a/erts/emulator/beam/bif.tab
+++ b/erts/emulator/beam/bif.tab
@@ -600,6 +600,8 @@ bif maps:values/1
 
 bif erts_internal:cmp_term/2
 
+bif ets:take/2
+
 #
 # New in 17.1
 #
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..6e8d892561 100644
--- a/erts/emulator/beam/erl_db_hash.c
+++ b/erts/emulator/beam/erl_db_hash.c
@@ -382,7 +382,7 @@ static HashDbTerm* search_list(DbTableHash* tb, Eterm key,
 static void shrink(DbTableHash* tb, int nactive);
 static void grow(DbTableHash* tb, int nactive);
 static Eterm build_term_list(Process* p, HashDbTerm* ptr1, HashDbTerm* ptr2,
-			   DbTableHash*);
+			   Uint sz, DbTableHash*);
 static int analyze_pattern(DbTableHash *tb, Eterm pattern, 
 			   struct mp_info *mpi);
 
@@ -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,49 @@ 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;
+    Uint sz = b1->dbterm.size + 2;
+
+    if (tb->common.status & (DB_BAG | DB_DUPLICATE_BAG)) {
+        while (b2 && has_key(tb, b2, key, hval)) {
+	    if (b2->hvalue != INVALID_HASH)
+		sz += b2->dbterm.size + 2;
+
+            b2 = b2->next;
+        }
+    }
+    copy = build_term_list(p, b1, b2, sz, 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);
-
-    while(b1 != 0) {
-	if (has_live_key(tb,b1,key,hval)) {
-	    HashDbTerm* b2 = b1->next;
-	    Eterm copy;
+    b = BUCKET(tb, ix);
 
-	    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:
@@ -1240,7 +1257,7 @@ static int db_slot_hash(Process *p, DbTable *tbl, Eterm slot_term, Eterm *ret)
     lck = RLOCK_HASH(tb, slot);
     nactive = NACTIVE(tb);
     if (slot < nactive) {
-	*ret = build_term_list(p, BUCKET(tb, slot), 0, tb);
+	*ret = build_term_list(p, BUCKET(tb, slot), NULL, 0, tb);
 	retval = DB_ERROR_NONE;
     }
     else if (slot == nactive) {
@@ -2069,6 +2086,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)
 */
@@ -2483,23 +2540,23 @@ static int free_seg(DbTableHash *tb, int free_records)
 ** Copy terms from ptr1 until ptr2
 ** works for ptr1 == ptr2 == 0  => []
 ** or ptr2 == 0
+** sz is either precalculated heap size or 0 if not known
 */
 static Eterm build_term_list(Process* p, HashDbTerm* ptr1, HashDbTerm* ptr2,
-			   DbTableHash* tb)
+			     Uint sz, DbTableHash* tb)
 {
-    int sz = 0;
     HashDbTerm* ptr;
     Eterm list = NIL;
     Eterm copy;
     Eterm *hp, *hend;
 
-    ptr = ptr1;
-    while(ptr != ptr2) {
-
-	if (ptr->hvalue != INVALID_HASH)
-	    sz += ptr->dbterm.size + 2;
-
-	ptr = ptr->next;
+    if (!sz) {
+	ptr = ptr1;
+	while(ptr != ptr2) {
+	    if (ptr->hvalue != INVALID_HASH)
+		sz += ptr->dbterm.size + 2;
+	    ptr = ptr->next;
+	}
     }
 
     hp = HAlloc(p, sz);
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] */ );