Merge branch 'sverker/ets_compress/OTP-8922' into dev

* sverker/ets_compress/OTP-8922:
  ETS 'compressed' option.

1 files changed, 78 insertions, 18 deletions

diff --git a/erts/emulator/beam/erl_db_util.h b/erts/emulator/beam/erl_db_util.h
index 0f333e8b34..10ba755e80 100644
--- a/erts/emulator/beam/erl_db_util.h
+++ b/erts/emulator/beam/erl_db_util.h
@@ -52,22 +52,27 @@
 				     is broken.*/
 #define DB_ERROR_UNSPEC   -10    /* Unspecified error */
 
+/*#define DEBUG_CLONE*/
 
 /*
  * A datatype for a database entry stored out of a process heap
  */
 typedef struct db_term {
     struct erl_off_heap_header* first_oh; /* Off heap data for term. */
-    Uint size;		   /* Size of term in "words" */
-    Eterm tpl[1];          /* Untagged "constant pointer" to top tuple */
-                           /* (assumed to be first in buffer) */
+    Uint size;		   /* Heap size of term in "words" */
+#ifdef DEBUG_CLONE
+    Eterm* debug_clone;    /* An uncompressed copy */
+#endif
+    Eterm tpl[1];          /* Term data. Top tuple always first */
+
+    /* Compression: is_immed and key element are uncompressed.
+       Compressed elements are stored in external format after each other
+       last in dbterm. The top tuple elements contains byte offsets, to
+       the start of the data, tagged as headers.
+       The allocated size of the dbterm in bytes is stored at tpl[arity+1].
+     */
 } DbTerm;
 
-/* "Assign" a value to DbTerm.tpl */
-#define DBTERM_SET_TPL(dbtermPtr,tplPtr) ASSERT((tplPtr)==(dbtermPtr->tpl))
-/* Get start of term buffer */
-#define DBTERM_BUF(dbtermPtr) ((dbtermPtr)->tpl)
-
 union db_table;
 typedef union db_table DbTable;
 
@@ -186,6 +191,12 @@ typedef struct db_table_method
 
 } DbTableMethod;
 
+typedef struct db_fixation {
+    Eterm pid;
+    Uint counter;
+    struct db_fixation *next;
+} DbFixation;
+
 /*
  * This structure contains data for all different types of database
  * tables. Note that these fields must match the same fields
@@ -194,13 +205,6 @@ typedef struct db_table_method
  * operations may be the same on different types of tables.
  */
 
