[t:2811], merge to main \!\!\!

git-svn-id: file:///svn/toku/tokudb@23495 c7de825b-a66e-492c-adef-691d508d4ae1

buildheader/db.h_4_1

@@ -214,7 +214,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 524288
 #define DB_TXN_NOWAIT 2048
 #define DB_TXN_SYNC 4096
-#define DB_INHERIT_ISOLATION 1
+#define DB_TXN_SNAPSHOT 1
+#define DB_READ_UNCOMMITTED 2
+#define DB_READ_COMMITTED 4
+#define DB_INHERIT_ISOLATION 8
+#define DB_SERIALIZABLE 16
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -226,6 +230,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -351,10 +357,9 @@ struct __toku_db {
 #define db_struct_i(x) ((x)->i)
   int (*key_range64)(DB*, DB_TXN *, DBT *, u_int64_t *less, u_int64_t *equal, u_int64_t *greater, int *is_exact);
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
+  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   void *app_private; /* 32-bit offset=16 size=4, 64=bit offset=32 size=8 */
   DB_ENV *dbenv; /* 32-bit offset=20 size=4, 64=bit offset=40 size=8 */
-  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
   int (*row_size_supported) (DB*, u_int32_t) /* Test whether a row size is supported. */;
@@ -362,6 +367,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void* __toku_dummy0[20];
   char __toku_dummy1[96];
@@ -448,7 +454,8 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[11];
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
+  void* __toku_dummy0[10];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=272 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=280 size=8 */

buildheader/db.h_4_3

@@ -216,7 +216,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 268435456
 #define DB_TXN_NOWAIT 4096
 #define DB_TXN_SYNC 8192
-#define DB_INHERIT_ISOLATION 1
+#define DB_TXN_SNAPSHOT 1
+#define DB_READ_UNCOMMITTED 2
+#define DB_READ_COMMITTED 4
+#define DB_INHERIT_ISOLATION 8
+#define DB_SERIALIZABLE 16
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -228,6 +232,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -361,10 +367,9 @@ struct __toku_db {
 #define db_struct_i(x) ((x)->i)
   int (*key_range64)(DB*, DB_TXN *, DBT *, u_int64_t *less, u_int64_t *equal, u_int64_t *greater, int *is_exact);
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
+  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   void *app_private; /* 32-bit offset=16 size=4, 64=bit offset=32 size=8 */
   DB_ENV *dbenv; /* 32-bit offset=20 size=4, 64=bit offset=40 size=8 */
-  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
   int (*row_size_supported) (DB*, u_int32_t) /* Test whether a row size is supported. */;
@@ -372,6 +377,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void* __toku_dummy0[23];
   char __toku_dummy1[96];
@@ -464,7 +470,8 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[9];
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
+  void* __toku_dummy0[8];
   char __toku_dummy1[112];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=264 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=272 size=8 */

buildheader/db.h_4_4

@@ -218,7 +218,9 @@ typedef enum {
 #define DB_TXN_SYNC 16384
 #define DB_READ_UNCOMMITTED 67108864
 #define DB_READ_COMMITTED 33554432
-#define DB_INHERIT_ISOLATION 1
+#define DB_TXN_SNAPSHOT 1
+#define DB_INHERIT_ISOLATION 2
+#define DB_SERIALIZABLE 4
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -230,6 +232,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -365,10 +369,9 @@ struct __toku_db {
 #define db_struct_i(x) ((x)->i)
   int (*key_range64)(DB*, DB_TXN *, DBT *, u_int64_t *less, u_int64_t *equal, u_int64_t *greater, int *is_exact);
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
+  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   void *app_private; /* 32-bit offset=16 size=4, 64=bit offset=32 size=8 */
   DB_ENV *dbenv; /* 32-bit offset=20 size=4, 64=bit offset=40 size=8 */
-  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
   int (*row_size_supported) (DB*, u_int32_t) /* Test whether a row size is supported. */;
@@ -376,6 +379,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void* __toku_dummy0[25];
   char __toku_dummy1[96];
@@ -472,7 +476,8 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[11];
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
+  void* __toku_dummy0[10];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=188 size=4, 64=bit offset=272 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=192 size=4, 64=bit offset=280 size=8 */

buildheader/db.h_4_5

@@ -216,9 +216,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 2048
 #define DB_TXN_NOWAIT 16384
 #define DB_TXN_SYNC 32768
+#define DB_TXN_SNAPSHOT 268435456
 #define DB_READ_UNCOMMITTED 134217728
 #define DB_READ_COMMITTED 67108864
 #define DB_INHERIT_ISOLATION 1
+#define DB_SERIALIZABLE 2
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -230,6 +232,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -365,10 +369,9 @@ struct __toku_db {
 #define db_struct_i(x) ((x)->i)
   int (*key_range64)(DB*, DB_TXN *, DBT *, u_int64_t *less, u_int64_t *equal, u_int64_t *greater, int *is_exact);
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
+  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   void *app_private; /* 32-bit offset=16 size=4, 64=bit offset=32 size=8 */
   DB_ENV *dbenv; /* 32-bit offset=20 size=4, 64=bit offset=40 size=8 */
-  int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
   int (*row_size_supported) (DB*, u_int32_t) /* Test whether a row size is supported. */;
@@ -376,6 +379,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void* __toku_dummy0[28];
   char __toku_dummy1[96];
@@ -472,7 +476,8 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[15];
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
+  void* __toku_dummy0[14];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=204 size=4, 64=bit offset=304 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=208 size=4, 64=bit offset=312 size=8 */

buildheader/db.h_4_6

@@ -217,9 +217,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 4096
 #define DB_TXN_NOWAIT 1024
 #define DB_TXN_SYNC 16384
+#define DB_TXN_SNAPSHOT 268435456
 #define DB_READ_UNCOMMITTED 134217728
 #define DB_READ_COMMITTED 67108864
 #define DB_INHERIT_ISOLATION 1
+#define DB_SERIALIZABLE 2
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -231,6 +233,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -370,7 +374,6 @@ struct __toku_db {
   char __toku_dummy0[8];
   void *app_private; /* 32-bit offset=20 size=4, 64=bit offset=32 size=8 */
   DB_ENV *dbenv; /* 32-bit offset=24 size=4, 64=bit offset=40 size=8 */
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
   int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
@@ -379,6 +382,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void* __toku_dummy1[32];
   char __toku_dummy2[80];
@@ -476,7 +480,8 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
-  void* __toku_dummy0[25];
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
+  void* __toku_dummy0[24];
   char __toku_dummy1[104];
   int (*c_close) (DBC *); /* 32-bit offset=244 size=4, 64=bit offset=384 size=8 */
   int (*c_count) (DBC *, db_recno_t *, u_int32_t); /* 32-bit offset=248 size=4, 64=bit offset=392 size=8 */

buildheader/make_db_h.c

@@ -41,6 +41,7 @@ void print_db_notices (void) {
     }                                           \
     assert(which < 32);                         \
     printf("#define %s %d\n", #name, bit);      \
+    flags |= bit;                               \
 } while (0)
 
 #define dodefine_track_enum(flags, name) do {assert(name>=0 && name<256); \
@@ -69,6 +70,8 @@ enum {
         TOKUDB_DICTIONARY_NO_HEADER    = -100006,
         TOKUDB_CANCELED                = -100007,
         TOKUDB_NO_DATA                 = -100008,
+        TOKUDB_ACCEPT                  = -100009,
+        TOKUDB_MVCC_DICTIONARY_TOO_NEW = -100010,
 };
 
 static void print_defines (void) {
@@ -177,13 +180,27 @@ static void print_defines (void) {
         dodefine_track(txn_flags, DB_TXN_WRITE_NOSYNC);
         dodefine_track(txn_flags, DB_TXN_NOWAIT);
         dodefine_track(txn_flags, DB_TXN_SYNC);
+#ifdef DB_TXN_SNAPSHOT
+        dodefine_track(txn_flags, DB_TXN_SNAPSHOT);
+#endif
 #ifdef DB_READ_UNCOMMITTED
         dodefine_track(txn_flags, DB_READ_UNCOMMITTED);
 #endif
 #ifdef DB_READ_COMMITTED
         dodefine_track(txn_flags, DB_READ_COMMITTED);
+#endif
+        //Add them if they didn't exist
+#ifndef DB_TXN_SNAPSHOT
+        dodefine_from_track(txn_flags, DB_TXN_SNAPSHOT);
+#endif
+#ifndef DB_READ_UNCOMMITTED
+        dodefine_from_track(txn_flags, DB_READ_UNCOMMITTED);
+#endif
+#ifndef DB_READ_COMMITTED
+        dodefine_from_track(txn_flags, DB_READ_COMMITTED);
 #endif
         dodefine_from_track(txn_flags, DB_INHERIT_ISOLATION);
+        dodefine_from_track(txn_flags, DB_SERIALIZABLE);
     }
     
     printf("#endif\n");
@@ -199,6 +216,8 @@ static void print_defines (void) {
     dodefine(TOKUDB_DICTIONARY_NO_HEADER);
     dodefine(TOKUDB_CANCELED);
     dodefine(TOKUDB_NO_DATA);
+    dodefine(TOKUDB_ACCEPT);
+    dodefine(TOKUDB_MVCC_DICTIONARY_TOO_NEW);
 
     /* LOADER flags */
     printf("/* LOADER flags */\n");
@@ -597,7 +616,6 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
 
 	const char *extra[]={"int (*key_range64)(DB*, DB_TXN *, DBT *, u_int64_t *less, u_int64_t *equal, u_int64_t *greater, int *is_exact)",
 			     "int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *)",
-			     "int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*)",
 			     "int (*pre_acquire_table_lock)(DB*, DB_TXN*)",
 			     "const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/",
 			     "const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/",
@@ -606,6 +624,7 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
                              "int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */",
 			     "int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */",
                              "int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */",
+                             "int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */",
                              "int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION)",
 			     NULL};
 	print_struct("db", 1, db_fields32, db_fields64, sizeof(db_fields32)/sizeof(db_fields32[0]), extra);
@@ -650,6 +669,7 @@ int main (int argc __attribute__((__unused__)), char *const argv[] __attribute__
 			     "int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     "int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
 			     "int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *)",
+			     "int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*)",
 			     NULL};
 	assert(sizeof(dbc_fields32)==sizeof(dbc_fields64));
 	print_struct("dbc", INTERNAL_AT_END, dbc_fields32, dbc_fields64, sizeof(dbc_fields32)/sizeof(dbc_fields32[0]), extra);

buildheader/tdb.h

@@ -217,9 +217,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 4096
 #define DB_TXN_NOWAIT 1024
 #define DB_TXN_SYNC 16384
+#define DB_TXN_SNAPSHOT 268435456
 #define DB_READ_UNCOMMITTED 134217728
 #define DB_READ_COMMITTED 67108864
 #define DB_INHERIT_ISOLATION 1
+#define DB_SERIALIZABLE 2
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -231,6 +233,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -339,7 +343,6 @@ struct __toku_db {
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
   void *app_private;
   DB_ENV *dbenv;
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
   int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
@@ -348,6 +351,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void *api_internal;
   int (*close) (DB*, u_int32_t);
@@ -413,6 +417,7 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
   int (*c_close) (DBC *);
   int (*c_count) (DBC *, db_recno_t *, u_int32_t);
   int (*c_del) (DBC *, u_int32_t);

db-benchmark-test/scanscan.c

@@ -187,9 +187,8 @@ static void scanscan_setup (void) {
     r = db->open(db, tid, dbfilename, NULL, DB_BTREE, 0, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);                           assert(r==0);
 #ifdef TOKUDB
     if (prelock) {
-	r = db->pre_acquire_read_lock(db,
-				      tid,
-				      db->dbt_neg_infty(), db->dbt_pos_infty());
+	r = db->pre_acquire_table_lock(db,
+				      tid);
 	assert(r==0);
     }
 #endif

include/db.h

@@ -217,9 +217,11 @@ typedef enum {
 #define DB_TXN_WRITE_NOSYNC 4096
 #define DB_TXN_NOWAIT 1024
 #define DB_TXN_SYNC 16384
+#define DB_TXN_SNAPSHOT 268435456
 #define DB_READ_UNCOMMITTED 134217728
 #define DB_READ_COMMITTED 67108864
 #define DB_INHERIT_ISOLATION 1
+#define DB_SERIALIZABLE 2
 #endif
 /* TOKUDB specific error codes */
 #define TOKUDB_OUT_OF_LOCKS -100000
@@ -231,6 +233,8 @@ typedef enum {
 #define TOKUDB_DICTIONARY_NO_HEADER -100006
 #define TOKUDB_CANCELED -100007
 #define TOKUDB_NO_DATA -100008
+#define TOKUDB_ACCEPT -100009
+#define TOKUDB_MVCC_DICTIONARY_TOO_NEW -100010
 /* LOADER flags */
 #define LOADER_USE_PUTS 1
 /* in wrap mode, top-level function txn_begin is renamed, but the field isn't renamed, so we have to hack it here.*/
@@ -339,7 +343,6 @@ struct __toku_db {
   int (*stat64)(DB *, DB_TXN *, DB_BTREE_STAT64 *);
   void *app_private;
   DB_ENV *dbenv;
-  int (*pre_acquire_read_lock)(DB*, DB_TXN*, const DBT*, const DBT*);
   int (*pre_acquire_table_lock)(DB*, DB_TXN*);
   const DBT* (*dbt_pos_infty)(void) /* Return the special DBT that refers to positive infinity in the lock table.*/;
   const DBT* (*dbt_neg_infty)(void)/* Return the special DBT that refers to negative infinity in the lock table.*/;
@@ -348,6 +351,7 @@ struct __toku_db {
   int (*set_descriptor) (DB*, u_int32_t version, const DBT* descriptor) /* set row/dictionary descriptor for a db.  Available only while db is open */;
   int (*getf_set)(DB*, DB_TXN*, u_int32_t, DBT*, YDB_CALLBACK_FUNCTION, void*) /* same as DBC->c_getf_set without a persistent cursor) */;
   int (*flatten)(DB*, DB_TXN*) /* Flatten a dictionary, similar to (but faster than) a table scan */;
+  int (*optimize)(DB*) /* Run garbage collecion and promote all transactions older than oldest. Amortized (happens during flattening) */;
   int (*get_fragmentation)(DB*,TOKU_DB_FRAGMENTATION);
   void *api_internal;
   int (*close) (DB*, u_int32_t);
@@ -413,6 +417,7 @@ struct __toku_dbc {
   int (*c_getf_set)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
   int (*c_getf_set_range_reverse)(DBC *, u_int32_t, DBT *, YDB_CALLBACK_FUNCTION, void *);
+  int (*c_pre_acquire_read_lock)(DBC*, const DBT*, const DBT*);
   int (*c_close) (DBC *);
   int (*c_count) (DBC *, db_recno_t *, u_int32_t);
   int (*c_del) (DBC *, u_int32_t);

newbrt/brt-internal.h

@@ -182,6 +182,7 @@ struct brt_header {
     TXNID txnid_that_created_or_locked_when_empty;
     TXNID root_that_created_or_locked_when_empty;
     TXNID txnid_that_suppressed_recovery_logs;
+    TXNID root_xid_that_created;
     struct toku_list live_brts;
     struct toku_list zombie_brts;
     struct toku_list checkpoint_before_commit_link;
@@ -251,7 +252,7 @@ struct brtenv {
 
 extern void toku_brtnode_flush_callback (CACHEFILE cachefile, int fd, BLOCKNUM nodename, void *brtnode_v, void *extraargs, long size, BOOL write_me, BOOL keep_me, BOOL for_checkpoint);
 extern int toku_brtnode_fetch_callback (CACHEFILE cachefile, int fd, BLOCKNUM nodename, u_int32_t fullhash, void **brtnode_pv, long *sizep, void*extraargs);
-extern int toku_brt_alloc_init_header(BRT t);
+extern int toku_brt_alloc_init_header(BRT t, TOKUTXN txn);
 extern int toku_read_brt_header_and_store_in_cachefile (CACHEFILE cf, struct brt_header **header, BOOL* was_open);
 extern CACHEKEY* toku_calculate_root_offset_pointer (BRT brt, u_int32_t *root_hash);
 
@@ -293,9 +294,8 @@ struct brt_cursor {
     OMTCURSOR omtcursor;
     u_int64_t  root_put_counter; // what was the count on the BRT when we validated the cursor?
     TXNID      oldest_living_xid;// what was the oldest live txnid when we created the cursor?
-    TOKULOGGER logger; // to give access to list of live transactions, needed for read_committed queries
-    TXNID ancestor_id; // txnid of ancestor, needed for read_committed queries
-    BOOL is_read_committed; // true if query is read_committed, false otherwise
+    BOOL is_snapshot_read; // true if query is read_committed, false otherwise
+    TOKUTXN ttxn;
     struct brt_cursor_leaf_info  leaf_info;
 };
 
@@ -342,6 +342,7 @@ enum brt_layout_version_e {
     BRT_LAYOUT_VERSION_10 = 10, // Diff from 9 to 10: Variable number of compressed sub-blocks per block, disk byte order == intel byte order, Subtree estimates instead of just leafentry estimates, translation table, dictionary descriptors, checksum in header, subdb support removed from brt layer
     BRT_LAYOUT_VERSION_11 = 11, // Diff from 10 to 11: Nested transaction leafentries (completely redesigned).  BRT_CMDs on disk now support XIDS (multiple txnids) instead of exactly one.
     BRT_LAYOUT_VERSION_12 = 12, // Diff from 11 to 12: Added BRT_CMD 'BRT_INSERT_NO_OVERWRITE', compressed block format, num old blocks
+    BRT_LAYOUT_VERSION_13 = 13, // Diff from 12 to 13: Added MVCC
     BRT_NEXT_VERSION,           // the version after the current version
     BRT_LAYOUT_VERSION   = BRT_NEXT_VERSION-1, // A hack so I don't have to change this line.
     BRT_LAYOUT_MIN_SUPPORTED_VERSION = BRT_LAYOUT_VERSION_12 // Minimum version supported

newbrt/brt-serialize.c

@@ -1144,6 +1144,8 @@ static u_int32_t
 serialize_brt_header_min_size (u_int32_t version) {
     u_int32_t size = 0;
     switch(version) {
+        case BRT_LAYOUT_VERSION_13:
+            size += 8;  //TXNID that created
         case BRT_LAYOUT_VERSION_12:
             size += 8;  // Number of blocks in old version.
 	    // fall through to add up bytes in previous version
@@ -1196,6 +1198,7 @@ int toku_serialize_brt_header_to_wbuf (struct wbuf *wbuf, struct brt_header *h,
     wbuf_int(wbuf, h->flags);
     wbuf_int(wbuf, h->layout_version_original);
     wbuf_ulonglong(wbuf, h->num_blocks_to_upgrade);
+    wbuf_TXNID(wbuf, h->root_xid_that_created);
     u_int32_t checksum = x1764_finish(&wbuf->checksum);
     wbuf_int(wbuf, checksum);
     assert(wbuf->ndone == wbuf->size);
@@ -1440,6 +1443,7 @@ deserialize_brtheader (int fd, struct rbuf *rb, struct brt_header **brth) {
     deserialize_descriptor_from(fd, h, &h->descriptor);
     h->layout_version_original = rbuf_int(&rc);    
     h->num_blocks_to_upgrade   = rbuf_ulonglong(&rc);
+    rbuf_TXNID(&rc, &h->root_xid_that_created);
     (void)rbuf_int(&rc); //Read in checksum and ignore (already verified).
     if (rc.ndone!=rc.size) {ret = EINVAL; goto died1;}
     toku_free(rc.buf);

newbrt/brt-test-helpers.c

@@ -87,7 +87,7 @@ int toku_testsetup_insert_to_leaf (BRT brt, BLOCKNUM blocknum, char *key, int ke
     //Generate a leafentry (committed insert key,val)
     r = apply_msg_to_leafentry(&cmd, NULL, //No old leafentry
                                &lesize, &disksize, &leafentry, 
-                               node->u.l.buffer, &node->u.l.buffer_mempool, 0);
+                               node->u.l.buffer, &node->u.l.buffer_mempool, 0, NULL, NULL);
     assert(r==0);
 
 

newbrt/brt.h

@@ -63,6 +63,8 @@ int toku_brt_lookup (BRT brt, DBT *k, BRT_GET_CALLBACK_FUNCTION getf, void *getf
 // Returns 0 if successful
 int toku_brt_insert (BRT brt, DBT *k, DBT *v, TOKUTXN txn);
 
+int toku_brt_optimize (BRT brt);
+
 // Effect: Insert a key and data pair into a brt if the oplsn is newer than the brt lsn.  This function is called during recovery.
 // Returns 0 if successful
 int toku_brt_maybe_insert (BRT brt, DBT *k, DBT *v, TOKUTXN txn, BOOL oplsn_valid, LSN oplsn, int do_logging, enum brt_msg_type type);
@@ -125,12 +127,12 @@ int toku_verify_brt (BRT brt);
 //int show_brt_blocknumbers(BRT);
 
 typedef struct brt_cursor *BRT_CURSOR;
-int toku_brt_cursor (BRT, BRT_CURSOR*, TOKULOGGER, TXNID, BOOL);
+int toku_brt_cursor (BRT, BRT_CURSOR*, TOKUTXN, BOOL);
 
 // get is deprecated in favor of the individual functions below
 int toku_brt_cursor_get (BRT_CURSOR cursor, DBT *key, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v, int get_flags);
 
-int toku_brt_flatten(BRT, TOKULOGGER logger);
+int toku_brt_flatten(BRT, TOKUTXN ttxn);
 int toku_brt_cursor_first(BRT_CURSOR cursor, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v);
 int toku_brt_cursor_last(BRT_CURSOR cursor, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v);
 int toku_brt_cursor_next(BRT_CURSOR cursor, BRT_GET_CALLBACK_FUNCTION getf, void *getf_v);
@@ -210,7 +212,10 @@ int toku_brt_zombie_needed (BRT brt);
 
 int toku_brt_get_fragmentation(BRT brt, TOKU_DB_FRAGMENTATION report);
 int toku_brt_header_set_panic(struct brt_header *h, int panic, char *panic_string);
-BOOL toku_brt_is_empty (BRT brt);
+
+BOOL toku_brt_is_empty (BRT brt, BOOL *try_again);
+// Effect: Return TRUE iff the tree is empty.  (However if  *try_again is set to TRUE by toku_brt_is_empty, then the answer is inconclusive, and the function should
+//  be tried again.  It's a good idea to release the big ydb lock in this case.
 
 double get_tdiff(void) __attribute__((__visibility__("default")));
 

newbrt/brtdump.c

@@ -119,6 +119,7 @@ dump_node (int f, BLOCKNUM blocknum, struct brt_header *h) {
 				 case BRT_COMMIT_BROADCAST_ALL: printf("COMMIT_BROADCAST_ALL"); goto ok;
 				 case BRT_COMMIT_BROADCAST_TXN: printf("COMMIT_BROADCAST_TXN"); goto ok;
 				 case BRT_ABORT_BROADCAST_TXN: printf("ABORT_BROADCAST_TXN"); goto ok;
+				 case BRT_OPTIMIZE: printf("OPTIMIZE"); goto ok;
 				 }
 				 printf("HUH?");
 			     ok:

newbrt/brtloader-internal.h

@@ -126,6 +126,7 @@ struct brtloader_s {
     int N;
     DB **dbs; // N of these
     DESCRIPTOR *descriptors; // N of these.
+    TXNID      *root_xids_that_created; // N of these.
     const char **new_fnames_in_env; // N of these.  The file names that the final data will be written to (relative to env).
 
     uint64_t *extracted_datasizes; // N of these.
@@ -158,6 +159,7 @@ struct brtloader_s {
     int progress_callback_result; // initially zero, if any call to the poll function callback returns nonzero, we save the result here (and don't call the poll callback function again).
 
     LSN load_lsn; //LSN of the fsynced 'load' log entry.  Write this LSN (as checkpoint_lsn) in brt headers made by this loader.
+    TXNID load_root_xid; //(Root) transaction that performed the load.
 
     QUEUE *fractal_queues; // an array of work queues, one for each secondary index.
     toku_pthread_t *fractal_threads;
@@ -184,6 +186,7 @@ struct fractal_thread_args {
     QUEUE                    q;
     uint64_t                 total_disksize_estimate;
     int                      errno_result; // the final result.
+    int                      which_db;
 };
 
 void toku_brt_loader_set_n_rows(BRTLOADER bl, u_int64_t n_rows);
@@ -214,7 +217,8 @@ int toku_loader_write_brt_from_q_in_C (BRTLOADER                bl,
 				       int                      fd, // write to here
 				       int                      progress_allocation,
 				       QUEUE                    q,
-				       uint64_t                  total_disksize_estimate);
+				       uint64_t                 total_disksize_estimate,
+                                       int                      which_db);
 
 int brt_loader_mergesort_row_array (struct row rows[/*n*/], int n, int which_db, DB *dest_db, brt_compare_func, BRTLOADER, struct rowset *);
 
@@ -231,12 +235,12 @@ int toku_brt_loader_internal_init (/* out */ BRTLOADER *blp,
 				   CACHETABLE cachetable,
 				   generate_row_for_put_func g,
 				   DB *src_db,
-				   int N, DB*dbs[/*N*/],
-				   const DESCRIPTOR descriptors[/*N*/],
+				   int N, BRT brts[/*N*/],
 				   const char *new_fnames_in_env[/*N*/],
 				   brt_compare_func bt_compare_functions[/*N*/],
 				   const char *temp_file_template,
-				   LSN load_lsn);
+				   LSN load_lsn,
+                                   TOKUTXN txn);
 
 void toku_brtloader_internal_destroy (BRTLOADER bl, BOOL is_error);
 

newbrt/brtloader.h

@@ -23,12 +23,12 @@ int toku_brt_loader_open (BRTLOADER *bl,
 			  generate_row_for_put_func g,
 			  DB *src_db,
 			  int N,
-			  DB *dbs[/*N*/],
-			  const DESCRIPTOR descriptors[/*N*/],
+			  BRT brts[/*N*/],
 			  const char * new_fnames_in_env[/*N*/],
 			  brt_compare_func bt_compare_functions[/*N*/],
 			  const char *temp_file_template,
-                          LSN load_lsn);
+                          LSN load_lsn,
+                          TOKUTXN txn);
 
 int toku_brt_loader_put (BRTLOADER bl, DBT *key, DBT *val);
 

newbrt/brttypes.h

@@ -21,6 +21,7 @@ extern "C" {
 typedef struct brt *BRT;
 struct brt_header;
 struct wbuf;
+struct dbuf;
 
 typedef unsigned int ITEMLEN;
 typedef const void *bytevec;
@@ -95,6 +96,7 @@ enum brt_msg_type {
     BRT_COMMIT_BROADCAST_TXN = 9, // Broadcast to all leafentries, (commit specific transaction).
     BRT_ABORT_BROADCAST_TXN  = 10, // Broadcast to all leafentries, (commit specific transaction).
     BRT_INSERT_NO_OVERWRITE = 11,
+    BRT_OPTIMIZE = 12,
 };
 
 typedef struct xids_t *XIDS;
@@ -125,6 +127,13 @@ typedef int (*generate_row_for_del_func)(DB *dest_db, DB *src_db, DBT *dest_val,
 typedef struct memarena *MEMARENA;
 typedef struct rollback_log_node *ROLLBACK_LOG_NODE;
 
+typedef enum __TXN_SNAPSHOT_TYPE { 
+    TXN_SNAPSHOT_NONE=0,
+    TXN_SNAPSHOT_ROOT=1,
+    TXN_SNAPSHOT_CHILD=2
+} TXN_SNAPSHOT_TYPE;
+
+
 #if defined(__cplusplus) || defined(__cilkplusplus)
 };
 #endif

newbrt/dbufio.c

@@ -378,7 +378,7 @@ dbufio_print(DBUFIO_FILESET bfs) {
     fprintf(stderr, "%s:%d bfs=%p", __FILE__, __LINE__, bfs);
     if (bfs->panic)
         fprintf(stderr, " panic=%d", bfs->panic_errno);
-    fprintf(stderr, " N=%d %d %"PRIuMAX, bfs->N, bfs->n_not_done, bfs->bufsize);
+    //fprintf(stderr, " N=%d %d %"PRIuMAX, bfs->N, bfs->n_not_done, bfs->bufsize);
     for (int i = 0; i < bfs->N; i++) {
         struct dbufio_file *dbf = &bfs->files[i];
         if (dbf->error_code[0] || dbf->error_code[1])

newbrt/leafentry.h

@@ -64,26 +64,51 @@ extern "C" {
 #pragma pack(push, 1)
 #endif
 struct __attribute__ ((__packed__)) leafentry {
-    u_int8_t  num_xrs;
-    u_int32_t keylen;
-    u_int32_t innermost_inserted_vallen;
+    uint8_t  attributes;
+    uint32_t keylen;
     union {
-        struct __attribute__ ((__packed__)) leafentry_committed {
-            u_int8_t key_val[0];     //Actual key, then actual val
-        } comm;
-        struct __attribute__ ((__packed__)) leafentry_provisional {
-            u_int8_t innermost_type;
-            TXNID    xid_outermost_uncommitted;
-            u_int8_t key_val_xrs[];  //Actual key,
-                                     //then actual innermost inserted val,
-                                     //then transaction records.
-        } prov;
+        struct __attribute__ ((__packed__)) leafentry_clean {
+            uint32_t vallen;
+            uint8_t  key_val[0];     //Actual key, then actual val
+        } clean;
+        struct __attribute__ ((__packed__)) leafentry_mvcc {
+            uint32_t num_cxrs;
+            uint8_t  num_pxrs;
+            u_int8_t key_xrs[0]; //Actual key,
+                                 //then interesting TXNIDs
+                                 //then interesting lengths (type bit is MSB of length)
+                                 //then interesting data
+                                 //then other transaction records
+        } mvcc;
     } u;
 };
 #if TOKU_WINDOWS
 #pragma pack(pop)
 #endif
 
+enum { LE_CLEAN = 0, LE_MVCC = 1 };
+#define LE_CLEAN_MEMSIZE(keylen, vallen)                         \
+    (sizeof(((LEAFENTRY)NULL)->attributes)      /* num_uxrs */   \
+    +sizeof(((LEAFENTRY)NULL)->keylen)          /* keylen */     \
+    +sizeof(((LEAFENTRY)NULL)->u.clean.vallen)  /* vallen */     \
+    +keylen                                     /* actual key */ \
+    +vallen)                                    /* actual val */
+
+#define LE_MVCC_COMMITTED_HEADER_MEMSIZE                          \
+    (sizeof(((LEAFENTRY)NULL)->attributes)      /* num_uxrs */    \
+    +sizeof(((LEAFENTRY)NULL)->keylen)          /* keylen */      \
+    +sizeof(((LEAFENTRY)NULL)->u.mvcc.num_cxrs) /* committed */   \
+    +sizeof(((LEAFENTRY)NULL)->u.mvcc.num_pxrs) /* provisional */ \
+    +sizeof(TXNID)                              /* transaction */ \
+    +sizeof(uint32_t)                           /* length+bit */  \
+    +sizeof(uint32_t))                          /* length+bit */ 
+
+#define LE_MVCC_COMMITTED_MEMSIZE(keylen, vallen)    \
+    (LE_MVCC_COMMITTED_HEADER_MEMSIZE                \
+    +keylen                        /* actual key */  \
+    +vallen)                       /* actual val */
+
+
 typedef struct leafentry *LEAFENTRY;
 
 
@@ -95,16 +120,13 @@ void wbuf_LEAFENTRY(struct wbuf *w, LEAFENTRY le);
 void wbuf_nocrc_LEAFENTRY(struct wbuf *w, LEAFENTRY le);
 int print_leafentry (FILE *outf, LEAFENTRY v); // Print a leafentry out in human-readable form.
 
-int le_outermost_is_del(LEAFENTRY le);
-int le_is_provdel(LEAFENTRY le); // Return true if it is a provisional delete.
+int le_latest_is_del(LEAFENTRY le); // Return true if it is a provisional delete.
+uint32_t le_num_xids(LEAFENTRY le); //Return how many xids exist (0 does not count)
 int le_has_xids(LEAFENTRY le, XIDS xids); // Return true transaction represented by xids is still provisional in this leafentry (le's xid stack is a superset or equal to xids)
 void*     le_latest_key (LEAFENTRY le); // Return the latest key (return NULL for provisional deletes)
 u_int32_t le_latest_keylen (LEAFENTRY le); // Return the latest keylen.
-void* le_outermost_key_and_len (LEAFENTRY le, u_int32_t *len);
-void* le_latest_key_and_len (LEAFENTRY le, u_int32_t *len);
 void*     le_latest_val (LEAFENTRY le); // Return the latest val (return NULL for provisional deletes)
 u_int32_t le_latest_vallen (LEAFENTRY le); // Return the latest vallen.  Returns 0 for provisional deletes.
-void* le_outermost_val_and_len (LEAFENTRY le, u_int32_t *len);
 void* le_latest_val_and_len (LEAFENTRY le, u_int32_t *len);
 
  // Return any key or value (even if it's only provisional).
@@ -114,12 +136,7 @@ void* le_key_and_len (LEAFENTRY le, u_int32_t *len);
 
 u_int64_t le_outermost_uncommitted_xid (LEAFENTRY le);
 
- // Return any key or value (even if it's only provisional)  If more than one exist, choose innermost (newest)
-void* le_innermost_inserted_val (LEAFENTRY le);
-u_int32_t le_innermost_inserted_vallen (LEAFENTRY le);
-void* le_innermost_inserted_val_and_len (LEAFENTRY le, u_int32_t *len);
-
-void le_full_promotion(LEAFENTRY le, size_t *new_leafentry_memorysize, size_t *new_leafentry_disksize);
+void le_clean_xids(LEAFENTRY le, size_t *new_leafentry_memorysize, size_t *new_leafentry_disksize);
 //Effect: Fully promotes le.  Returns new memory/disk size.
 //        Reuses the memory of le.
 //        Memory size is guaranteed to reduce.
@@ -127,18 +144,16 @@ void le_full_promotion(LEAFENTRY le, size_t *new_leafentry_memorysize, size_t *n
 //        Pointer to le is reused.
 //           No need to update omt if it just points to the leafentry.
 //        Does not change results of:
-//           le_is_provdel()
+//           le_latest_is_del()
 //           le_latest_keylen()
 //           le_latest_vallen()
 //           le_keylen()
-//           le_innermost_inserted_vallen()
 //        le_outermost_uncommitted_xid will return 0 after this.
 //        Changes results of following pointer functions, but memcmp of old/new answers would say they're the same.
 //          Note: You would have to memdup the old answers before calling le_full_promotion, if you want to run the comparison
 //           le_latest_key()
 //           le_latest_val()
 //           le_key()
-//           le_innermost_inserted_val()
 //        le_outermost_uncommitted_xid will return 0 after this
 //        key/val pointers will change, but data pointed to by them will be the same
 //           as before
@@ -146,6 +161,18 @@ void le_full_promotion(LEAFENTRY le, size_t *new_leafentry_memorysize, size_t *n
 //Requires: le is not marked committed
 //Requires: The outermost uncommitted xid in le has actually committed (le was not yet updated to reflect that)
 
+void
+le_committed_mvcc(uint8_t *key, uint32_t keylen,
+                  uint8_t *val, uint32_t vallen,
+                  TXNID xid,
+                  void (*bytes)(struct dbuf *dbuf, const void *bytes, int nbytes),
+                  struct dbuf *d);
+void
+le_clean(uint8_t *key, uint32_t keylen,
+         uint8_t *val, uint32_t vallen,
+         void (*bytes)(struct dbuf *dbuf, const void *bytes, int nbytes),
+         struct dbuf *d);
+
 #if defined(__cplusplus) || defined(__cilkplusplus)
 };
 #endif

newbrt/log-internal.h

@@ -83,6 +83,9 @@ struct tokulogger {
     // To access these, you must have the input lock
     LSN lsn; // the next available lsn
     OMT live_txns; // a sorted tree.  Old comment said should be a hashtable.  Do we still want that?
+    OMT live_root_txns; // a sorted tree.
+    OMT snapshot_txnids;    //contains TXNID pairs (x,y) | x is snapshot txn, y is oldest in its live list
+    OMT live_list_reverse;  //contains TXNID pairs (x,y) | y is oldest txnid s.t. x is in y's live list
     struct logbuf inbuf; // data being accumulated for the write
 
     // To access these, you must have the output condition lock.
@@ -117,11 +120,15 @@ struct brtcachefile_pair {
 
 struct tokutxn {
     u_int64_t txnid64; /* this happens to be the first lsn */
+    u_int64_t ancestor_txnid64; /* this is the lsn of root transaction */
+    u_int64_t snapshot_txnid64; /* this is the lsn of the snapshot */
     TOKULOGGER logger;
     TOKUTXN    parent;
 
     u_int64_t  rollentry_raw_count;  // the total count of every byte in the transaction and all its children.
     OMT        open_brts; // a collection of the brts that we touched.  Indexed by filenum.
+    TXN_SNAPSHOT_TYPE snapshot_type;
+    OMT        live_root_txn_list; // the root txns live when the root ancestor (self if a root) started
     XIDS       xids;      //Represents the xid list
     BOOL       force_fsync_on_commit;  //This transaction NEEDS an fsync once (if) it commits.  (commit means root txn)
     TXN_PROGRESS_POLL_FUNCTION progress_poll_fun;
@@ -190,7 +197,7 @@ static inline int toku_logsizeof_BYTESTRING (BYTESTRING bs) {
 
 static inline char *fixup_fname(BYTESTRING *f) {
     assert(f->len>0);
-    char *fname = toku_xmalloc(f->len+1);
+    char *fname = (char*)toku_xmalloc(f->len+1);
     memcpy(fname, f->data, f->len);
     fname[f->len]=0;
     return fname;

newbrt/logger.c

@@ -66,6 +66,9 @@ int toku_logger_create (TOKULOGGER *resultp) {
     result->lg_max = 100<<20; // 100MB default
     // lsn is uninitialized
     r = toku_omt_create(&result->live_txns); if (r!=0) goto panic;
+    r = toku_omt_create(&result->live_root_txns); if (r!=0) goto panic;
+    r = toku_omt_create(&result->snapshot_txnids); if (r!=0) goto panic;
+    r = toku_omt_create(&result->live_list_reverse); if (r!=0) goto panic;
     result->inbuf  = (struct logbuf) {0, LOGGER_MIN_BUF_SIZE, toku_xmalloc(LOGGER_MIN_BUF_SIZE), ZERO_LSN};
     result->outbuf = (struct logbuf) {0, LOGGER_MIN_BUF_SIZE, toku_xmalloc(LOGGER_MIN_BUF_SIZE), ZERO_LSN};
     // written_lsn is uninitialized
@@ -172,7 +175,13 @@ toku_logger_open_rollback(TOKULOGGER logger, CACHETABLE cachetable, BOOL create)
     //Must have no data blocks (rollback logs or otherwise).
     toku_block_verify_no_data_blocks_except_root_unlocked(t->h->blocktable, t->h->root);
     toku_brtheader_unlock(t->h);
-    assert(toku_brt_is_empty(t));
+    BOOL try_again = TRUE;;
+    BOOL is_empty;
+    while (try_again) {
+	try_again = FALSE;
+	is_empty = toku_brt_is_empty(t, &try_again);
+    }
+    assert(is_empty);
     return r;
 }
 
@@ -202,8 +211,20 @@ toku_logger_close_rollback(TOKULOGGER logger, BOOL recovery_failed) {
             assert(!toku_list_empty(&h->live_brts));  // there is always one brt associated with the header
 	    brt_to_close = toku_list_struct(toku_list_head(&h->live_brts), struct brt, live_brt_link);
             assert(brt_to_close);
+	    assert(!brt_to_close->h->dirty);
             toku_brtheader_unlock(h);
-            assert(toku_brt_is_empty(brt_to_close));
+	    {
+		// This almost doesn't work.  If there were anything in there, then the header might get dirtied by
+		// toku_brt_is_empty().  But it turns out absolutely nothing is in there, so it's OK to assert that it's empty.
+		BOOL try_again = TRUE;
+		BOOL is_empty;
+		while (try_again) {
+		    try_again = FALSE;
+		    is_empty = toku_brt_is_empty(brt_to_close, &try_again);
+		}
+		assert(is_empty);
+	    }
+	    assert(!h->dirty); // it should not have been dirtied by the toku_brt_is_empty test.
         }
 
         char *error_string_ignore = NULL;
@@ -247,6 +268,9 @@ int toku_logger_close(TOKULOGGER *loggerp) {
     logger->is_panicked=TRUE; // Just in case this might help.
     if (logger->directory) toku_free(logger->directory);
     toku_omt_destroy(&logger->live_txns);
+    toku_omt_destroy(&logger->live_root_txns);
+    toku_omt_destroy(&logger->snapshot_txnids);
+    toku_omt_destroy(&logger->live_list_reverse);
     toku_logfilemgr_destroy(&logger->logfilemgr);
     toku_free(logger);
     *loggerp=0;

newbrt/recover.c

@@ -448,7 +448,7 @@ recover_transaction(TOKUTXN *txnp, TXNID xid, TXNID parentxid, TOKULOGGER logger
         assert(r == 0);
         assert(txn==NULL);
     }
-    r = toku_txn_begin_with_xid(parent, &txn, logger, xid);
+    r = toku_txn_begin_with_xid(parent, &txn, logger, xid, TXN_SNAPSHOT_NONE);
     assert(r == 0);
     if (txnp) *txnp = txn;
     return 0;

newbrt/rollback.c

@@ -139,15 +139,91 @@ toku_apply_txn (TOKUTXN txn, YIELDF yield, void*yieldv, LSN lsn,
     return r;
 }
 
+int
+toku_find_xid_by_xid (OMTVALUE v, void *xidv) {
+    TXNID *xid = v;
+    TXNID *xidfind = xidv;
+    if (*xid<*xidfind) return -1;
+    if (*xid>*xidfind) return +1;
+    return 0;
+}
+
+int
+toku_find_pair_by_xid (OMTVALUE v, void *xidv) {
+    XID_PAIR pair = v;
+    TXNID *xidfind = xidv;
+    if (pair->xid1<*xidfind) return -1;
+    if (pair->xid1>*xidfind) return +1;
+    return 0;
+}
+
+static int
+live_list_reverse_note_txn_end_iter(OMTVALUE live_xidv, u_int32_t UU(index), void*txnv) {
+    TOKUTXN txn = txnv;
+    TXNID xid   = txn->txnid64;
+    TXNID *live_xid = live_xidv;
+    OMTVALUE pairv;
+    XID_PAIR pair;
+    uint32_t idx;
+
+    int r;
+    OMT reverse = txn->logger->live_list_reverse;
+    r = toku_omt_find_zero(reverse, toku_find_pair_by_xid, live_xid, &pairv, &idx, NULL);
+    invariant(r==0);
+    pair = pairv;
+    invariant(pair->xid1 == *live_xid); //sanity check
+    if (pair->xid2 == xid) {
+        //There is a record that needs to be either deleted or updated
+        TXNID *olderxid;
+        OMTVALUE olderv;
+        uint32_t olderidx;
+        OMT snapshot = txn->logger->snapshot_txnids;
+        BOOL should_delete = TRUE;
+        // find the youngest txn in snapshot that is older than xid
+        r = toku_omt_find(snapshot, toku_find_xid_by_xid, &xid, -1, &olderv, &olderidx, NULL);
+        if (r==0) {
+            //There is an older txn
+            olderxid = olderv;
+            invariant(*olderxid < xid);
+            if (*olderxid >= *live_xid) {
+                //Older txn is new enough, we need to update.
+                pair->xid2 = *olderxid;
+                should_delete = FALSE;
+            }
+        }
+        else {
+            invariant(r==DB_NOTFOUND);
+        }
+        if (should_delete) {
+            //Delete record
+            toku_free(pair);
+            r = toku_omt_delete_at(reverse, idx);
+            invariant(r==0);
+        }
+    }
+    return r;
+}
+
+static int
+live_list_reverse_note_txn_end(TOKUTXN txn) {
+    int r;
+
+    r = toku_omt_iterate(txn->live_root_txn_list, live_list_reverse_note_txn_end_iter, txn);
+    invariant(r==0);
+    return r;
+}
+
+
 void toku_rollback_txn_close (TOKUTXN txn) {
     assert(txn->spilled_rollback_head.b == ROLLBACK_NONE.b);
     assert(txn->spilled_rollback_tail.b == ROLLBACK_NONE.b);
     assert(txn->current_rollback.b == ROLLBACK_NONE.b);
+    int r;
+    {
         {
             //Remove txn from list (omt) of live transactions
             OMTVALUE txnagain;
             u_int32_t idx;
-        int r;
             r = toku_omt_find_zero(txn->logger->live_txns, find_xid, txn, &txnagain, &idx, NULL);
             assert(r==0);
             assert(txn==txnagain);
@@ -155,12 +231,49 @@ void toku_rollback_txn_close (TOKUTXN txn) {
             assert(r==0);
         }
 
+        if (txn->parent==NULL) {
+            OMTVALUE txnagain;
+            u_int32_t idx;
+            //Remove txn from list of live root txns
+            r = toku_omt_find_zero(txn->logger->live_root_txns, find_xid, txn, &txnagain, &idx, NULL);
+            assert(r==0);
+            assert(txn==txnagain);
+            r = toku_omt_delete_at(txn->logger->live_root_txns, idx);
+            assert(r==0);
+        }
+        if (txn->snapshot_type != TXN_SNAPSHOT_NONE && (txn->parent==NULL || txn->snapshot_type == TXN_SNAPSHOT_CHILD)) {
+            {
+                u_int32_t idx;
+                OMTVALUE v;
+                //Free memory used for snapshot_txnids
+                r = toku_omt_find_zero(txn->logger->snapshot_txnids, toku_find_xid_by_xid, &txn->txnid64, &v, &idx, NULL);
+                invariant(r==0);
+                TXNID *xid = v;
+                invariant(*xid == txn->txnid64);
+                r = toku_omt_delete_at(txn->logger->snapshot_txnids, idx);
+                invariant(r==0);
+                toku_free(v);
+            }
+            live_list_reverse_note_txn_end(txn);
+            {
+                //Free memory used for live root txns local list
+                invariant(toku_omt_size(txn->live_root_txn_list) > 0);
+                OMTVALUE v;
+                //store a single array of txnids
+                r = toku_omt_fetch(txn->live_root_txn_list, 0, &v, NULL);
+                invariant(r==0);
+                toku_free(v);
+                toku_omt_destroy(&txn->live_root_txn_list);
+            }
+        }
+    }
+
     assert(txn->logger->oldest_living_xid <= txn->txnid64);
     if (txn->txnid64 == txn->logger->oldest_living_xid) {
         TOKULOGGER logger = txn->logger;
 
         OMTVALUE oldest_txnv;
-        int r = toku_omt_fetch(logger->live_txns, 0, &oldest_txnv, NULL);
+        r = toku_omt_fetch(logger->live_txns, 0, &oldest_txnv, NULL);
         if (r==0) {
             TOKUTXN oldest_txn = oldest_txnv;
             assert(oldest_txn != txn); // We just removed it

newbrt/rollback.h

@@ -38,6 +38,8 @@ int toku_txn_note_close_brt (BRT brt);
 int toku_logger_txn_rollback_raw_count(TOKUTXN txn, u_int64_t *raw_count);
 
 int toku_txn_find_by_xid (BRT brt, TXNID xid, TOKUTXN *txnptr);
+int toku_find_pair_by_xid (OMTVALUE v, void *txnv);
+int toku_find_xid_by_xid (OMTVALUE v, void *xidv);
 
 // these routines in roll.c
 int toku_rollback_fileentries (int fd, TOKUTXN txn, YIELDF yield, void *yieldv, LSN lsn);

newbrt/tests/brt-serialize-sub-block-test.c

@@ -50,7 +50,7 @@ static void test_sub_block(int n) {
     assert(error == 0);
 
     BRT_CURSOR cursor;
-    error = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    error = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(error == 0);
 
     for (i=0; ; i++) {

newbrt/tests/brt-test-cursor-2.c

@@ -51,7 +51,7 @@ static void test_multiple_brt_cursor_dbts(int n, DB *db) {
     }
 
     for (i=0; i<n; i++) {
-        r = toku_brt_cursor(brt, &cursors[i], NULL, TXNID_NONE, FALSE);
+        r = toku_brt_cursor(brt, &cursors[i], NULL, FALSE);
         assert(r == 0);
     }
 

newbrt/tests/brt-test-cursor.c

@@ -19,7 +19,7 @@ static void assert_cursor_notfound(BRT brt, int position) {
     BRT_CURSOR cursor=0;
     int r;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     struct check_pair pair = {0,0,0,0,0};
@@ -35,7 +35,7 @@ static void assert_cursor_value(BRT brt, int position, long long value) {
     BRT_CURSOR cursor=0;
     int r;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("key: ");
@@ -52,7 +52,7 @@ static void assert_cursor_first_last(BRT brt, long long firstv, long long lastv)
     BRT_CURSOR cursor=0;
     int r;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("first key: ");
@@ -250,7 +250,7 @@ static void assert_cursor_walk(BRT brt, int n) {
     int i;
     int r;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("key: ");
@@ -316,7 +316,7 @@ static void assert_cursor_rwalk(BRT brt, int n) {
     int i;
     int r;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("key: ");
@@ -402,7 +402,7 @@ static void assert_cursor_walk_inorder(BRT brt, int n) {
     int r;
     char *prevkey = 0;
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("key: ");
@@ -504,7 +504,7 @@ static void test_brt_cursor_split(int n, DB *db) {
         assert(r==0);
     }
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     if (test_cursor_debug && verbose) printf("key: ");
@@ -569,7 +569,7 @@ static void test_multiple_brt_cursors(int n, DB *db) {
 
     int i;
     for (i=0; i<n; i++) {
-        r = toku_brt_cursor(brt, &cursors[i], NULL, TXNID_NONE, FALSE);
+        r = toku_brt_cursor(brt, &cursors[i], NULL, FALSE);
         assert(r == 0);
     }
 
@@ -619,7 +619,7 @@ static void test_multiple_brt_cursor_walk(int n, DB *db) {
     int c;
     /* create the cursors */
     for (c=0; c<ncursors; c++) {
-        r = toku_brt_cursor(brt, &cursors[c], NULL, TXNID_NONE, FALSE);
+        r = toku_brt_cursor(brt, &cursors[c], NULL, FALSE);
         assert(r == 0);
     }
 
@@ -706,7 +706,7 @@ static void test_brt_cursor_set(int n, int cursor_op, DB *db) {
         assert(r == 0);
     }
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     /* set cursor to random keys in set { 0, 10, 20, .. 10*(n-1) } */
@@ -779,7 +779,7 @@ static void test_brt_cursor_set_range(int n, DB *db) {
         assert(r == 0);
     }
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(r==0);
 
     /* pick random keys v in 0 <= v < 10*n, the cursor should point
@@ -829,7 +829,7 @@ static void test_brt_cursor_delete(int n, DB *db) {
     error = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_brt_cursor_keycompare, db);
     assert(error == 0);
 
-    error = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);
+    error = toku_brt_cursor(brt, &cursor, NULL, FALSE);
     assert(error == 0);
 
     DBT key, val;

newbrt/tests/brt-test.c

@@ -255,7 +255,7 @@ static void test_cursor_last_empty(void) {
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, toku_get_n_items_malloced()); toku_print_malloced_items();
     r = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, toku_builtin_compare_fun, null_db);  assert(r==0);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, toku_get_n_items_malloced()); toku_print_malloced_items();
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);            assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);            assert(r==0);
     {
 	struct check_pair pair = {0,0,0,0,0};
 	r = toku_brt_cursor_get(cursor, NULL, lookup_checkf, &pair, DB_LAST);
@@ -291,7 +291,7 @@ static void test_cursor_next (void) {
     r = toku_brt_insert(brt, toku_fill_dbt(&kbt, "hello", 6), toku_fill_dbt(&vbt, "there", 6), null_txn);
     r = toku_brt_insert(brt, toku_fill_dbt(&kbt, "byebye", 7), toku_fill_dbt(&vbt, "byenow", 7), null_txn);
     if (verbose) printf("%s:%d calling toku_brt_cursor(...)\n", __FILE__, __LINE__);
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);            assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);            assert(r==0);
     toku_init_dbt(&kbt);
     //printf("%s:%d %d alloced\n", __FILE__, __LINE__, toku_get_n_items_malloced()); toku_print_malloced_items();
     toku_init_dbt(&vbt);
@@ -383,7 +383,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
     }
     {
 	BRT_CURSOR cursor=0;
-	r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);            assert(r==0);
+	r = toku_brt_cursor(brt, &cursor, NULL, FALSE);            assert(r==0);
 
 	for (i=0; i<2; i++) {
 	    unsigned char a[4],b[4];
@@ -423,7 +423,7 @@ static void test_wrongendian_compare (int wrong_p, unsigned int N) {
 	    toku_cachetable_verify(ct);
 	}
 	BRT_CURSOR cursor=0;
-	r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);            assert(r==0);
+	r = toku_brt_cursor(brt, &cursor, NULL, FALSE);            assert(r==0);
 	
 	for (i=0; i<N; i++) {
 	    unsigned char a[4],b[4];
@@ -567,7 +567,7 @@ static void test_brt_delete_present(int n) {
     /* cursor should not find anything */
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE);
     assert(r == 0);
 
     {
@@ -698,7 +698,7 @@ static void test_brt_delete_cursor_first(int n) {
     /* cursor should find the last key: n-1 */
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE);
     assert(r == 0);
 
     {
@@ -799,7 +799,7 @@ static void test_new_brt_cursor_create_close (void) {
 
     int i;
     for (i=0; i<n; i++) {
-        r = toku_brt_cursor(brt, &cursors[i], NULL, TXNID_NONE, FALSE); assert(r == 0);
+        r = toku_brt_cursor(brt, &cursors[i], NULL, FALSE); assert(r == 0);
     }
 
     for (i=0; i<n; i++) {
@@ -834,7 +834,7 @@ static void test_new_brt_cursor_first(int n, int dup_mode) {
 
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
     toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
@@ -887,7 +887,7 @@ static void test_new_brt_cursor_last(int n, int dup_mode) {
 
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     toku_init_dbt(&key); key.flags = DB_DBT_REALLOC;
     toku_init_dbt(&val); val.flags = DB_DBT_REALLOC;
@@ -940,7 +940,7 @@ static void test_new_brt_cursor_next(int n, int dup_mode) {
 
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     for (i=0; ; i++) {
 	int kk = toku_htonl(i);
@@ -984,7 +984,7 @@ static void test_new_brt_cursor_prev(int n, int dup_mode) {
 
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     for (i=n-1; ; i--) {
 	int kk = toku_htonl(i);
@@ -1028,7 +1028,7 @@ static void test_new_brt_cursor_current(int n, int dup_mode) {
 
     BRT_CURSOR cursor=0;
 
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     for (i=0; ; i++) {
 	{
@@ -1113,7 +1113,7 @@ static void test_new_brt_cursor_set_range(int n, int dup_mode) {
         r = toku_brt_insert(brt, toku_fill_dbt(&key, &k, sizeof k), toku_fill_dbt(&val, &v, sizeof v), 0); assert(r == 0);
     }
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE); assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE); assert(r==0);
 
     /* pick random keys v in 0 <= v < 10*n, the cursor should point
        to the smallest key in the tree that is >= v */
@@ -1170,7 +1170,7 @@ static void test_new_brt_cursor_set(int n, int cursor_op, DB *db) {
         r = toku_brt_insert(brt, toku_fill_dbt(&key, &k, sizeof k), toku_fill_dbt(&val, &v, sizeof v), 0); assert(r == 0);
     }
 
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE); assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE); assert(r==0);
 
     /* set cursor to random keys in set { 0, 10, 20, .. 10*(n-1) } */
     for (i=0; i<n; i++) {

newbrt/tests/brtloader-error-injector.h

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: $"
+#ident "$Id$"
 #ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
 #ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
 

newbrt/tests/brtloader-test-bad-generate.c

@@ -75,19 +75,17 @@ static void test_extractor(int nrows, int nrowsets, BOOL expect_fail) {
 
     // open the brtloader. this runs the extractor.
     const int N = 1;
-    DB *dbs[N];
-    DESCRIPTOR descriptors[N];
+    BRT brts[N];
     const char *fnames[N];
     brt_compare_func compares[N];
     for (int i = 0; i < N; i++) {
-        dbs[i] = NULL;
-        descriptors[i] = NULL;
+        brts[i] = NULL;
         fnames[i] = "";
         compares[i] = compare_int;
     }
 
     BRTLOADER loader;
-    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, dbs, descriptors, fnames, compares, "tempXXXXXX", ZERO_LSN);
+    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, brts, fnames, compares, "tempXXXXXX", ZERO_LSN, TXNID_NONE);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

newbrt/tests/brtloader-test-extractor-errors.c

@@ -87,13 +87,11 @@ static void test_extractor(int nrows, int nrowsets, BOOL expect_fail, const char
 
     // open the brtloader. this runs the extractor.
     const int N = 1;
-    DB *dbs[N];
-    DESCRIPTOR descriptors[N];
+    BRT brts[N];
     const char *fnames[N];
     brt_compare_func compares[N];
     for (int i = 0; i < N; i++) {
-        dbs[i] = NULL;
-        descriptors[i] = NULL;
+        brts[i] = NULL;
         fnames[i] = "";
         compares[i] = compare_int;
     }
@@ -102,7 +100,7 @@ static void test_extractor(int nrows, int nrowsets, BOOL expect_fail, const char
     sprintf(temp, "%s/%s", testdir, "tempXXXXXX");
 
     BRTLOADER loader;
-    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, dbs, descriptors, fnames, compares, "tempXXXXXX", ZERO_LSN);
+    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, brts, fnames, compares, "tempXXXXXX", ZERO_LSN, TXNID_NONE);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

newbrt/tests/brtloader-test-extractor.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: brtloader-test-extractor-errors.c 20466 2010-05-20 17:45:19Z prohaska $"
+#ident "$Id$"
 #ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
 #ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
 
@@ -300,13 +300,11 @@ static void test_extractor(int nrows, int nrowsets, const char *testdir) {
 
     // open the brtloader. this runs the extractor.
     const int N = 1;
-    DB *dbs[N];
-    DESCRIPTOR descriptors[N];
+    BRT brts[N];
     const char *fnames[N];
     brt_compare_func compares[N];
     for (int i = 0; i < N; i++) {
-        dbs[i] = NULL;
-        descriptors[i] = NULL;
+        brts[i] = NULL;
         fnames[i] = "";
         compares[i] = compare_int;
     }
@@ -315,7 +313,7 @@ static void test_extractor(int nrows, int nrowsets, const char *testdir) {
     sprintf(temp, "%s/%s", testdir, "tempXXXXXX");
 
     BRTLOADER loader;
-    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, dbs, descriptors, fnames, compares, temp, ZERO_LSN);
+    r = toku_brt_loader_open(&loader, NULL, generate, NULL, N, brts, fnames, compares, temp, ZERO_LSN, TXNID_NONE);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

newbrt/tests/brtloader-test-merge-files-dbufio.c

@@ -315,8 +315,7 @@ static void test (const char *directory, BOOL is_error) {
     }
 
     BRTLOADER bl;
-    DB **XMALLOC_N(N_DEST_DBS, dbs);
-    DESCRIPTOR *XMALLOC_N(N_DEST_DBS, descriptors);
+    BRT *XCALLOC_N(N_DEST_DBS, brts);
     const char **XMALLOC_N(N_DEST_DBS, new_fnames_in_env);
     for (int i=0; i<N_DEST_DBS; i++) {
 	char s[100];
@@ -338,12 +337,12 @@ static void test (const char *directory, BOOL is_error) {
 					       ct,
 					       (generate_row_for_put_func)NULL,
 					       (DB*)NULL,
-					       N_DEST_DBS, dbs,
-					       descriptors,
+					       N_DEST_DBS, brts,
 					       new_fnames_in_env,
 					       bt_compare_functions,
 					       "tempxxxxxx",
-					       *lsnp);
+					       *lsnp,
+                                               TXNID_NONE);
 	assert(r==0);
     }
 
@@ -454,8 +453,7 @@ static void test (const char *directory, BOOL is_error) {
     destroy_dbufio_fileset(bfs);
     toku_free(fnames);
     toku_free(fds);
-    toku_free(dbs);
-    toku_free(descriptors);
+    toku_free(brts);
     toku_free(new_fnames_in_env);
     toku_free(bt_compare_functions);
     toku_free(lsnp);

newbrt/tests/brtloader-test-open.c

@@ -42,13 +42,11 @@ static void test_loader_open(int ndbs) {
     BRTLOADER loader;
 
     // open the brtloader. this runs the extractor.
-    DB *dbs[ndbs];
-    DESCRIPTOR descriptors[ndbs];
+    BRT brts[ndbs];
     const char *fnames[ndbs];
     brt_compare_func compares[ndbs];
     for (int i = 0; i < ndbs; i++) {
-        dbs[i] = NULL;
-        descriptors[i] = NULL;
+        brts[i] = NULL;
         fnames[i] = "";
         compares[i] = my_compare;
     }
@@ -59,7 +57,7 @@ static void test_loader_open(int ndbs) {
     for (i = 0; ; i++) {
         set_my_malloc_trigger(i+1);
 
-        r = toku_brt_loader_open(&loader, NULL, NULL, NULL, ndbs, dbs, descriptors, fnames, compares, "", ZERO_LSN);
+        r = toku_brt_loader_open(&loader, NULL, NULL, NULL, ndbs, brts, fnames, compares, "", ZERO_LSN, TXNID_NONE);
         if (r == 0)
             break;
     }

newbrt/tests/brtloader-test-vm.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: brtloader-test.c 20778 2010-05-28 20:38:42Z yfogel $"
+#ident "$Id$"
 #ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
 #ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
 

newbrt/tests/brtloader-test-writer-errors.c

@@ -132,7 +132,7 @@ static void write_dbfile (char *template, int n, char *output_name, BOOL expect_
     brt_loader_set_error_function(&bl.error_callback, NULL, NULL);
     brt_loader_set_poll_function(&bl.poll_callback, loader_poll_callback, NULL);
 
-    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q2, size_est);
+    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q2, size_est, 0);
     // if (!(expect_error ? r != 0 : r == 0)) printf("WARNING%%d expect_error=%d r=%d\n", __LINE__, expect_error, r); 
     assert(expect_error ? r != 0 : r == 0);
 

newbrt/tests/brtloader-test-writer.c

@@ -55,7 +55,7 @@ static void verify_dbfile(int n, const char *name) {
     r = toku_brt_open(t, name, 0, 0, ct, null_txn, 0); assert(r==0);
 
     BRT_CURSOR cursor = NULL;
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     int i;
     for (i=0; ; i++) {
@@ -162,7 +162,7 @@ static void test_write_dbfile (char *template, int n, char *output_name) {
     assert(fd>=0);
 
     if (verbose) traceit("write to file");
-    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q2, size_est);
+    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q2, size_est, 0);
     assert(r==0);
 
     r = queue_destroy(q2);

newbrt/tests/brtloader-test.c

@@ -254,7 +254,7 @@ static void verify_dbfile(int n, int sorted_keys[], const char *sorted_vals[], c
     r = toku_brt_open(t, name, 0, 0, ct, null_txn, 0); assert(r==0);
 
     BRT_CURSOR cursor = NULL;
-    r = toku_brt_cursor(t, &cursor, NULL, TXNID_NONE, FALSE); assert(r == 0);
+    r = toku_brt_cursor(t, &cursor, NULL, FALSE); assert(r == 0);
 
     int i;
     for (i=0; i<n; i++) {
@@ -325,7 +325,7 @@ static void verify_dbfile(int n, int sorted_keys[], const char *sorted_vals[], c
     int fd = open(output_name, O_RDWR | O_CREAT | O_BINARY, S_IRWXU|S_IRWXG|S_IRWXO);
     assert(fd>=0);
     
-    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q, size_est);
+    r = toku_loader_write_brt_from_q_in_C(&bl, &desc, fd, 1000, q, size_est, 0);
     assert(r==0);
 
     destroy_merge_fileset(&fs);

newbrt/tests/is_empty.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "$Id$"
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
+
+#include "test.h"
+#include "includes.h"
+#include "toku_os.h"
+#include "checkpoint.h"
+
+
+#define TESTDIR "dir." __FILE__ 
+#define FILENAME "test0.brt"
+
+static void do_yield (voidfp f, void *fv, void *UU(v)) {
+    if (f) f(fv);
+}
+
+
+static void test_it (int N) {
+    BRT brt;
+    int r;
+    system("rm -rf " TESTDIR);
+    r = toku_os_mkdir(TESTDIR, S_IRWXU);                                                                    CKERR(r);
+
+    TOKULOGGER logger;
+    r = toku_logger_create(&logger);                                                                        CKERR(r);
+    r = toku_logger_open(TESTDIR, logger);                                                                  CKERR(r);
+
+
+    CACHETABLE ct;
+    r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, logger);                                               CKERR(r);
+    toku_cachetable_set_env_dir(ct, TESTDIR);
+
+    toku_logger_set_cachetable(logger, ct);
+
+    r = toku_logger_open_rollback(logger, ct, TRUE);                                                        CKERR(r);
+
+    TOKUTXN txn;
+    r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                    CKERR(r);
+
+    r = toku_open_brt(FILENAME, 1, &brt, 1024, ct, txn, toku_builtin_compare_fun, NULL);                    CKERR(r);
+
+    r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                        CKERR(r);
+    toku_txn_close_txn(txn);
+
+    r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                CKERR(r);
+    r = toku_close_brt(brt, NULL);                                                                          CKERR(r);
+
+    unsigned int rands[N];
+    for (int i=0; i<N; i++) {
+	r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                    CKERR(r);
+	r = toku_open_brt(FILENAME, 0, &brt, 1024, ct, txn, toku_builtin_compare_fun, NULL);                CKERR(r);
+	r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                        CKERR(r);
+	toku_txn_close_txn(txn);
+
+	r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                    CKERR(r);
+	char key[100],val[300];
+	DBT k, v;
+	rands[i] = random();
+	snprintf(key, sizeof(key), "key%x.%x", rands[i], i);
+	memset(val, 'v', sizeof(val));
+	val[sizeof(val)-1]=0;
+	r = toku_brt_insert(brt, toku_fill_dbt(&k, key, 1+strlen(key)), toku_fill_dbt(&v, val, 1+strlen(val)), txn);
+	r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                        CKERR(r);
+	toku_txn_close_txn(txn);
+
+
+	r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                CKERR(r);
+	r = toku_close_brt(brt, NULL);                                                                          CKERR(r);
+
+	if (verbose) printf("i=%d\n", i);
+    }
+    for (int i=0; i<N; i++) {
+	r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                    CKERR(r);
+	r = toku_open_brt(FILENAME, 0, &brt, 1024, ct, txn, toku_builtin_compare_fun, NULL);                CKERR(r);
+	r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                        CKERR(r);
+	toku_txn_close_txn(txn);
+
+	r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                    CKERR(r);
+	char key[100];
+	DBT k;
+	snprintf(key, sizeof(key), "key%x.%x", rands[i], i);
+	r = toku_brt_delete(brt, toku_fill_dbt(&k, key, 1+strlen(key)), txn);
+
+	if (0) {
+	BOOL try_again = TRUE;
+	BOOL is_empty;
+	while (try_again) {
+	    try_again = FALSE;
+	    is_empty = toku_brt_is_empty(brt, &try_again);
+	}
+	assert(!is_empty);
+	}
+	
+	r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                        CKERR(r);
+	toku_txn_close_txn(txn);
+
+
+	r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                CKERR(r);
+	r = toku_close_brt(brt, NULL);                                                                          CKERR(r);
+
+	if (verbose) printf("d=%d\n", i);
+    }
+    r = toku_txn_begin_txn((TOKUTXN)0, &txn, logger, TXN_SNAPSHOT_ROOT);                                       CKERR(r);
+    r = toku_open_brt(FILENAME, 0, &brt, 1024, ct, txn, toku_builtin_compare_fun, NULL);                       CKERR(r);
+    r = toku_txn_commit_txn(txn, FALSE, do_yield, NULL, NULL, NULL);                                           CKERR(r);
+    toku_txn_close_txn(txn);
+
+    if (0) {
+    BOOL try_again = TRUE;
+    BOOL is_empty;
+    while (try_again) {
+	try_again = FALSE;
+	is_empty = toku_brt_is_empty(brt, &try_again);
+    }
+    assert(is_empty);
+    }
+
+    r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                   CKERR(r);
+    r = toku_close_brt(brt, NULL);                                                                             CKERR(r);
+
+    r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                   CKERR(r);
+    r = toku_logger_close_rollback(logger, FALSE);                                                             CKERR(r);
+    r = toku_checkpoint(ct, logger, NULL, NULL, NULL, NULL);                                                   CKERR(r);
+    r = toku_cachetable_close(&ct);                                                                            CKERR(r);
+    r = toku_logger_close(&logger);                                                        assert(r==0);
+
+}
+    
+
+int test_main (int argc, const char *argv[]) {
+    default_parse_args(argc, argv);
+    for (int i=1; i<=64; i++) {
+	test_it(i);
+    }
+    return 0;
+}

newbrt/tests/shortcut.c

@@ -22,7 +22,7 @@ test_main (int argc __attribute__((__unused__)), const char *argv[]  __attribute
 
     r = toku_brt_create_cachetable(&ct, 0, ZERO_LSN, NULL_LOGGER);                               assert(r==0);
     r = toku_open_brt(fname, 1, &brt, 1<<12, ct, null_txn, test_brt_cursor_keycompare, db);   assert(r==0);
-    r = toku_brt_cursor(brt, &cursor, NULL, TXNID_NONE, FALSE);                                        assert(r==0);
+    r = toku_brt_cursor(brt, &cursor, NULL, FALSE);               assert(r==0);
 
     int i;
     for (i=0; i<1000; i++) {

newbrt/tests/test.h

+#ident "$Id$"
+
 #include <toku_portability.h>
 #include "toku_assert.h"
 #include <stdio.h>

newbrt/txn.h

@@ -9,8 +9,20 @@
 extern "C" {
 #endif
 
-int toku_txn_begin_txn (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGER logger);
-int toku_txn_begin_with_xid (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGER logger, TXNID xid);
+int toku_txn_begin_txn (
+    TOKUTXN parent_tokutxn, 
+    TOKUTXN *tokutxn, 
+    TOKULOGGER logger,
+    TXN_SNAPSHOT_TYPE snapshot_type
+    );
+
+int toku_txn_begin_with_xid (
+    TOKUTXN parent_tokutxn, 
+    TOKUTXN *tokutxn, 
+    TOKULOGGER logger, 
+    TXNID xid, 
+    TXN_SNAPSHOT_TYPE snapshot_type
+    );
 int toku_txn_load_txninfo (TOKUTXN txn, TXNINFO info);
 
 int toku_txn_commit_txn (TOKUTXN txn, int nosync, YIELDF yield, void *yieldv,
@@ -48,6 +60,15 @@ typedef struct txn_status {
 
 void toku_txn_get_status(TXN_STATUS s);
 
+BOOL toku_is_txn_in_live_root_txn_list(TOKUTXN txn, TXNID xid);
+
+TXNID toku_get_oldest_in_live_root_txn_list(TOKUTXN txn);
+
+typedef struct {
+    TXNID xid1;
+    TXNID xid2;
+} XID_PAIR_S, *XID_PAIR;
+
 #if defined(__cplusplus) || defined(__cilkplusplus)
 };
 #endif

newbrt/ule.h

@@ -18,6 +18,7 @@ extern "C" {
 
 //1 does much slower debugging
 #define ULE_DEBUG 0
+#define GARBAGE_COLLECTION_DEBUG 0
 
 /////////////////////////////////////////////////////////////////////////////////
 // Following data structures are the unpacked format of a leafentry. 
@@ -33,8 +34,8 @@ enum {XR_INSERT      = 1,
       XR_PLACEHOLDER = 3};
 
 typedef struct {     // unpacked transaction record
-    u_int8_t   type;     // delete/insert/placeholder
-    u_int32_t  vallen;   // number of bytes in value
+    uint8_t   type;     // delete/insert/placeholder
+    uint32_t  vallen;   // number of bytes in value
     void *    valp;     // pointer to value  (Where is value really stored?)
     TXNID     xid;      // transaction id
     // Note: when packing ule into a new leafentry, will need
@@ -44,10 +45,13 @@ typedef struct {     // unpacked transaction record
 // Unpacked Leaf Entry is of fixed size because it's just on the 
 // stack and we care about ease of access more than the memory footprint.
 typedef struct {     // unpacked leaf entry
-    u_int8_t   num_uxrs;   // how many of uxrs[] are valid
-    u_int32_t  keylen;
+    uint32_t  num_puxrs;   // how many of uxrs[] are provisional
+    uint32_t  num_cuxrs;   // how many of uxrs[] are committed
+    uint32_t  keylen;
     void *    keyp;
-    UXR_S      uxrs[MAX_TRANSACTION_RECORDS];    // uxrs[0] is outermost, uxrs[num_uxrs-1] is innermost
+    UXR_S     uxrs_static[MAX_TRANSACTION_RECORDS*2];    // uxrs[0] is oldest committed (txn commit time, not txn start time), uxrs[num_cuxrs] is outermost provisional value (if any exist/num_puxrs > 0)
+    UXR       uxrs;                                      //If num_cuxrs < MAX_TRANSACTION_RECORDS then &uxrs_static[0].
+                                                         //Otherwise we use a dynamically allocated array of size num_cuxrs + 1 + MAX_TRANSATION_RECORD.
 } ULE_S, *ULE;
 
 int apply_msg_to_leafentry(BRT_MSG   msg,
@@ -57,7 +61,24 @@ int apply_msg_to_leafentry(BRT_MSG   msg,
 			   LEAFENTRY *new_leafentry_p,
 			   OMT omt, 
 			   struct mempool *mp, 
-			   void **maybe_free);
+			   void **maybe_free,
+                           OMT snapshot_xids,
+                           OMT live_list_reverse);
+
+void 
+test_msg_modify_ule(ULE ule, BRT_MSG msg);
+
+//Callback contract:
+//  Returns:
+//      0:  Ignore this entry and go on to next one.
+//      TOKUDB_ACCEPT: Quit early, accept this transaction record and return appropriate data
+//      r|r!=0&&r!=TOKUDB_ACCEPT:  Quit early, return r
+typedef int(*LE_ITERATE_CALLBACK)(TXNID id, TOKUTXN context);
+
+int le_iterate_is_empty(LEAFENTRY le, LE_ITERATE_CALLBACK f, BOOL *is_empty, TOKUTXN context);
+
+int le_iterate_val(LEAFENTRY le, LE_ITERATE_CALLBACK f, void** valpp, u_int32_t *vallenp, TOKUTXN context);
+
 
 //////////////////////////////////////////////////////////////////////////////////////
 //Functions exported for test purposes only (used internally for non-test purposes).
@@ -72,7 +93,9 @@ int le_pack(ULE ule,                            // data to be packed into new le
 
 
 size_t le_memsize_from_ule (ULE ule);
+void ule_cleanup(ULE ule);
 
+TXNID toku_get_youngest_live_list_txnid_for(TXNID xc, OMT live_list_reverse);
 #if defined(__cplusplus) || defined(__cilkplusplus)
 };
 #endif

newbrt/xids-internal.h

@@ -22,7 +22,7 @@ extern "C" {
 #endif
 
 typedef struct __attribute__((__packed__)) xids_t {
-    u_int8_t  num_stored_xids;    // maximum value of MAX_TRANSACTION_RECORDS - 1 ...
+    u_int8_t  num_xids;    // maximum value of MAX_TRANSACTION_RECORDS - 1 ...
                            // ... because transaction 0 is implicit
     TXNID     ids[];
 } XIDS_S;

newbrt/xids.c

@@ -35,7 +35,7 @@
 //  xids<> do not store the implicit transaction id of 0 at index 0.
 //  The accessor functions make the id of 0 explicit at index 0.
 //  The number of xids physically stored in the xids array is in
-//  the variable num_stored_xids.
+//  the variable num_xids.
 //
 // The xids struct is immutable.  The caller gets an initial version of XIDS
 // by calling xids_get_root_xids(), which returns the constant struct
@@ -54,7 +54,7 @@
 XIDS
 xids_get_root_xids(void) {
     static const struct xids_t root_xids = {
-        .num_stored_xids = 0
+        .num_xids = 0
     };
 
     XIDS rval = (XIDS)&root_xids;
@@ -69,23 +69,22 @@ xids_create_child(XIDS   parent_xids,		// xids list for parent transaction
 		  XIDS * xids_p,		// xids list created
 		  TXNID  this_xid) {		// xid of this transaction (new innermost)
     int rval;
-    assert(parent_xids);
-    assert(this_xid > xids_get_innermost_xid(parent_xids));
-    u_int32_t num_stored_xids = parent_xids->num_stored_xids + 1;
-    u_int32_t num_xids        = num_stored_xids + 1;
-    assert(num_xids > 0);
-    assert(num_xids <= MAX_TRANSACTION_RECORDS + 1);
-    if (num_xids > MAX_TRANSACTION_RECORDS) rval = EINVAL;
+    invariant(parent_xids);
+    invariant(this_xid > xids_get_innermost_xid(parent_xids));
+    u_int32_t num_child_xids = parent_xids->num_xids + 1;
+    invariant(num_child_xids > 0);
+    invariant(num_child_xids <= MAX_TRANSACTION_RECORDS);
+    if (num_child_xids == MAX_TRANSACTION_RECORDS) rval = EINVAL;
     else {
-        XIDS xids = toku_malloc(sizeof(*xids) + num_stored_xids*sizeof(xids->ids[0]));
+        XIDS xids = toku_malloc(sizeof(*xids) + num_child_xids*sizeof(xids->ids[0]));
         if (!xids) rval = ENOMEM;
         else {
-            xids->num_stored_xids = num_stored_xids;
+            xids->num_xids = num_child_xids;
             memcpy(xids->ids,
                    parent_xids->ids,
-                   parent_xids->num_stored_xids*sizeof(parent_xids->ids[0])); 
+                   parent_xids->num_xids*sizeof(parent_xids->ids[0])); 
             TXNID this_xid_disk = toku_htod64(this_xid);
-            xids->ids[num_stored_xids-1] = this_xid_disk;
+            xids->ids[num_child_xids-1] = this_xid_disk;
             *xids_p = xids;
             rval = 0;
         }
@@ -96,14 +95,12 @@ xids_create_child(XIDS   parent_xids,		// xids list for parent transaction
 void
 xids_create_from_buffer(struct rbuf *rb,		// xids list for parent transaction
 		        XIDS * xids_p) {		// xids list created
-    u_int32_t num_stored_xids = rbuf_char(rb);
-    u_int32_t num_xids        = num_stored_xids + 1;
-    assert(num_xids > 0);
-    assert(num_xids <= MAX_TRANSACTION_RECORDS);
-    XIDS xids = toku_xmalloc(sizeof(*xids) + num_stored_xids*sizeof(xids->ids[0]));
-    xids->num_stored_xids = num_stored_xids;
+    u_int32_t num_xids = rbuf_char(rb);
+    invariant(num_xids < MAX_TRANSACTION_RECORDS);
+    XIDS xids = toku_xmalloc(sizeof(*xids) + num_xids*sizeof(xids->ids[0]));
+    xids->num_xids = num_xids;
     u_int8_t index;
-    for (index = 0; index < xids->num_stored_xids; index++) {
+    for (index = 0; index < xids->num_xids; index++) {
         rbuf_TXNID(rb, &xids->ids[index]);
         if (index > 0)
             assert(xids->ids[index] > xids->ids[index-1]);
@@ -119,19 +116,14 @@ xids_destroy(XIDS *xids_p) {
 }
 
 
-
-
 // Return xid at requested position.  
 // If requesting an xid out of range (which will be the case if xids array is empty)
 // then return 0, the xid of the root transaction.
 TXNID 
 xids_get_xid(XIDS xids, u_int8_t index) {
-    TXNID rval = 0;
-    if (index > 0) {
-        assert(index < xids_get_num_xids(xids));
-        rval = xids->ids[index-1];
+    invariant(index < xids_get_num_xids(xids));
+    TXNID rval = xids->ids[index];
     rval = toku_dtoh64(rval);
-    }
     return rval;
 }
 
@@ -142,7 +134,7 @@ xids_find_index_of_xid(XIDS xids, TXNID target_xid) {
     u_int8_t index = 0;  // search outer to inner
     TXNID current_xid = xids_get_xid(xids, index);
     while (current_xid != target_xid) {
-        assert(current_xid < target_xid);
+        invariant(current_xid < target_xid);
         index++;
         current_xid = xids_get_xid(xids, index); // Next inner txnid in xids.
     }
@@ -151,14 +143,18 @@ xids_find_index_of_xid(XIDS xids, TXNID target_xid) {
 
 u_int8_t 
 xids_get_num_xids(XIDS xids) {
-    u_int8_t rval = xids->num_stored_xids+1; //+1 for the id of 0 made explicit by xids<> accessors
+    u_int8_t rval = xids->num_xids;
     return rval;
 }
 
+
 // Return innermost xid 
 TXNID 
 xids_get_innermost_xid(XIDS xids) {
-    TXNID rval = xids_get_xid(xids, xids_get_num_xids(xids)-1);
+    TXNID rval = TXNID_NONE;
+    if (xids_get_num_xids(xids)) {
+        rval = xids_get_xid(xids, xids_get_num_xids(xids)-1);
+    }
     return rval;
 }
 
@@ -172,23 +168,23 @@ xids_cpy(XIDS target, XIDS source) {
 u_int32_t 
 xids_get_size(XIDS xids){
     u_int32_t rval;
-    u_int8_t num_stored_xids = xids->num_stored_xids;
-    rval = sizeof(*xids) + num_stored_xids * sizeof(xids->ids[0]);
+    u_int8_t num_xids = xids->num_xids;
+    rval = sizeof(*xids) + num_xids * sizeof(xids->ids[0]);
     return rval;
 }
 
 u_int32_t 
 xids_get_serialize_size(XIDS xids){
     u_int32_t rval;
-    u_int8_t num_stored_xids = xids->num_stored_xids;
-    rval = 1 + //num stored xids
-           8 * num_stored_xids;
+    u_int8_t num_xids = xids->num_xids;
+    rval = 1 + //num xids
+           8 * num_xids;
     return rval;
 }
 
 void
 toku_calc_more_murmur_xids (struct x1764 *mm, XIDS xids) {
-    x1764_add(mm, &xids->num_stored_xids, 1);
+    x1764_add(mm, &xids->num_xids, 1);
     u_int8_t index;
     u_int8_t num_xids = xids_get_num_xids(xids);
     for (index = 0; index < num_xids; index++) {
@@ -199,22 +195,22 @@ toku_calc_more_murmur_xids (struct x1764 *mm, XIDS xids) {
 
 unsigned char *
 xids_get_end_of_array(XIDS xids) {
-    TXNID *r = xids->ids + xids->num_stored_xids;
+    TXNID *r = xids->ids + xids->num_xids;
     return (unsigned char*)r;
 }
 
 void wbuf_xids(struct wbuf *wb, XIDS xids) {
-    wbuf_char(wb, (unsigned char)xids->num_stored_xids);
+    wbuf_char(wb, (unsigned char)xids->num_xids);
     u_int8_t index;
-    for (index = 0; index < xids->num_stored_xids; index++) {
+    for (index = 0; index < xids->num_xids; index++) {
         wbuf_TXNID(wb, xids->ids[index]);
     }
 }
 
 void wbuf_nocrc_xids(struct wbuf *wb, XIDS xids) {
-    wbuf_nocrc_char(wb, (unsigned char)xids->num_stored_xids);
+    wbuf_nocrc_char(wb, (unsigned char)xids->num_xids);
     u_int8_t index;
-    for (index = 0; index < xids->num_stored_xids; index++) {
+    for (index = 0; index < xids->num_xids; index++) {
         wbuf_nocrc_TXNID(wb, xids->ids[index]);
     }
 }

release/examples/db-scan.c

@@ -130,9 +130,7 @@ static void scanscan_setup (void) {
     r = db->open(db, tid, dbfilename, NULL, DB_BTREE, 0, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);                           assert(r==0);
 #ifdef TOKUDB
     if (prelock) {
-	r = db->pre_acquire_read_lock(db,
-				      tid,
-				      db->dbt_neg_infty(), db->dbt_pos_infty());
+	r = db->pre_acquire_table_lock(db, tid);
 	assert(r==0);
     }
 #endif

src/loader.c

@@ -212,9 +212,9 @@ int toku_loader_create_loader(DB_ENV *env,
 	}
 	else {
 	    char **XMALLOC_N(N, new_inames_in_env);
-	    DESCRIPTOR *XMALLOC_N(N, descriptors);
+	    BRT *XMALLOC_N(N, brts);
 	    for (int i=0; i<N; i++) {
-		descriptors[i] = &dbs[i]->i->brt->h->descriptor;
+		brts[i] = dbs[i]->i->brt;
 	    }
 	    loader->i->ekeys = NULL;
 	    loader->i->evals = NULL;
@@ -222,29 +222,30 @@ int toku_loader_create_loader(DB_ENV *env,
 	    r = locked_ydb_load_inames (env, txn, N, dbs, new_inames_in_env, &load_lsn);
 	    if ( r!=0 ) {
 		toku_free(new_inames_in_env);
-		toku_free(descriptors);
+		toku_free(brts);
 		rval = r;
 		goto create_exit;
 	    }
+            TOKUTXN ttxn = txn ? db_txn_struct_i(txn)->tokutxn : NULL;
 	    r = toku_brt_loader_open(&loader->i->brt_loader,
 				     loader->i->env->i->cachetable,
 				     loader->i->env->i->generate_row_for_put,
 				     src_db,
 				     N,
-				     dbs,
-				     descriptors,
+				     brts,
 				     (const char **)new_inames_in_env,
 				     compare_functions,
 				     loader->i->temp_file_template,
-				     load_lsn);
+				     load_lsn,
+                                     ttxn);
             if ( r!=0 ) {
 		toku_free(new_inames_in_env);
-		toku_free(descriptors);
+		toku_free(brts);
                 rval = r;
                 goto create_exit;
             }
 	    loader->i->inames_in_env = new_inames_in_env;
-	    toku_free(descriptors);
+	    toku_free(brts);
 	    rval = 0;
 	}
     }

src/tests/Makefile

@@ -87,6 +87,7 @@ BDB_DONTRUN_TESTS = \
         checkpoint_callback \
         checkpoint_stress \
         checkpoint_truncate_1 \
+	cursor-isolation \
         diskfull \
         env-put-multiple \
         env_startup \
@@ -110,6 +111,7 @@ BDB_DONTRUN_TESTS = \
         loader-tpch-load \
         manyfiles \
 	multiprocess \
+	mvcc-create-table \
         powerfail \
 	preload-3.1-db \
         progress \
@@ -137,6 +139,7 @@ BDB_DONTRUN_TESTS = \
         root_fifo_32 \
         root_fifo_41 \
         stat64 stat64_flatten \
+	stress-gc \
         test1324 \
         test1426 \
         test1572 \
@@ -469,6 +472,8 @@ test_groupcommit_perf.bdbrun test_groupcommit_perf.tdbrun: VGRIND=
 
 # Don't run valgrind on the diskfull test (since it aborts)
 diskfull.tdbrun: VGRIND=
+stress-gc.tdbrun: VGRIND=
+
 
 libs:
 	cd ..;$(MAKE)

src/tests/cursor-isolation.c

+// Test that flag settings for cursor isolation works
+
+#include "test.h"
+
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+int test_main (int argc, char * const argv[]) {
+  parse_args(argc, argv);
+  int r;
+  r = system("rm -rf " ENVDIR);
+  CKERR(r);
+  toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+  DB_ENV *env;
+  r = db_env_create(&env, 0);                                                         CKERR(r);
+  env->set_errfile(env, stderr);
+  r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      CKERR(r);
+    
+  DB *db;
+  {
+    DB_TXN *txna;
+    r = env->txn_begin(env, NULL, &txna, 0);                                        CKERR(r);
+
+    r = db_create(&db, env, 0);                                                     CKERR(r);
+    r = db->open(db, txna, "foo.db", NULL, DB_BTREE, DB_CREATE, 0666);              CKERR(r);
+
+    DBT key,val;
+    r = db->put(db, txna, dbt_init(&key, "a", 4), dbt_init(&val, "a", 4), 0);       CKERR(r);
+
+    r = txna->commit(txna, 0);                                                      CKERR(r);
+  }
+
+  DB_TXN *txn_serializable, *txn_committed, *txn_uncommitted;
+  DBC* cursor = NULL;
+  r = env->txn_begin(env, NULL, &txn_serializable, DB_SERIALIZABLE);                          CKERR(r);
+  r = env->txn_begin(env, NULL, &txn_committed, DB_READ_COMMITTED);                          CKERR(r);
+  r = env->txn_begin(env, NULL, &txn_uncommitted, DB_READ_UNCOMMITTED);                          CKERR(r);
+
+
+  r = db->cursor(db, txn_serializable, &cursor, DB_SERIALIZABLE|DB_READ_COMMITTED); CKERR2(r, EINVAL);
+  r = db->cursor(db, txn_serializable, &cursor, DB_SERIALIZABLE|DB_READ_UNCOMMITTED); CKERR2(r, EINVAL);
+  r = db->cursor(db, txn_serializable, &cursor, DB_READ_UNCOMMITTED|DB_READ_COMMITTED); CKERR2(r, EINVAL);
+
+
+  r = db->cursor(db, txn_serializable, &cursor, 0); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_serializable, &cursor, DB_SERIALIZABLE); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_serializable, &cursor, DB_READ_COMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_serializable, &cursor, DB_READ_UNCOMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_committed, &cursor, 0); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_committed, &cursor, DB_SERIALIZABLE); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_committed, &cursor, DB_READ_COMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_committed, &cursor, DB_READ_UNCOMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_uncommitted, &cursor, 0); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_uncommitted, &cursor, DB_SERIALIZABLE); CKERR(r);
+  r = cursor->c_close(cursor); CKERR(r);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_uncommitted, &cursor, DB_READ_COMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+  r = db->cursor(db, txn_uncommitted, &cursor, DB_READ_UNCOMMITTED); CKERR2(r, EINVAL);
+  cursor = NULL;
+
+
+
+    
+  r = txn_serializable->commit(txn_serializable, 0);                                                          CKERR(r);
+  r = txn_committed->commit(txn_committed, 0);                                             CKERR(r);
+  r = txn_uncommitted->commit(txn_uncommitted, 0);                                             CKERR(r);
+
+
+
+  r = db->close(db, 0);                                                               CKERR(r);
+  r = env->close(env, 0);                                                             CKERR(r);
+    
+  return 0;
+}

src/tests/mvcc-create-table.c

+// Test that isolation works right for subtransactions.
+// In particular, check to see what happens if a subtransaction has different isolation level from its parent.
+
+#include "test.h"
+
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+int test_main (int argc, char * const argv[]) {
+    parse_args(argc, argv);
+    int r;
+    r = system("rm -rf " ENVDIR);
+    CKERR(r);
+    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                         CKERR(r);
+    env->set_errfile(env, stderr);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      CKERR(r);
+    
+    DB *db;
+
+    DB_TXN* txna = NULL;
+    DB_TXN* txnb = NULL;
+    DB_TXN* txnc = NULL;
+    DBC* c;
+    r = env->txn_begin(env, NULL, &txna, 0);                                        CKERR(r);
+
+    r = db_create(&db, env, 0);                                                     CKERR(r);
+    r = db->open(db, txna, "foo.db", NULL, DB_BTREE, DB_CREATE, 0666);              CKERR(r);
+
+    DBT key,val;
+    r = db->put(db, txna, dbt_init(&key, "a", 4), dbt_init(&val, "a", 4), 0);       CKERR(r);
+
+    r = env->txn_begin(env, NULL, &txnb, DB_TXN_SNAPSHOT);                                        CKERR(r);
+    r = env->txn_begin(env, NULL, &txnc, DB_READ_COMMITTED);                                       CKERR(r);
+    r = txna->commit(txna, 0);                                                      CKERR(r);
+
+    r = db->cursor(db, txnb, &c, 0); assert(r == TOKUDB_MVCC_DICTIONARY_TOO_NEW);
+    r = db->cursor(db, txnc, &c, 0); assert(r == TOKUDB_MVCC_DICTIONARY_TOO_NEW);
+
+
+    r = txnb->commit(txnb, 0);                                                      CKERR(r);
+    r = txnc->commit(txnc, 0);                                                      CKERR(r);
+
+    r = db->close(db, 0);                                                               CKERR(r);
+    r = env->close(env, 0);                                                             CKERR(r);
+    
+    return 0;
+}

src/tests/mvcc-many-committed.c

+// Test that isolation works right for subtransactions.
+// In particular, check to see what happens if a subtransaction has different isolation level from its parent.
+
+#include "test.h"
+
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+int test_main (int argc, char * const argv[]) {
+    parse_args(argc, argv);
+    int r;
+    r = system("rm -rf " ENVDIR);
+    CKERR(r);
+    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+    DB_ENV *env;
+    u_int32_t i = 0;
+    u_int32_t num_read_txns = 1000;
+    r = db_env_create(&env, 0);                                                         CKERR(r);
+    env->set_errfile(env, stderr);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                      CKERR(r);
+    
+    DB *db;
+
+    DB_TXN* create_txn;
+    DB_TXN* read_txns[num_read_txns];
+    DB_TXN* read_uncommitted_txn;
+    memset(read_txns, 0, sizeof(read_txns));
+
+    r = env->txn_begin(env, NULL, &create_txn, 0);                                        CKERR(r);
+
+    r = db_create(&db, env, 0);                                                     CKERR(r);
+    r = db->open(db, create_txn, "foo.db", NULL, DB_BTREE, DB_CREATE, 0666);              CKERR(r);
+    r = create_txn->commit(create_txn, 0);                                                      CKERR(r);
+
+    DBT key,val;
+
+    for (i = 0; i < num_read_txns; i++) {
+        DB_TXN* put_txn = NULL;
+        u_int32_t data = i;
+        r = env->txn_begin(env, NULL, &put_txn, DB_TXN_SNAPSHOT);
+        CKERR(r);
+        r = db->put(
+            db, 
+            put_txn, 
+            dbt_init(&key, "a", 4), 
+            dbt_init(&val, &data, 4), 
+            0
+            );       
+        CKERR(r);
+        r = put_txn->commit(put_txn, 0);
+        CKERR(r);
+        //this should read the above put
+        r = env->txn_begin(env, NULL, &read_txns[i], DB_TXN_SNAPSHOT);
+        CKERR(r);
+            
+    }
+
+    for (i = 0; i < num_read_txns; i++) {
+        DBT curr_key, curr_val;
+        memset(&curr_key, 0, sizeof(curr_key));
+        memset(&curr_val, 0, sizeof(curr_val));
+        DBC* snapshot_cursor = NULL;
+        r = db->cursor(db, read_txns[i], &snapshot_cursor, 0); CKERR(r);        
+        r = snapshot_cursor->c_get(snapshot_cursor, &curr_key, &curr_val, DB_NEXT); CKERR(r);
+        assert(((char *)(curr_key.data))[0] == 'a');
+        assert((*(u_int32_t *)(curr_val.data)) == i);
+        assert(curr_key.size == 4);
+        assert(curr_val.size == 4);
+        snapshot_cursor->c_close(snapshot_cursor);
+    }
+    {
+        DBT curr_key, curr_val;
+        memset(&curr_key, 0, sizeof(curr_key));
+        memset(&curr_val, 0, sizeof(curr_val));
+        r = env->txn_begin(env, NULL, &read_uncommitted_txn, DB_READ_UNCOMMITTED);
+        CKERR(r);
+        DBC* read_uncommitted_cursor = NULL;
+        r = db->cursor(db, read_uncommitted_txn, &read_uncommitted_cursor, 0); CKERR(r);        
+        r = read_uncommitted_cursor->c_get(
+            read_uncommitted_cursor, 
+            &curr_key, 
+            &curr_val, 
+            DB_NEXT
+            ); 
+        CKERR(r);
+        assert(((char *)(curr_key.data))[0] == 'a');
+        assert((*(u_int32_t *)(curr_val.data)) == (num_read_txns - 1));
+        assert(curr_key.size == 4);
+        assert(curr_val.size == 4);
+        read_uncommitted_cursor->c_close(read_uncommitted_cursor);
+    }
+    for (i = 0; i < num_read_txns; i++) {
+        r = read_txns[i]->commit(read_txns[i], 0);
+        CKERR(r);
+    }
+    r = read_uncommitted_txn->commit(read_uncommitted_txn, 0);
+
+    r = db->close(db, 0);                                                               CKERR(r);
+    r = env->close(env, 0);                                                             CKERR(r);
+    
+    return 0;
+}

src/tests/stress-gc.c

+// Test that isolation works right for subtransactions.
+// In particular, check to see what happens if a subtransaction has different isolation level from its parent.
+
+#include "test.h"
+
+const int envflags = DB_INIT_MPOOL|DB_CREATE|DB_THREAD |DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_TXN|DB_PRIVATE;
+
+int test_main (int argc, char * const argv[]) {
+    parse_args(argc, argv);
+    int r;
+    r = system("rm -rf " ENVDIR);
+    CKERR(r);
+    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
+    DB_ENV *env;
+    r = db_env_create(&env, 0);
+    CKERR(r);
+    env->set_errfile(env, stderr);
+    r = env->open(env, ENVDIR, envflags, S_IRWXU+S_IRWXG+S_IRWXO);
+    CKERR(r);
+
+    int max_txns = 2000;
+    int num_runs = 10000;
+    DB_TXN* txns[max_txns];
+    memset(txns, 0, sizeof(txns));
+    int num_txns = 0;
+    int i;
+
+    for (i = 0; i < num_runs; i++) {
+        int rand_num = random()%max_txns;
+        //if (i%50 == 0) {
+        //    printf("num_txns %d\n", num_txns);
+        //    printf("iteration %d\n", i);
+        //}
+        if (rand_num >= num_txns) {
+            // add a txn
+            assert(txns[num_txns] == NULL);
+            // 7 out of 8 times, it is snapshot, otherwise, serializable
+            int is_snapshot = (random() % 8 != 0);
+            r = env->txn_begin(env, NULL, &txns[num_txns], is_snapshot ? DB_TXN_SNAPSHOT : 0); 
+            CKERR(r);
+            num_txns++;
+        }
+        else {
+            // commit the txn
+            r = txns[rand_num]->commit(txns[rand_num], 0);
+            CKERR(r);
+            int j;
+            for (j = rand_num; j < num_txns-1; j++) {
+                txns[j] = txns[j+1];
+            }
+            txns[num_txns-1] = NULL;
+            num_txns--;
+        }
+    }
+
+    for (i = 0; i < num_txns; i++) {
+        r = txns[i]->commit(txns[i], 0);
+        CKERR(r);
+    }
+
+    r = env->close(env, 0);
+    CKERR(r);
+    
+    return 0;
+}

src/tests/test944.c

-/* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
-#include "test.h"
-#include <db.h>
-#include <sys/stat.h>
-
-unsigned char N=8;
-
-static int
-fact (int n) {
-    if (n<=2) return n;
-    else return n*fact(n-1);
-}
-
-static void
-swapc (unsigned char *a, unsigned char *b) {
-    unsigned char tmp=*a;
-    *a=*b;
-    *b=tmp;
-}
-
-DB_ENV *env;
-DB *db;
-
-static void
-run (int choice) {
-    unsigned char v[N];
-    int i;
-    int r;
-    for (i=0; i<N; i++) {
-	v[i]=(unsigned char)i;
-    }
-    for (i=0; i<N; i++) {
-	int nchoices=N-i;
-	swapc(&v[i], &v[i+choice%nchoices]);
-	choice=choice/nchoices;
-    }
-    if (0) {
-	for (i=0; i<N; i++) {
-	    printf("%d ", v[i]);
-	}
-
-	printf("\n");
-    }
-    DB_TXN *txn;
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	for (i=0; i<N; i+=2) {
-	    DBT kdbt,vdbt;
-	    char key=v[i];
-	    char val=v[i+1];
-	    r=db->put(db, txn, dbt_init(&kdbt, &key, 1), dbt_init(&vdbt, &val, 1), DB_YESOVERWRITE); CKERR(r);
-	}
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);	
-	DBC *c;
-	r=db->cursor(db, txn, &c, 0);                                 CKERR(r);
-	DBT kdbt,vdbt;
-	dbt_init(&kdbt, 0, 0);
-	dbt_init(&vdbt, 0, 0);
-	i=0;
-	while (0==(r=c->c_get(c, &kdbt, &vdbt, DB_NEXT))) {
-	    //printf("Got %d %d\n", *(unsigned char*)kdbt.data, *(unsigned char*)vdbt.data);
-	    i++;
-	}
-	CKERR2(r, DB_NOTFOUND);
-	//printf("i=%d N=%d\n", i, N);
-	assert(i==N/2);
-	r=c->c_close(c);                                                          CKERR(r);
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	for (i=0; i<N; i+=2) {
-	    DBT kdbt;
-	    char key=v[i];
-	    r=db->del(db, txn, dbt_init(&kdbt, &key, 1), 0);                      CKERR(r);
-	    //printf(" del %d\n", key);
-	}
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    return;
-#if 0
-    char v101=101, v102=102, v1=1, v2=2;
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	DBT k,v;
-	r=db->put(db, txn, dbt_init(&k, &v1, 1), dbt_init(&v, &v101, 1), 0); CKERR(r);
-	r=db->put(db, txn, dbt_init(&k, &v2, 1), dbt_init(&v, &v102, 1), 0); CKERR(r);
-	r=txn->commit(txn, 0);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	DBC *c;
-	r=db->cursor(db, txn, &c, 0);                                 CKERR(r);
-	DBT k,v;
-	r=c->c_get(c, dbt_init(&k, 0, 0), dbt_init(&v, 0, 0), DB_FIRST); CKERR(r);
-	assert(*(char*)k.data==v1); assert(*(char*)v.data==v101);
-	r=c->c_get(c, dbt_init(&k, 0, 0), dbt_init(&v, 0, 0), DB_NEXT);  CKERR(r);
-	assert(*(char*)k.data==v2); assert(*(char*)v.data==v102);
-	r=c->c_get(c, dbt_init(&k, 0, 0), dbt_init(&v, 0, 0), DB_NEXT);  assert(r!=0);
-	r=c->c_close(c);                                                   CKERR(r);
-	r=txn->commit(txn, 0);                                             CKERR(r);
-    }
-#endif
-}
-
-int
-test_main(int argc, char *const argv[]) {
-    parse_args(argc, argv);
-
-    int r;
-    r = system("rm -rf " ENVDIR);
-    CKERR(r);
-    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
-
-    DB_TXN *txn;
-    {
-        r = db_env_create(&env, 0);                                   CKERR(r);
-#ifdef USE_TDB
-	r = env->set_redzone(env, 0);                                 CKERR(r);
-#endif
-	r=env->open(env, ENVDIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_CREATE|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
-	env->set_errfile(env, stderr);
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	r=db_create(&db, env, 0);                                     CKERR(r);
-	r=db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO);  CKERR(r);
-	r=txn->commit(txn, 0);                                        CKERR(r);
-    }
-    int i;
-    //printf("fact(%d)=%d\n", N, fact(N));
-    for (i=0; i<fact(N); i++) {
-	run(i);
-    }
-    {
-	r=db->close(db, 0);                                           CKERR(r);
-	r=env->close(env, 0);                                         CKERR(r);
-    }
-
-    return 0;
-}

src/tests/test944a.c

-/* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "Copyright (c) 2007 Tokutek Inc.  All rights reserved."
-#include "test.h"
-#include <db.h>
-#include <sys/stat.h>
-
-unsigned char N=8;
-
-static int
-fact (int n) {
-    if (n<=2) return n;
-    else return n*fact(n-1);
-}
-
-static void
-swapc (unsigned char *a, unsigned char *b) {
-    unsigned char tmp=*a;
-    *a=*b;
-    *b=tmp;
-}
-
-DB_ENV *env;
-DB *db;
-
-static void
-run (int choice) {
-    unsigned char v[N];
-    int i;
-    int r;
-    for (i=0; i<N; i++) {
-	v[i]=(char)i;
-    }
-    for (i=0; i<N; i++) {
-	int nchoices=N-i;
-	swapc(&v[i], &v[i+choice%nchoices]);
-	choice=choice/nchoices;
-    }
-    if (0) {
-	for (i=0; i<N; i++) {
-	    printf("%d ", v[i]);
-	}
-
-	printf("\n");
-    }
-    DB_TXN *txn;
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	for (i=0; i<N; i+=2) {
-	    DBT kdbt,vdbt;
-	    char key=v[i];
-	    char val=v[i+1];
-	    r=db->put(db, txn, dbt_init(&kdbt, &key, 1), dbt_init(&vdbt, &val, 1), DB_YESOVERWRITE); CKERR(r);
-	}
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);	
-	DBC *c;
-	r=db->cursor(db, txn, &c, 0);                                 CKERR(r);
-	DBT kdbt,vdbt;
-	dbt_init(&kdbt, 0, 0);
-	dbt_init(&vdbt, 0, 0);
-	i=0;
-	while (0==(r=c->c_get(c, &kdbt, &vdbt, DB_NEXT))) {
-	    i++;
-	}
-	CKERR2(r, DB_NOTFOUND);
-	assert(i==N/2);
-	r=c->c_close(c);                                                          CKERR(r);
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	for (i=0; i<N; i+=2) {
-	    DBT kdbt;
-	    char key=v[i];
-	    r=db->del(db, txn, dbt_init(&kdbt, &key, 1), 0);                      CKERR(r);
-	}
-	r=txn->commit(txn, DB_TXN_NOSYNC);                                        CKERR(r);
-    }
-    return;
-#if 0
-    char v101=101, v102=102, v1=1, v2=2;
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	DBT k,v;
-	r=db->put(db, txn, dbt_init(&k, &v1, 1), dbt_init(&v, &v101, 1), 0); CKERR(r);
-	r=db->put(db, txn, dbt_init(&k, &v2, 1), dbt_init(&v, &v102, 1), 0); CKERR(r);
-	r=txn->commit(txn, 0);                                        CKERR(r);
-    }
-    {
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	DBC *c;
-	r=db->cursor(db, txn, &c, 0);                                 CKERR(r);
-	DBT k,v;
-	r=c->c_get(c, dbt_init_malloc(&k), dbt_init_malloc(&v), DB_FIRST); CKERR(r);
-	assert(*(char*)k.data==v1); assert(*(char*)v.data==v101);
-	r=c->c_get(c, dbt_init_malloc(&k), dbt_init_malloc(&v), DB_NEXT);  CKERR(r);
-	assert(*(char*)k.data==v2); assert(*(char*)v.data==v102);
-	r=c->c_get(c, dbt_init_malloc(&k), dbt_init_malloc(&v), DB_NEXT);  assert(r!=0);
-	r=c->c_close(c);                                                   CKERR(r);
-	r=txn->commit(txn, 0);                                             CKERR(r);
-    }
-#endif
-}
-
-int
-test_main(int argc, char *const argv[]) {
-    parse_args(argc, argv);
-
-    int r;
-    r = system("rm -rf " ENVDIR);
-    CKERR(r);
-    toku_os_mkdir(ENVDIR, S_IRWXU+S_IRWXG+S_IRWXO);
-
-    DB_TXN *txn;
-    {
-        r = db_env_create(&env, 0);                                   CKERR(r);
-#ifdef USE_TDB
-	r = env->set_redzone(env, 0);                                 CKERR(r);
-#endif
-	r=env->open(env, ENVDIR, DB_INIT_LOCK|DB_INIT_LOG|DB_INIT_MPOOL|DB_INIT_TXN|DB_CREATE|DB_PRIVATE, S_IRWXU+S_IRWXG+S_IRWXO); CKERR(r);
-	env->set_errfile(env, stderr);
-	r=env->txn_begin(env, 0, &txn, 0);                            CKERR(r);
-	r=db_create(&db, env, 0);                                     CKERR(r);
-	r=db->open(db, txn, "foo.db", 0, DB_BTREE, DB_CREATE, S_IRWXU+S_IRWXG+S_IRWXO);  CKERR(r);
-	r=txn->commit(txn, 0);                                        CKERR(r);
-    }
-    int i;
-    for (i=0; i<fact(N); i++) {
-	run(i);
-    }
-    {
-	r=db->close(db, 0);                                           CKERR(r);
-	r=env->close(env, 0);                                         CKERR(r);
-    }
-
-    return 0;
-}

src/ydb-internal.h

@@ -190,11 +190,19 @@ int toku_env_is_panicked(DB_ENV *dbenv);
 void toku_locked_env_err(const DB_ENV * env, int error, const char *fmt, ...) 
                          __attribute__((__format__(__printf__, 3, 4)));
 
+typedef enum __toku_isolation_level { 
+    TOKU_ISO_SERIALIZABLE=0,
+    TOKU_ISO_SNAPSHOT=1,
+    TOKU_ISO_READ_COMMITTED=2, 
+    TOKU_ISO_READ_UNCOMMITTED=3
+} TOKU_ISOLATION;
+
 struct __toku_db_txn_internal {
     //TXNID txnid64; /* A sixty-four bit txn id. */
     struct tokutxn *tokutxn;
     struct __toku_lth *lth;  //Hash table holding list of dictionaries this txn has touched
     u_int32_t flags;
+    TOKU_ISOLATION iso;
     DB_TXN *child;
     struct toku_list dbs_that_must_close_before_abort;
 };
@@ -202,6 +210,7 @@ struct __toku_db_txn_internal {
 struct __toku_dbc_internal {
     struct brt_cursor *c;
     DB_TXN *txn;
+    TOKU_ISOLATION iso;
     struct simple_dbt skey_s,sval_s;
     struct simple_dbt *skey,*sval;
 };

toku_include/toku_portability.h

@@ -39,6 +39,7 @@ extern "C" {
 #define __attribute__(x)      /* Nothing */
 #endif
 
+#include <malloc.h>
 #include "toku_stdint.h"
 
 #ifndef TOKU_OFF_T_DEFINED
@@ -63,6 +64,7 @@ typedef int64_t toku_off_t;
 #if defined(__ICC)
 // Intel linux
 
+#include <alloca.h>
 #include <toku_stdint.h>
 #include <unistd.h>
 #include <sys/types.h>
@@ -86,6 +88,7 @@ typedef int64_t toku_off_t;
 #if __FreeBSD__
 #include <stdarg.h>
 #endif
+#include <alloca.h>
 
 #define cast_to_typeof(v) (__typeof__(v))
 #else

windows/file.c

@@ -600,3 +600,11 @@ ftello(FILE *stream) {
     return offset;
 }
 
+ssize_t 
+toku_os_read(int fd, void *buf, size_t count) {
+    return read(fd, buf, count);
+}
+
+
+
+