Merge #2623 changes onto main line. The new test fails, but I need to get a...

Merge #2623 changes onto main line.  The new test fails, but I need to get a merge from main again.  Refs #2623. [t:2623]
{{{
svn merge -r 20348:20378 https://svn.tokutek.com/tokudb/toku/tokudb.2623
}}}
.


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

newbrt/brt-internal.h

@@ -18,10 +18,9 @@
 #include "leafentry.h"
 #include "block_table.h"
 #include "leaflock.h"
+#include "c_dialects.h"
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#endif
+C_BEGIN
 
 #ifndef BRT_FANOUT
 #define BRT_FANOUT 16
@@ -365,9 +364,6 @@ int toku_db_badformat(void);
 int toku_brt_remove_on_commit(TOKUTXN child, DBT* iname_dbt_p);
 int toku_brt_remove_now(CACHETABLE ct, DBT* iname_dbt_p);
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-};
-#endif
-
+C_END
 
 #endif

newbrt/brt.h

@@ -13,10 +13,9 @@
 #include "cachetable.h"
 #include "log.h"
 #include "brt-search.h"
+#include "c_dialects.h"
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#endif
+C_BEGIN
 
 // A callback function is invoked with the key, and the data.
 // The pointers (to the bytevecs) must not be modified.  The data must be copied out before the callback function returns.
@@ -230,8 +229,6 @@ BOOL toku_brt_is_recovery_logging_suppressed (BRT);
 #define TOKU_MULTIPLE_MAIN_THREADS 0
 #endif
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-};
-#endif
+C_END
 
 #endif

newbrt/brtloader-internal.h

 /* -*- mode: C; c-basic-offset: 4 -*- */
 #ifndef _TOKU_BRTLOADER_INTERNAL_H
 #define _TOKU_BRTLOADER_INTERNAL_H
-#ident "$Id: pqueue.c$"
+#ident "$Id$"
 #ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
 
 #include <db.h>
@@ -9,10 +9,10 @@
 #include "brtloader.h"
 #include "queue.h"
 #include "toku_pthread.h"
+#include "dbufio.h"
+#include "c_dialects.h"
 
-#if defined(__cplusplus)
-extern "C" {
-#endif
+C_BEGIN
 
 /* These functions are exported to allow the tests to compile. */
 
@@ -186,18 +186,17 @@ int merge_row_arrays_base (struct row dest[/*an+bn*/], struct row a[/*an*/], int
 
 int merge_files (struct merge_fileset *fs, BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func, int progress_allocation, QUEUE);
 
-#if defined(__cilkplusplus)
-extern "Cilk++" {
-#endif
+CILK_BEGIN
 int sort_and_write_rows (struct rowset rows, struct merge_fileset *fs, BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func,
 			 int progress_allocation);
 
 int mergesort_row_array (struct row rows[/*n*/], int n, int which_db, DB *dest_db, brt_compare_func, BRTLOADER, struct rowset *);
 
+int toku_merge_some_files_using_dbufio (const BOOL to_q, FIDX dest_data, QUEUE q, int n_sources, DBUFIO_FILESET bfs, FIDX srcs_fidxs[/*n_sources*/], BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func compare, int progress_allocation);
+
 //int write_file_to_dbfile (int outfile, FIDX infile, BRTLOADER bl, const struct descriptor *descriptor, int progress_allocation);
-#if defined(__cilkplusplus)
-};
-#endif
+CILK_END
+
 int brt_loader_sort_and_write_rows (struct rowset *rows, struct merge_fileset *fs, BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func,
 				    int progress_allocation);
 
@@ -217,9 +216,17 @@ int brtloader_fi_close (struct file_infos *fi, FIDX idx);
 int brtloader_fi_reopen (struct file_infos *fi, FIDX idx, const char *mode);
 int brtloader_fi_unlink (struct file_infos *fi, FIDX idx);
 
-#if defined(__cplusplus)
-};
-#endif
-
+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 struct descriptor *descriptors[/*N*/],
+				   const char *new_fnames_in_env[/*N*/],
+				   brt_compare_func bt_compare_functions[/*N*/],
+				   const char *temp_file_template,
+				   LSN load_lsn);
+
+C_END
 
 #endif