-typedef struct db_fixation {
-    Eterm pid;
-    Uint counter;
-    struct db_fixation *next;
-} DbFixation;
-
-
 typedef struct db_table_common {
     erts_refc_t ref;
     erts_refc_t fixref;       /* fixation counter */
@@ -226,6 +230,7 @@ typedef struct db_table_common {
     Uint32 status;            /* bit masks defined  below */
     int slot;                 /* slot index in meta_main_tab */
     int keypos;               /* defaults to 1 */
+    int compress;
 } DbTableCommon;
 
 /* These are status bit patterns */
@@ -252,6 +257,54 @@ typedef struct db_table_common {
 #define IS_FIXED(T) (NFIXED(T) != 0) 
 
 Eterm erts_ets_copy_object(Eterm, Process*);
+Eterm db_copy_from_comp(DbTableCommon* tb, DbTerm* bp, Eterm** hpp,
+			ErlOffHeap* off_heap);
+int db_eq_comp(DbTableCommon* tb, Eterm a, DbTerm* b);
+DbTerm* db_alloc_tmp_uncompressed(DbTableCommon* tb, DbTerm* org);
+
+ERTS_GLB_INLINE Eterm db_copy_object_from_ets(DbTableCommon* tb, DbTerm* bp,
+					      Eterm** hpp, ErlOffHeap* off_heap);
+ERTS_GLB_INLINE int db_eq(DbTableCommon* tb, Eterm a, DbTerm* b);
+ERTS_GLB_INLINE Eterm db_do_read_element(DbUpdateHandle* handle, Sint position);
+
+#if ERTS_GLB_INLINE_INCL_FUNC_DEF
+ERTS_GLB_INLINE Eterm db_copy_object_from_ets(DbTableCommon* tb, DbTerm* bp,
+					      Eterm** hpp, ErlOffHeap* off_heap)
+{
+    if (tb->compress) {
+	return db_copy_from_comp(tb, bp, hpp, off_heap);
+    }
+    else {
+	return copy_shallow(bp->tpl, bp->size, hpp, off_heap);
+    }
+}
+
+ERTS_GLB_INLINE int db_eq(DbTableCommon* tb, Eterm a, DbTerm* b)
+{
+    if (!tb->compress) {
+	return eq(a, make_tuple(b->tpl));
+    }
+    else {
+	return db_eq_comp(tb, a, b);
+    }
+}
+
+/* Must be called to read elements after db_lookup_dbterm.
+** Will decompress if needed. */
+ERTS_GLB_INLINE Eterm db_do_read_element(DbUpdateHandle* handle, Sint position)
+{
+    Eterm elem = handle->dbterm->tpl[position];
+    if (!is_header(elem)) {
+	return elem;
+    }
+    ASSERT(((DbTableCommon*)handle->tb)->compress);
+    ASSERT(!handle->mustResize);
+    handle->dbterm = db_alloc_tmp_uncompressed((DbTableCommon*)handle->tb, handle->dbterm);
+    handle->mustResize = 1;
+    return handle->dbterm->tpl[position];
+}
+
+#endif /* ERTS_GLB_INLINE_INCL_FUNC_DEF */
 
 /* optimised version of copy_object (normal case? atomic object) */
 #define COPY_OBJECT(obj, p, objp) \
@@ -277,14 +330,19 @@ Eterm db_set_trace_control_word_1(Process *p, Eterm val);
 
 void db_initialize_util(void);
 Eterm db_getkey(int keypos, Eterm obj);
-void db_free_term_data(DbTerm* p);
-void* db_get_term(DbTableCommon *tb, DbTerm* old, Uint offset, Eterm obj);
+void db_cleanup_offheap_comp(DbTerm* p);
+void db_free_term(DbTable *tb, void* basep, Uint offset);
+void* db_store_term(DbTableCommon *tb, DbTerm* old, Uint offset, Eterm obj);
+void* db_store_term_comp(DbTableCommon *tb, DbTerm* old, Uint offset, Eterm obj);
+Eterm db_copy_element_from_ets(DbTableCommon* tb, Process* p, DbTerm* obj,
+			       Uint pos, Eterm** hpp, Uint extra);
 int db_has_variable(Eterm obj);
 int db_is_variable(Eterm obj);
+Eterm db_do_read_element(DbUpdateHandle* handle, Sint position);
 void db_do_update_element(DbUpdateHandle* handle,
 			  Sint position,
 			  Eterm newval);
-void db_finalize_update_element(DbUpdateHandle* handle);
+void db_finalize_resize(DbUpdateHandle* handle, Uint offset);
 Eterm db_add_counter(Eterm** hpp, Eterm counter, Eterm incr);
 Eterm db_match_set_lint(Process *p, Eterm matchexpr, Uint flags);
 Binary *db_match_set_compile(Process *p, Eterm matchexpr, 
@@ -366,6 +424,8 @@ Binary *db_match_compile(Eterm *matchexpr, Eterm *guards,
 			 Eterm *body, int num_matches, 
 			 Uint flags, 
 			 DMCErrInfo *err_info);
+Eterm db_prog_match_and_copy(DbTableCommon* tb, Process* c_p, Binary* bprog,
+			     int all, DbTerm* obj, Eterm** hpp, Uint extra);
 /* Returns newly allocated MatchProg binary with refc == 0*/
 Eterm db_prog_match(Process *p, Binary *prog, Eterm term, Eterm *termp, int arity,
 		    Uint32 *return_flags /* Zeroed on enter */);