1 files changed, 111 insertions, 64 deletions

diff --git a/erts/emulator/beam/erl_db.c b/erts/emulator/beam/erl_db.c
index c009a3bde8..0a50af4d1a 100644
--- a/erts/emulator/beam/erl_db.c
+++ b/erts/emulator/beam/erl_db.c
@@ -81,16 +81,9 @@ static BIF_RETTYPE db_bif_fail(Process* p, Uint freason,
 /* Get a key from any table structure and a tagged object */
 #define TERM_GETKEY(tb, obj) db_getkey((tb)->common.keypos, (obj)) 
 
- 
-/* How safe are we from double-hits or missed objects
-** when iterating without fixation? */ 
-enum DbIterSafety {
-    ITER_UNSAFE,      /* Must fixate to be safe */
-    ITER_SAFE_LOCKED, /* Safe while table is locked, not between trap calls */
-    ITER_SAFE         /* No need to fixate at all */
-};
 #  define ITERATION_SAFETY(Proc,Tab) \
-    ((IS_TREE_TABLE((Tab)->common.status) || ONLY_WRITER(Proc,Tab)) ? ITER_SAFE \
+    ((IS_TREE_TABLE((Tab)->common.status) || IS_CATREE_TABLE((Tab)->common.status) \
+      || ONLY_WRITER(Proc,Tab)) ? ITER_SAFE                             \
      : (((Tab)->common.status & DB_FINE_LOCKED) ? ITER_UNSAFE : ITER_SAFE_LOCKED))
 
 #define DID_TRAP(P,Ret) (!is_value(Ret) && ((P)->freason == TRAP))
@@ -194,9 +187,6 @@ static int fixed_tabs_find(DbFixation* first, DbFixation* fix)
 #define ERTS_RBT_WANT_DELETE
 #define ERTS_RBT_WANT_FOREACH
 #define ERTS_RBT_WANT_FOREACH_DESTROY
-#ifdef DEBUG
-# define ERTS_RBT_WANT_LOOKUP
-#endif
 #define ERTS_RBT_UNDEF
 
 #include "erl_rbtree.h"
@@ -359,6 +349,7 @@ typedef enum {
 
 extern DbTableMethod db_hash;
 extern DbTableMethod db_tree;
+extern DbTableMethod db_catree;
 
 int user_requested_db_max_tabs;
 int erts_ets_realloc_always_moves;
@@ -407,21 +398,17 @@ static void
 free_dbtable(void *vtb)
 {
     DbTable *tb = (DbTable *) vtb;
-#ifdef HARDDEBUG
-	if (erts_atomic_read_nob(&tb->common.memory_size) != sizeof(DbTable)) {
-	    erts_fprintf(stderr, "ets: free_dbtable memory remain=%ld fix=%x\n",
-			 erts_atomic_read_nob(&tb->common.memory_size)-sizeof(DbTable),
-			 tb->common.fixations);
-	}
-#endif
-	erts_rwmtx_destroy(&tb->common.rwlock);
-	erts_mtx_destroy(&tb->common.fixlock);
-	ASSERT(is_immed(tb->common.heir_data));
 
-        if (tb->common.btid)
-            erts_bin_release(tb->common.btid);
+    ASSERT(erts_atomic_read_nob(&tb->common.memory_size) == sizeof(DbTable));
+
+    erts_rwmtx_destroy(&tb->common.rwlock);
+    erts_mtx_destroy(&tb->common.fixlock);
+    ASSERT(is_immed(tb->common.heir_data));
 
-	erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
+    if (tb->common.btid)
+        erts_bin_release(tb->common.btid);
+
+    erts_db_free(ERTS_ALC_T_DB_TABLE, tb, (void *) tb, sizeof(DbTable));
 }
 
 static void schedule_free_dbtable(DbTable* tb)
@@ -1076,7 +1063,7 @@ BIF_RETTYPE ets_update_element_3(BIF_ALIST_3)
     DB_BIF_GET_TABLE(tb, DB_WRITE, LCK_WRITE_REC, BIF_ets_update_element_3);
 
     UseTmpHeap(2,BIF_P);
-    if (!(tb->common.status & (DB_SET | DB_ORDERED_SET))) {
+    if (!(tb->common.status & (DB_SET | DB_ORDERED_SET | DB_CA_ORDERED_SET))) {
 	goto bail_out;
     }
     if (is_tuple(BIF_ARG_3)) {
@@ -1165,7 +1152,7 @@ do_update_counter(Process *p, DbTable* tb,
 
     UseTmpHeap(5, p);
 
-    if (!(tb->common.status & (DB_SET | DB_ORDERED_SET))) {
+    if (!(tb->common.status & (DB_SET | DB_ORDERED_SET | DB_CA_ORDERED_SET))) {
 	goto bail_out;
     }
     if (is_integer(arg3)) { /* Incr */
@@ -1647,15 +1634,15 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
 	val = CAR(list_val(list));
 	if (val == am_bag) {
 	    status |= DB_BAG;
-	    status &= ~(DB_SET | DB_DUPLICATE_BAG | DB_ORDERED_SET);
+	    status &= ~(DB_SET | DB_DUPLICATE_BAG | DB_ORDERED_SET | DB_CA_ORDERED_SET);
 	}
 	else if (val == am_duplicate_bag) {
 	    status |= DB_DUPLICATE_BAG;
-	    status &= ~(DB_SET | DB_BAG | DB_ORDERED_SET);
+	    status &= ~(DB_SET | DB_BAG | DB_ORDERED_SET | DB_CA_ORDERED_SET);
 	}
 	else if (val == am_ordered_set) {
 	    status |= DB_ORDERED_SET;
-	    status &= ~(DB_SET | DB_BAG | DB_DUPLICATE_BAG);
+	    status &= ~(DB_SET | DB_BAG | DB_DUPLICATE_BAG | DB_CA_ORDERED_SET);
 	}
 	else if (is_tuple(val)) {
 	    Eterm *tp = tuple_val(val);
@@ -1716,7 +1703,13 @@ BIF_RETTYPE ets_new_2(BIF_ALIST_2)
     if (is_not_nil(list)) { /* bad opt or not a well formed list */
 	BIF_ERROR(BIF_P, BADARG);
     }
-    if (IS_HASH_TABLE(status)) {
+    if (IS_TREE_TABLE(status) && is_fine_locked && !(status & DB_PRIVATE)) {
+        meth = &db_catree;
+        status |= DB_CA_ORDERED_SET;
+        status &= ~(DB_SET | DB_BAG | DB_DUPLICATE_BAG | DB_ORDERED_SET);
+        status |= DB_FINE_LOCKED;
+    }
+    else if (IS_HASH_TABLE(status)) {
 	meth = &db_hash;
 	if (is_fine_locked && !(status & DB_PRIVATE)) {
 	    status |= DB_FINE_LOCKED;
@@ -2283,6 +2276,7 @@ static BIF_RETTYPE ets_select_delete_trap_1(BIF_ALIST_1)
     Eterm ret;
     Eterm *tptr;
     db_lock_kind_t kind = LCK_WRITE_REC;
+    enum DbIterSafety safety = ITER_SAFE;
 
     CHECK_TABLES();
     ASSERT(is_tuple(a1));
@@ -2292,10 +2286,11 @@ static BIF_RETTYPE ets_select_delete_trap_1(BIF_ALIST_1)
     DB_TRAP_GET_TABLE(tb, tptr[1], DB_WRITE, kind,
                       &ets_select_delete_continue_exp);
 
-    cret = tb->common.meth->db_select_delete_continue(p,tb,a1,&ret);
+    cret = tb->common.meth->db_select_delete_continue(p,tb,a1,&ret,&safety);
 
-    if(!DID_TRAP(p,ret) && ITERATION_SAFETY(p,tb) != ITER_SAFE) {  
-	unfix_table_locked(p, tb, &kind);
+    if(!DID_TRAP(p,ret) && safety != ITER_SAFE) {
+        ASSERT(erts_refc_read(&tb->common.fix_count,1));
+        unfix_table_locked(p, tb, &kind);
     }
 
     db_unlock(tb, kind);
@@ -2333,7 +2328,8 @@ BIF_RETTYPE ets_internal_select_delete_2(BIF_ALIST_2)
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
     }
-    cret = tb->common.meth->db_select_delete(BIF_P, tb, BIF_ARG_1, BIF_ARG_2, &ret);
+    cret = tb->common.meth->db_select_delete(BIF_P, tb, BIF_ARG_1, BIF_ARG_2,
+                                             &ret, safety);
 
     if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
 	fix_table_locked(BIF_P,tb);
@@ -2725,7 +2721,7 @@ ets_select3(Process* p, DbTable* tb, Eterm tid, Eterm ms, Sint chunk_size)
     cret = tb->common.meth->db_select_chunk(p, tb, tid,
 					    ms, chunk_size,
 					    0 /* not reversed */,
-					    &ret);
+					    &ret, safety);
     if (DID_TRAP(p,ret) && safety != ITER_SAFE) {
 	fix_table_locked(p, tb);
     }
@@ -2752,7 +2748,8 @@ ets_select3(Process* p, DbTable* tb, Eterm tid, Eterm ms, Sint chunk_size)
 }
 
 
-/* We get here instead of in the real BIF when trapping */
+/* Trap here from: ets_select_1/2/3
+ */
 static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
 {
     Process *p = BIF_P;
@@ -2763,6 +2760,7 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
     Eterm ret;
     Eterm *tptr;
     db_lock_kind_t kind = LCK_READ;
+    enum DbIterSafety safety = ITER_SAFE;
 
     CHECK_TABLES();
 
@@ -2772,11 +2770,13 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
     DB_TRAP_GET_TABLE(tb, tptr[1], DB_READ, kind,
                       &ets_select_continue_exp);
 
-    cret = tb->common.meth->db_select_continue(p, tb, a1,
-					       &ret);
+    cret = tb->common.meth->db_select_continue(p, tb, a1, &ret, &safety);
 
-    if (!DID_TRAP(p,ret) && ITERATION_SAFETY(p,tb) != ITER_SAFE) {
-	unfix_table_locked(p, tb, &kind);
+    if (!DID_TRAP(p,ret)) {
+        if (safety != ITER_SAFE) {
+            ASSERT(erts_refc_read(&tb->common.fix_count,1));
+            unfix_table_locked(p, tb, &kind);
+        }
     }
     db_unlock(tb, kind);
 
@@ -2801,8 +2801,12 @@ static BIF_RETTYPE ets_select_trap_1(BIF_ALIST_1)
 BIF_RETTYPE ets_select_1(BIF_ALIST_1)
 {
     return ets_select1(BIF_P, BIF_ets_select_1, BIF_ARG_1);
+    /* TRAP: ets_select_trap_1 */
 }
 
+/*
+ * Common impl for select/1, select_reverse/1, match/1 and match_object/1
+ */
 static BIF_RETTYPE ets_select1(Process *p, int bif_ix, Eterm arg1)
 {
     BIF_RETTYPE result;
@@ -2810,7 +2814,7 @@ static BIF_RETTYPE ets_select1(Process *p, int bif_ix, Eterm arg1)
     int cret;
     Eterm ret;
     Eterm *tptr;
-    enum DbIterSafety safety;
+    enum DbIterSafety safety, safety_copy;
 
     CHECK_TABLES();
 
@@ -2835,7 +2839,8 @@ static BIF_RETTYPE ets_select1(Process *p, int bif_ix, Eterm arg1)
 	local_fix_table(tb);
     }
 
-    cret = tb->common.meth->db_select_continue(p,tb, arg1, &ret);
+    safety_copy = safety;
+    cret = tb->common.meth->db_select_continue(p,tb, arg1, &ret, &safety_copy);
 
     if (DID_TRAP(p,ret) && safety != ITER_SAFE) {
 	fix_table_locked(p, tb);
@@ -2867,6 +2872,7 @@ BIF_RETTYPE ets_select_2(BIF_ALIST_2)
     DbTable* tb;
     DB_BIF_GET_TABLE(tb, DB_READ, LCK_READ, BIF_ets_select_2);
     return ets_select2(BIF_P, tb, BIF_ARG_1, BIF_ARG_2);
+    /* TRAP: ets_select_trap_1 */
 }
 
 static BIF_RETTYPE
@@ -2884,7 +2890,7 @@ ets_select2(Process* p, DbTable* tb, Eterm tid, Eterm ms)
 	local_fix_table(tb);
     }
 
-    cret = tb->common.meth->db_select(p, tb, tid, ms, 0, &ret);
+    cret = tb->common.meth->db_select(p, tb, tid, ms, 0, &ret, safety);
 
     if (DID_TRAP(p,ret) && safety != ITER_SAFE) {
 	fix_table_locked(p, tb);
@@ -2922,6 +2928,7 @@ static BIF_RETTYPE ets_select_count_1(BIF_ALIST_1)
     Eterm ret;
     Eterm *tptr;
     db_lock_kind_t kind = LCK_READ;
+    enum DbIterSafety safety = ITER_SAFE;
 
     CHECK_TABLES();
 
@@ -2931,9 +2938,10 @@ static BIF_RETTYPE ets_select_count_1(BIF_ALIST_1)
     DB_TRAP_GET_TABLE(tb, tptr[1], DB_READ, kind,
                       &ets_select_count_continue_exp);
 
-    cret = tb->common.meth->db_select_count_continue(p, tb, a1, &ret);
+    cret = tb->common.meth->db_select_count_continue(p, tb, a1, &ret, &safety);
 
-    if (!DID_TRAP(p,ret) && ITERATION_SAFETY(p,tb) != ITER_SAFE) {
+    if (!DID_TRAP(p,ret) && safety != ITER_SAFE) {
+        ASSERT(erts_refc_read(&tb->common.fix_count,1));
 	unfix_table_locked(p, tb, &kind);
     }
     db_unlock(tb, kind);
@@ -2971,7 +2979,8 @@ BIF_RETTYPE ets_select_count_2(BIF_ALIST_2)
     if (safety == ITER_UNSAFE) {
 	local_fix_table(tb);
     }
-    cret = tb->common.meth->db_select_count(BIF_P,tb, BIF_ARG_1, BIF_ARG_2, &ret);
+    cret = tb->common.meth->db_select_count(BIF_P,tb, BIF_ARG_1, BIF_ARG_2,
+                                            &ret, safety);
 
     if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
 	fix_table_locked(BIF_P, tb);
@@ -3010,6 +3019,7 @@ static BIF_RETTYPE ets_select_replace_1(BIF_ALIST_1)
     Eterm ret;
     Eterm *tptr;
     db_lock_kind_t kind = LCK_WRITE_REC;
+    enum DbIterSafety safety = ITER_SAFE;
 
     CHECK_TABLES();
     ASSERT(is_tuple(a1));
@@ -3019,9 +3029,10 @@ static BIF_RETTYPE ets_select_replace_1(BIF_ALIST_1)
     DB_TRAP_GET_TABLE(tb, tptr[1], DB_WRITE, kind,
                       &ets_select_replace_continue_exp);
 
-    cret = tb->common.meth->db_select_replace_continue(p,tb,a1,&ret);
+    cret = tb->common.meth->db_select_replace_continue(p,tb,a1,&ret,&safety);
 
-    if(!DID_TRAP(p,ret) && ITERATION_SAFETY(p,tb) != ITER_SAFE) {
+    if(!DID_TRAP(p,ret) && safety != ITER_SAFE) {
+        ASSERT(erts_refc_read(&tb->common.fix_count,1));
         unfix_table_locked(p, tb, &kind);
     }
 
@@ -3064,7 +3075,8 @@ BIF_RETTYPE ets_select_replace_2(BIF_ALIST_2)
     if (safety == ITER_UNSAFE) {
         local_fix_table(tb);
     }
-    cret = tb->common.meth->db_select_replace(BIF_P, tb, BIF_ARG_1, BIF_ARG_2, &ret);
+    cret = tb->common.meth->db_select_replace(BIF_P, tb, BIF_ARG_1, BIF_ARG_2,
+                                              &ret, safety);
 
     if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
         fix_table_locked(BIF_P,tb);
@@ -3116,7 +3128,7 @@ BIF_RETTYPE ets_select_reverse_3(BIF_ALIST_3)
     }
     cret = tb->common.meth->db_select_chunk(BIF_P,tb, BIF_ARG_1,
 					    BIF_ARG_2, chunk_size, 
-					    1 /* reversed */, &ret);
+					    1 /* reversed */, &ret, safety);
     if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
 	fix_table_locked(BIF_P, tb);
     }
@@ -3161,7 +3173,7 @@ BIF_RETTYPE ets_select_reverse_2(BIF_ALIST_2)
 	local_fix_table(tb);
     }
     cret = tb->common.meth->db_select(BIF_P,tb, BIF_ARG_1, BIF_ARG_2,
-				      1 /*reversed*/, &ret);
+				      1 /*reversed*/, &ret, safety);
 
     if (DID_TRAP(BIF_P,ret) && safety != ITER_SAFE) {
 	fix_table_locked(BIF_P, tb);
@@ -3506,6 +3518,7 @@ void init_db(ErtsDbSpinCount db_spin_count)
 
     db_initialize_hash();
     db_initialize_tree();
+    db_initialize_catree();
 
     /* Non visual BIF to trap to. */
     erts_init_trap_export(&ets_select_delete_continue_exp,
@@ -3696,7 +3709,7 @@ static SWord proc_cleanup_fixed_table(Process* p, DbFixation* fix)
 /*
  * erts_db_process_exiting() is called when a process terminates.
  * It returns 0 when completely done, and !0 when it wants to
- * yield. c_p->u.terminate can hold a pointer to a state while
+ * yield. *yield_state can hold a pointer to a state while
  * yielding.
  */
 #define ERTS_DB_INTERNAL_ERROR(LSTR) \
@@ -3704,7 +3717,7 @@ static SWord proc_cleanup_fixed_table(Process* p, DbFixation* fix)
 	   __FILE__, __LINE__)
 
 int
-erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
+erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks, void **yield_state)
 {
     typedef struct {
         enum {
@@ -3714,7 +3727,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
         }op;
         DbTable *tb;
     } CleanupState;
-    CleanupState *state = (CleanupState *) c_p->u.terminate;
+    CleanupState *state = (CleanupState *) *yield_state;
     Eterm pid = c_p->common.id;
     CleanupState default_state;
     SWord initial_reds = ERTS_BIF_REDS_LEFT(c_p);
@@ -3787,7 +3800,7 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
 
                 if (state != &default_state)
                     erts_free(ERTS_ALC_T_DB_PROC_CLEANUP, state);
-                c_p->u.terminate = NULL;
+                *yield_state = NULL;
 
                 BUMP_REDS(c_p, (initial_reds - reds));
                 return 0;
@@ -3807,12 +3820,12 @@ erts_db_process_exiting(Process *c_p, ErtsProcLocks c_p_locks)
  yield:
 
     if (state == &default_state) {
-	c_p->u.terminate = erts_alloc(ERTS_ALC_T_DB_PROC_CLEANUP,
-				      sizeof(CleanupState));
-	sys_memcpy(c_p->u.terminate, (void*) state, sizeof(CleanupState));
+	*yield_state = erts_alloc(ERTS_ALC_T_DB_PROC_CLEANUP,
+                                  sizeof(CleanupState));
+	sys_memcpy(*yield_state, (void*) state, sizeof(CleanupState));
     }
     else
-        ASSERT(state == c_p->u.terminate);
+        ASSERT(state == *yield_state);
 
     return !0;
 }
@@ -3907,7 +3920,7 @@ struct free_fixations_ctx
     SWord cnt;
 };
 
-static void free_fixations_op(DbFixation* fix, void* vctx)
+static int free_fixations_op(DbFixation* fix, void* vctx, Sint reds)
 {
     struct free_fixations_ctx* ctx = (struct free_fixations_ctx*) vctx;
     erts_aint_t diff;
@@ -3944,6 +3957,7 @@ static void free_fixations_op(DbFixation* fix, void* vctx)
         ERTS_ETS_MISC_MEM_ADD(-sizeof(DbFixation));
     }
     ctx->cnt++;
+    return 1;
 }
 
 int erts_db_execute_free_fixation(Process* p, DbFixation* fix)
@@ -4088,7 +4102,7 @@ struct fixing_procs_info_ctx
     Eterm list;
 };
 
-static void fixing_procs_info_op(DbFixation* fix, void* vctx)
+static int fixing_procs_info_op(DbFixation* fix, void* vctx, Sint reds)
 {
     struct fixing_procs_info_ctx* ctx = (struct fixing_procs_info_ctx*) vctx;
     Eterm* hp;
@@ -4098,6 +4112,7 @@ static void fixing_procs_info_op(DbFixation* fix, void* vctx)
     tpl = TUPLE2(hp, fix->procs.p->common.id, make_small(fix->counter));
     hp += 3;
     ctx->list = CONS(hp, tpl, ctx->list);
+    return 1;
 }
 
 static Eterm table_info(Process* p, DbTable* tb, Eterm What)
@@ -4114,6 +4129,8 @@ static Eterm table_info(Process* p, DbTable* tb, Eterm What)
 	    ret = am_duplicate_bag;
 	} else if (tb->common.status & DB_ORDERED_SET) {
 	    ret = am_ordered_set;
+	} else if (tb->common.status & DB_CA_ORDERED_SET) {
+	    ret = am_ordered_set;
 	} else { /*TT*/
 	    ASSERT(tb->common.status & DB_BAG);
 	    ret = am_bag;
@@ -4240,9 +4257,20 @@ static Eterm table_info(Process* p, DbTable* tb, Eterm What)
 			 make_small(stats.max_chain_len),
 			 make_small(stats.kept_items));
 	}
-	else {
+	else if (IS_CATREE_TABLE(tb->common.status)) {
+            DbCATreeStats stats;
+            Eterm* hp;
+
+            db_calc_stats_catree(&tb->catree, &stats);
+            hp = HAlloc(p, 4);
+            ret = TUPLE3(hp,
+                         make_small(stats.route_nodes),
+                         make_small(stats.base_nodes),
+                         make_small(stats.max_depth));
+
+        }
+        else
 	    ret = am_false;
-	}
     }
     return ret;
 }
@@ -4409,6 +4437,12 @@ void erts_lcnt_enable_db_lock_count(DbTable *tb, int enable) {
 
     if(IS_HASH_TABLE(tb->common.status)) {
         erts_lcnt_enable_db_hash_lock_count(&tb->hash, enable);
+    } else if(IS_CATREE_TABLE(tb->common.status)) {
+        /* erts_lcnt_enable_db_catree_lock_count is not thread safe so
+           the table needs to get locked */
+        db_lock(tb, LCK_WRITE);
+        erts_lcnt_enable_db_catree_lock_count(&tb->catree, enable);
+        db_unlock(tb, LCK_WRITE);
     }
 }
 
@@ -4441,3 +4475,16 @@ void erts_lcnt_update_db_locks(int enable) {
 #ifdef ETS_DBG_FORCE_TRAP
 erts_aint_t erts_ets_dbg_force_trap = 0;
 #endif
+
+int erts_ets_force_split(Eterm tid, int on)
+{
+    DbTable* tb = tid2tab(tid);
+    if (!tb || !IS_CATREE_TABLE(tb->common.type))
+        return 0;
+
+    db_lock(tb, LCK_WRITE);
+    if (!(tb->common.status & DB_DELETE))
+        db_catree_force_split(&tb->catree, on);
+    db_unlock(tb, LCK_WRITE);
+    return 1;
+}