newbrt/brtloader.c

@@ -24,6 +24,7 @@
 #include "pqueue.h"
 #include "trace_mem.h"
 #include "dbufio.h"
+#include "c_dialects.h"
 
 // to turn on tracing, 
 //   cd .../newbrt
@@ -35,8 +36,6 @@
 #include <cilk.h>
 #include <cilk_mutex.h>
 #include <fake_mutex.h>
-#define CILK_BEGIN extern "Cilk++" {
-#define CILK_END };
 #else
 // maybe #include <cilk_stub.h>
 #if !defined(CILK_STUB)
@@ -45,17 +44,10 @@
 #define cilk_sync
 #define cilk_for for
 #endif
-#define CILK_BEGIN
-#define CILK_END
 #endif
 
 // mark everything as C and selectively mark cilk functions
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#if 0
-} // make emacs happy about matching parens....
-#endif
-#endif
+C_BEGIN
 
 static size_t (*os_fwrite_fun)(const void *,size_t,size_t,FILE*)=NULL;
 void brtloader_set_os_fwrite (size_t (*fwrite_fun)(const void*,size_t,size_t,FILE*)) {
@@ -360,28 +352,17 @@ static int merge_fanin (BRTLOADER bl)
     return result;
 }
 
-// LAZY cleanup on error paths, ticket #2591
-int toku_brt_loader_open (/* out */ BRTLOADER *blp,
-                          CACHETABLE cachetable,
-			  generate_row_for_put_func g,
-			  DB *src_db,
-			  int N, DB*dbs[/*N*/],
-			  const struct descriptor *descriptors[/*N*/],
-			  const char *new_fnames_in_env[/*N*/],
-			  brt_compare_func bt_compare_functions[/*N*/],
-			  const char *temp_file_template,
-                          LSN load_lsn)
-/* Effect: called by DB_ENV->create_loader to create a brt loader.
- * Arguments:
- *   blp                  Return the brt loader here.
- *   g                    The function for generating a row
- *   src_db               The source database.  Needed by g.  May be NULL if that's ok with g.
- *   N                    The number of dbs to create.
- *   dbs                  An array of open databases.  Used by g.  The data will be put in these database.
- *   new_fnames           The file names (these strings are owned by the caller: we make a copy for our own purposes).
- *   temp_file_template   A template suitable for mkstemp()
- * Return value: 0 on success, an error number otherwise.
- */
+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 struct descriptor *descriptors[/*N*/],
+				   const char *new_fnames_in_env[/*N*/],
+				   brt_compare_func bt_compare_functions[/*N*/],
+				   const char *temp_file_template,
+				   LSN load_lsn)
+// Effect: Allocate and initialize a BRTLOADER, but do not create the extractor thread.
 {
     BRTLOADER CALLOC(bl); // initialized to all zeros (hence CALLOC)
     if (!bl) return errno;
@@ -463,19 +444,59 @@ int toku_brt_loader_open (/* out */ BRTLOADER *blp,
         int r = toku_pthread_mutex_init(&bl->mutex, NULL); 
         if (r != 0) { brtloader_destroy(bl, TRUE); return r; }
     }
-    {   
+
+    bl->extractor_live = TRUE;
+
+    *blp = bl;
+
+    return 0;
+}
+
+// LAZY cleanup on error paths, ticket #2591
+int toku_brt_loader_open (/* out */ BRTLOADER *blp,
+                          CACHETABLE cachetable,
+			  generate_row_for_put_func g,
+			  DB *src_db,
+			  int N, DB*dbs[/*N*/],
+			  const struct descriptor *descriptors[/*N*/],
+			  const char *new_fnames_in_env[/*N*/],
+			  brt_compare_func bt_compare_functions[/*N*/],
+			  const char *temp_file_template,
+                          LSN load_lsn)
+/* Effect: called by DB_ENV->create_loader to create a brt loader.
+ * Arguments:
+ *   blp                  Return the brt loader here.
+ *   g                    The function for generating a row
+ *   src_db               The source database.  Needed by g.  May be NULL if that's ok with g.
+ *   N                    The number of dbs to create.
+ *   dbs                  An array of open databases.  Used by g.  The data will be put in these database.
+ *   new_fnames           The file names (these strings are owned by the caller: we make a copy for our own purposes).
+ *   temp_file_template   A template suitable for mkstemp()
+ * Return value: 0 on success, an error number otherwise.
+ */
+{
+    int result = 0;
+    {
+	int r = toku_brt_loader_internal_init(blp, cachetable, g, src_db,
+					      N, dbs,
+					      descriptors,
+					      new_fnames_in_env,
+					      bt_compare_functions,
+					      temp_file_template,
+					      load_lsn);
+	if (r!=0) result = r;
+    }
+    if (result==0) {
+	BRTLOADER bl = *blp;
         int r = toku_pthread_create(&bl->extractor_thread, NULL, extractor_thread, (void*)bl); 
         if (r!=0) { 
+	    result = r;
             toku_pthread_mutex_destroy(&bl->mutex);
             brtloader_destroy(bl, TRUE);
-            return r; 
         }
     }
-    bl->extractor_live = TRUE;
-
-    *blp = bl;
     BL_TRACE(blt_open);
-    return 0;
+    return result;
 }
 
 static void brt_loader_set_panic(BRTLOADER bl, int error) {
@@ -1384,7 +1405,7 @@ int brt_loader_sort_and_write_rows (struct rowset *rows, struct merge_fileset *f
 #endif
 }
 
-static int merge_some_files_using_dbufio (const BOOL to_q, FIDX dest_data, QUEUE q, int n_sources, DBUFIO_FILESET bfs, FIDX srcs_fidxs[/*n_sources*/], BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func compare, int progress_allocation)
+int toku_merge_some_files_using_dbufio (const BOOL to_q, FIDX dest_data, QUEUE q, int n_sources, DBUFIO_FILESET bfs, FIDX srcs_fidxs[/*n_sources*/], BRTLOADER bl, int which_db, DB *dest_db, brt_compare_func compare, int progress_allocation)
 /* Effect: Given an array of FILE*'s each containing sorted, merge the data and write it to an output.  All the files remain open after the merge.
  *   This merge is performed in one pass, so don't pass too many files in.  If you need a tree of merges do it elsewhere.
  *   If TO_Q is true then we write rowsets into queue Q.  Otherwise we write into dest_data.
@@ -1571,7 +1592,7 @@ static int merge_some_files (const BOOL to_q, FIDX dest_data, QUEUE q, int n_sou
     }
 	
     if (result==0) {
-	int r = merge_some_files_using_dbufio (to_q, dest_data, q, n_sources, bfs, srcs_fidxs, bl, which_db, dest_db, compare, progress_allocation);
+	int r = toku_merge_some_files_using_dbufio (to_q, dest_data, q, n_sources, bfs, srcs_fidxs, bl, which_db, dest_db, compare, progress_allocation);
 	if (r!=0) { result = r; }
     }
 
@@ -2955,6 +2976,5 @@ int brt_loader_write_file_to_dbfile (int outfile, FIDX infile, BRTLOADER bl, con
 }
 #endif
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-}
-#endif
+C_END
+

newbrt/brtloader.h

@@ -6,9 +6,9 @@
 #ident "Copyright (c) 2007-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."
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#endif
+#include "c_dialects.h"
+
+C_BEGIN
 
 // The loader callbacks are C functions and need to be defined as such
 
@@ -51,8 +51,6 @@ int toku_brt_loader_finish_extractor(BRTLOADER bl);
 
 int toku_brt_loader_get_error(BRTLOADER bl, int *loader_errno);
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-}
-#endif
+C_END
 
 #endif // BRTLOADER_H

newbrt/c_dialects.h

+#ifndef C_DIALECTS_H
+#define C_DIALECTS_H
+#ident "$Id$"
+#ident "Copyright (c) 2007-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."
+
+#if defined(__cplusplus) || defined(__cilkplusplus)
+#define C_BEGIN extern "C" {
+#define C_END };
+#else
+#define C_BEGIN
+#define C_END
+#endif
+
+
+#if defined(__cilkplusplus)
+#define CILK_BEGIN extern "Cilk++" {
+#define CILK_END };
+#else
+#define CILK_BEGIN
+#define CILK_END
+#endif
+
+#endif

newbrt/dbufio.h

@@ -6,10 +6,9 @@
 
 #include <toku_portability.h>
 #include <toku_pthread.h>
+#include "c_dialects.h"
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#endif
+C_BEGIN
 
 /* Maintain a set of files for reading, with double buffering for the reads. */
 
@@ -23,9 +22,6 @@ int destroy_dbufio_fileset(DBUFIO_FILESET);
 
 int dbufio_fileset_read (DBUFIO_FILESET bfs, int filenum, void *buf_v, size_t count, size_t *n_read);
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-};
-#endif
-
+C_END
 
 #endif

newbrt/queue.h

@@ -6,10 +6,9 @@
 #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 "brttypes.h"
+#include "c_dialects.h"
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-extern "C" {
-#endif
+C_BEGIN
 
 // The abstraction:
 //
@@ -54,7 +53,5 @@ int queue_destroy (QUEUE q);
 // Requires: The queue must be empty and no consumer should try to dequeue after this (one way to do this is to make sure the consumer saw EOF).
 // Returns 0 on success.   If the queue is not empty, returns EINVAL.  Other errors are likely to be bad (some sort of mutex or condvar failure).
 
-#if defined(__cplusplus) || defined(__cilkplusplus)
-};
-#endif
+C_END
 #endif

newbrt/tests/Makefile

@@ -92,6 +92,8 @@ check_test-assert$(BINSUF): test-assert$(BINSUF) $(PTHREAD_LOCAL)
 	@# one argument, "ok" should not error
 	$(VGRIND) ./$< ok $(SUMMARIZE_CMD)
 
+check_brtloader-test-merge-files-dbufio: EXTRA_ARGS=dir.brtloader-test-merge-files-dbufio
+
 check_brtloader-test$(BINSUF): EXTRA_ARGS=dir.brtloader-test
 
 check_brtloader-test-bad-generate$(BINSUF): EXTRA_ARGS=dir.brtloader-test-bad-generate

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

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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-extractor-errors.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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-merge-files-dbufio.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."
+
+// test the loader write dbfile function
+
+#define DONT_DEPRECATE_WRITES
+#define DONT_DEPRECATE_MALLOC
+#include "includes.h"
+#include "test.h"
+#include "brtloader-internal.h"
+
+C_BEGIN
+
+static int event_count, event_count_trigger;
+
+static void reset_event_counts(void) {
+    event_count = event_count_trigger = 0;
+}
+
+static void event_hit(void) {
+}
+
+static int loader_poll_callback(void *UU(extra), float UU(progress)) {
+    int r;
+    event_count++;
+    if (event_count_trigger == event_count) {
+        event_hit();
+        r = 1;
+    } else {
+        r = 0;
+    }
+    return r;
+}
+
+static size_t bad_fwrite (const void *ptr, size_t size, size_t nmemb, FILE *stream) {
+    event_count++;
+    size_t r;
+    if (event_count_trigger == event_count) {
+        event_hit();
+	errno = ENOSPC;
+	r = -1;
+    } else {
+	r = fwrite(ptr, size, nmemb, stream);
+	if (r!=nmemb) {
+	    errno = ferror(stream);
+	}
+    }
+    return r;
+}
+
+static ssize_t bad_write(int fd, const void * bp, size_t len) {
+    ssize_t r;
+    event_count++;
+    if (event_count_trigger == event_count) {
+        event_hit();
+	errno = ENOSPC;
+	r = -1;
+    } else {
+	r = write(fd, bp, len);
+    }
+    return r;
+}
+
+static ssize_t bad_pwrite(int fd, const void * bp, size_t len, toku_off_t off) {
+    ssize_t r;
+    event_count++;
+    if (event_count_trigger == event_count) {
+        event_hit();
+	errno = ENOSPC;
+	r = -1;
+    } else {
+	r = pwrite(fd, bp, len, off);
+    }
+    return r;
+}
+
+static int my_malloc_event = 0;
+static int my_malloc_count = 0, my_big_malloc_count = 0;
+
+static void reset_my_malloc_counts(void) {
+    my_malloc_count = my_big_malloc_count = 0;
+}
+
+static void *my_malloc(size_t n) {
+    void *caller = __builtin_return_address(0);
+    if (!((void*)toku_malloc <= caller && caller <= (void*)toku_free))
+        goto skip;
+    my_malloc_count++;
+    if (n >= 64*1024) {
+        my_big_malloc_count++;
+        if (my_malloc_event) {
+            caller = __builtin_return_address(1);
+            if ((void*)toku_xmalloc <= caller && caller <= (void*)toku_malloc_report)
+                goto skip;
+            event_count++;
+            if (event_count == event_count_trigger) {
+                event_hit();
+                errno = ENOMEM;
+                return NULL;
+            }
+        }
+    }
+ skip:
+    return malloc(n);
+}
+
+static int qsort_compare_ints (const void *a, const void *b) {
+    int avalue = *(int*)a;
+    int bvalue = *(int*)b;
+    if (avalue<bvalue) return -1;
+    if (avalue>bvalue) return +1;
+    return 0;
+
+}
+
+static int compare_ints (DB *dest_db, const DBT *akey, const DBT *bkey) {
+    assert(dest_db==NULL);
+    assert(akey->size==sizeof(int));
+    assert(bkey->size==sizeof(int));
+    return qsort_compare_ints(akey->data, bkey->data);
+}
+
+static void err_cb(DB *db UU(), int dbn UU(), int err UU(), DBT *key UU(), DBT *val UU(), void *extra UU()) {
+    fprintf(stderr, "error in test");
+    abort();
+}
+
+enum { N_SOURCES = 2, N_RECORDS=10, N_DEST_DBS=1 };
+
+static char *make_fname(const char *directory, const char *fname, int idx) {
+    int len = strlen(directory)+strlen(fname)+20;
+    char *XMALLOC_N(len, result);
+    int r = snprintf(result, len, "%s%s%d", directory, fname, idx);
+    assert(r<len);
+    return result; // don't care that it's a little too long.
+}
+
+
+
+static void test (const char *directory) {
+
+    int *XMALLOC_N(N_SOURCES, fds);
+
+    char **XMALLOC_N(N_SOURCES, fnames);
+    for (int i=0; i<N_SOURCES; i++) {
+	fnames[i] = make_fname(directory, "temp", i);
+	fds[i] = open(fnames[0], O_CREAT|O_RDWR, S_IRWXU);
+	assert(fds[i]>=0);
+    }
+    for (int i=0; i<N_RECORDS; i++) {
+	int size=4;
+	int fdi = random()%N_SOURCES;
+	int fd  = fds[fdi];
+	{ int r = write(fd, &size, 4);  assert(r==4); }
+	{ int r = write(fd, &i,    4);  assert(r==4); }
+	{ int r = write(fd, &size, 4);  assert(r==4); }
+	{ int r = write(fd, &i,    4);  assert(r==4); }
+    }
+    for (int i=0; i<N_SOURCES; i++) {
+	off_t r = lseek(fds[i], 0, SEEK_SET);
+	assert(r==0);
+    }
+
+    toku_set_func_malloc(my_malloc);
+    brtloader_set_os_fwrite(bad_fwrite);
+    toku_set_func_write(bad_write);
+    toku_set_func_pwrite(bad_pwrite);
+
+    BRTLOADER bl;
+    DB **XMALLOC_N(N_DEST_DBS, dbs);
+    const struct descriptor **XMALLOC_N(N_DEST_DBS, descriptors);
+    const char **XMALLOC_N(N_DEST_DBS, new_fnames_in_env);
+    brt_compare_func *XMALLOC_N(N_DEST_DBS, bt_compare_functions);
+    bt_compare_functions[0] = compare_ints;
+    CACHETABLE ct;
+    enum {CACHETABLE_SIZE = 64*1024};
+    {
+	int r = toku_create_cachetable(&ct, CACHETABLE_SIZE, (LSN){1}, NULL);
+	assert(r==0);
+    }
+    LSN *XMALLOC(lsnp);
+    {
+	int r = toku_brt_loader_internal_init (&bl,
+					       ct,
+					       (generate_row_for_put_func)NULL,
+					       (DB*)NULL,
+					       N_DEST_DBS, dbs,
+					       descriptors,
+					       new_fnames_in_env,
+					       bt_compare_functions,
+					       (const char *)NULL, // temp_filetemplate
+					       *lsnp);
+	assert(r==0);
+    }
+
+    brt_loader_init_error_callback(&bl->error_callback);
+    brt_loader_set_error_function(&bl->error_callback, err_cb, NULL);
+    brt_loader_init_poll_callback(&bl->poll_callback);
+
+    brt_loader_set_poll_function(&bl->poll_callback, loader_poll_callback, NULL);
+
+    QUEUE q;
+    { int r = queue_create(&q, 1000); assert(r==0); }
+    DBUFIO_FILESET bfs;
+    const int MERGE_BUF_SIZE = 100000; // bigger than 64K so that we will trigger malloc issues.
+    { int r = create_dbufio_fileset(&bfs, N_SOURCES, fds, MERGE_BUF_SIZE);  assert(r==0); }
+    FIDX *XMALLOC_N(N_SOURCES, src_fidxs);
+    {
+	int r = toku_merge_some_files_using_dbufio(TRUE, FIDX_NULL, q, N_SOURCES, bfs, src_fidxs, bl, 0, (DB*)NULL, compare_ints, 10000);
+	assert(r==0);
+    }
+}
+
+
+static int usage(const char *progname, int n) {
+    fprintf(stderr, "Usage:\n %s [-v] [-q] [-r %d] [-s] [-m] directory\n", progname, n);
+    fprintf(stderr, "[-v] turn on verbose\n");
+    fprintf(stderr, "[-q] turn off verbose\n");
+    fprintf(stderr, "[-r %d] set the number of rows\n", n);
+    fprintf(stderr, "[-s] set the small loader size factor\n");
+    fprintf(stderr, "[-m] inject big malloc failures\n");
+    return 1;
+}
+
+int test_main (int argc, const char *argv[]) {
+    const char *progname=argv[0];
+    int n = 1;
+    argc--; argv++;
+    while (argc>0) {
+        if (strcmp(argv[0],"-h")==0) {
+            return usage(progname, n);
+	} else if (strcmp(argv[0],"-v")==0) {
+	    verbose=1;
+	} else if (strcmp(argv[0],"-q")==0) {
+	    verbose=0;
+        } else if (strcmp(argv[0],"-r") == 0) {
+            argc--; argv++;
+            n = atoi(argv[0]);
+        } else if (strcmp(argv[0],"-s") == 0) {
+            toku_brtloader_set_size_factor(1);
+        } else if (strcmp(argv[0],"-m") == 0) {
+            my_malloc_event = 1;
+	} else if (argc!=1) {
+            return usage(progname, n);
+	}
+        else {
+            break;
+        }
+	argc--; argv++;
+    }
+    assert(argc==1); // argv[1] is the directory in which to do the test.
+    const char* directory = argv[0];
+    char unlink_all[strlen(directory)+20];
+    snprintf(unlink_all, strlen(directory)+20, "rm -rf %s", directory);
+
+    int  templen = strlen(directory)+15;
+    char template[templen];
+    int tlen = snprintf(template, templen, "%s/tempXXXXXX", directory);
+    assert (tlen>0 && tlen<templen);
+
+    char output_name[templen];
+    int  olen = snprintf(output_name, templen, "%s/test.tokudb", directory);
+    assert (olen>0 && olen<templen);
+
+    // callibrate
+    int r;
+    r = system(unlink_all); CKERR(r);
+    r = toku_os_mkdir(directory, 0755); CKERR(r);
+    test(directory);
+
+    if (verbose) printf("my_malloc_count=%d big_count=%d\n", my_malloc_count, my_big_malloc_count);
+
+    int event_limit = event_count;
+    if (verbose) printf("event_limit=%d\n", event_limit);
+
+    for (int i = 1; i <= event_limit; i++) {
+        reset_event_counts();
+        reset_my_malloc_counts();
+        event_count_trigger = i;
+        r = system(unlink_all); CKERR(r);
+        r = toku_os_mkdir(directory, 0755); CKERR(r);
+
+	test(directory);
+    }
+
+    return 0;
+}
+
+C_END

newbrt/tests/brtloader-test-open.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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/pqueue-test.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
-#ident "$Id: pqueue.c$"
+#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."