close[t:4269] Merge 4269 fixes to main. {{{svn merge -r38251:38424...

close[t:4269] Merge 4269 fixes to main. {{{svn merge -r38251:38424 ../tokudb.4269}}}.  Closes #4269.

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

db-benchmark-test/db-benchmark-test.c

@@ -29,6 +29,8 @@ int which;
 enum { SERIAL_SPACING = 1<<6 };
 enum { DEFAULT_ITEMS_TO_INSERT_PER_ITERATION = 1<<20 };
 enum { DEFAULT_ITEMS_PER_TRANSACTION = 1<<14 };
+#define DEFAULT_N_ITEMS (1LL<<22)
+#define DEFAULT_N_ITERATIONS (DEFAULT_N_ITEMS/DEFAULT_ITEMS_TO_INSERT_PER_ITERATION)
 
 static void insert (long long v);
 #define CKERR(r) do { if (r!=0) fprintf(stderr, "%s:%d error %d %s\n", __FILE__, __LINE__, r, db_strerror(r)); assert(r==0); } while (0)
@@ -73,6 +75,8 @@ static int redzone = 0;
 static int redzone_set = 0;
 static int do_optimize = 0;
 static int unique_checks = 0;
+static long long n_iterations =  DEFAULT_N_ITERATIONS;
+
 
 static int use_random = 0;
 enum { MAX_RANDOM_C = 16000057 }; // prime-numbers.org
@@ -401,8 +405,8 @@ static void trace(const char *UU(s), int UU(n)) {
 
 static void insert (long long v) {
     int r;
-    unsigned char kc[keysize];
-    unsigned char vc[valsize];;
+    unsigned char *kc = toku_malloc(keysize);
+    unsigned char *vc = toku_malloc(valsize);
     DBT  kt, vt;
     fill_array(kc, keysize);
     long_long_to_array(kc, keysize, v); // Fill in the array first, then write the long long in.
@@ -451,6 +455,8 @@ static void insert (long long v) {
 	}
 	n_insertions_since_txn_began++;
     }
+    toku_free(kc);
+    toku_free(vc);
 }
 
 static void serial_insert_from (long long from) {
@@ -516,10 +522,6 @@ static void biginsert (long long n_elements, struct timeval *starttime) {
     }
 }
 
-
-
-const long long default_n_items = 1LL<<22;
-
 static int print_usage (const char *argv0) {
     fprintf(stderr, "Usage:\n");
     fprintf(stderr, " %s [-x] [--keysize KEYSIZE] [--valsize VALSIZE] [--noserial] [--norandom] [ n_iterations ]\n", argv0);
@@ -556,8 +558,8 @@ static int print_usage (const char *argv0) {
     fprintf(stderr, "    --insertmultiple      Use DB_ENV->put_multiple api.  Requires transactions.\n"); 
     fprintf(stderr, "    --redzone N           redzone in percent\n");
     fprintf(stderr, "    --srandom N           srandom(N)\n");
-    fprintf(stderr, "   n_iterations     how many iterations (default %lld)\n", default_n_items/DEFAULT_ITEMS_TO_INSERT_PER_ITERATION);
     fprintf(stderr, "    --engine_status       print engine status at end of test \n");
+    fprintf(stderr, "   n_iterations     how many iterations (default %lld)\n", DEFAULT_N_ITERATIONS);
 
     return 1;
 }
@@ -600,7 +602,7 @@ test1514(void) {
 
 static int test_main (int argc, char *const argv[]) {
     struct timeval t1,t2,t3;
-    long long total_n_items = default_n_items;
+    long long total_n_items = DEFAULT_N_ITEMS;
     char *endptr;
     int i;
     for (i=1; i<argc; i++) {
@@ -734,12 +736,12 @@ static int test_main (int argc, char *const argv[]) {
         /* if it looks like a number */
         char *end;
         errno=0;
-        long n_iterations = strtol(argv[i], &end, 10);
+        n_iterations = strtol(argv[i], &end, 10);
         if (errno!=0 || *end!=0 || end==argv[i]) {
             print_usage(argv[0]);
             return 1;
         }
-        total_n_items = items_per_iteration * (long long)n_iterations;
+        total_n_items = items_per_iteration * n_iterations;
 	i++;
     }
     if (i<argc) {

linux/tests/test-cpu-freq.c

@@ -3,10 +3,15 @@
 #include <toku_stdint.h>
 #include <toku_os.h>
 
+int verbose = 0;
+
 int main(void) {
     uint64_t cpuhz;
     int r = toku_os_get_processor_frequency(&cpuhz);
     assert(r == 0);
+    if (verbose) {
 	printf("%"PRIu64"\n", cpuhz);
+    }
+    assert(cpuhz>100000000);
     return 0;
 }

newbrt/brt.c

@@ -3629,6 +3629,13 @@ int toku_brt_get_nodesize(BRT brt, unsigned int *nodesize) {
     return 0;
 }
 
+void toku_brt_get_maximum_advised_key_value_lengths (unsigned int *max_key_len, unsigned int *max_val_len)
+// return the maximum advisable key value lengths.  The brt doesn't enforce these.
+{
+    *max_key_len = 32*1024;
+    *max_val_len = 32*1024*1024;
+}
+
 int toku_brt_set_basementnodesize(BRT brt, unsigned int basementnodesize) {
     brt->basementnodesize = basementnodesize;
     return 0;

newbrt/brt.h

@@ -44,6 +44,7 @@ int toku_brt_set_flags(BRT, unsigned int flags)  __attribute__ ((warn_unused_res
 int toku_brt_get_flags(BRT, unsigned int *flags)  __attribute__ ((warn_unused_result));
 int toku_brt_set_nodesize(BRT, unsigned int nodesize)  __attribute__ ((warn_unused_result));
 int toku_brt_get_nodesize(BRT, unsigned int *nodesize) __attribute__ ((warn_unused_result));
+void toku_brt_get_maximum_advised_key_value_lengths(unsigned int *klimit, unsigned int *vlimit);
 int toku_brt_set_basementnodesize(BRT, unsigned int basementnodesize)  __attribute__ ((warn_unused_result));
 int toku_brt_get_basementnodesize(BRT, unsigned int *basementnodesize) __attribute__ ((warn_unused_result));
 

newbrt/brtloader.c

@@ -1130,6 +1130,9 @@ static int process_primary_rows_internal (BRTLOADER bl, struct rowset *primary_r
     #pragma cilk grainsize = 1
 #endif
     cilk_for (int i = 0; i < bl->N; i++) {
+	unsigned int klimit,vlimit; // maximum row sizes.
+	toku_brt_get_maximum_advised_key_value_lengths(&klimit, &vlimit);
+
 	error_codes[i] = 0;
 	struct rowset *rows = &(bl->rows[i]);
 	struct merge_fileset *fs = &(bl->fs[i]);
@@ -1159,6 +1162,18 @@ static int process_primary_rows_internal (BRTLOADER bl, struct rowset *primary_r
                     inc_error_count();
                     break;
                 }
+		if (skey.size > klimit) {
+		    error_codes[i] = EINVAL;
+		    fprintf(stderr, "Key too big (keysize=%d bytes, limit=%d bytes)\n", skey.size, klimit);
+		    inc_error_count();
+		    break;
+		}
+		if (sval.size > vlimit) {
+		    error_codes[i] = EINVAL;
+		    fprintf(stderr, "Row too big (rowsize=%d bytes, limit=%d bytes)\n", sval.size, vlimit);
+		    inc_error_count();
+		    break;
+		}
 	    }
 
 	    bl->extracted_datasizes[i] += brtloader_leafentry_size(skey.size, sval.size, leafentry_xid(bl, i));

src/tests/Makefile

@@ -154,6 +154,7 @@ BDB_DONTRUN_TESTS = \
         loader-tpch-load \
 	lock-pressure \
         manyfiles \
+        maxsize-for-loader \
 	multiprocess \
 	mvcc-create-table \
         mvcc-many-committed \
@@ -1037,6 +1038,12 @@ keyrange-loader-get0.tdbrun: keyrange.tdb
 keyrange-loader-get1.tdbrun: keyrange.tdb
 	$(TDBVGRIND) ./$< $(VERBVERBOSE) --get 1 --loader 1 --envdir dir.$@ $(MAYBEINVERTER) $(SUMMARIZE_CMD)
 
+maxsize-for-loader.tdbrun: maxsize-for-loader-A.tdbrun maxsize-for-loader-B.tdbrun 
+	@echo -n
+maxsize-for-loader-A.tdbrun: maxsize-for-loader.tdb
+	$(TDBVGRIND) ./$< -e $@  -f 2> /dev/null $(SUMMARIZE_CMD)
+maxsize-for-loader-B.tdbrun: maxsize-for-loader.tdb
+	./$< -e $@                  2> /dev/null $(SUMMARIZE_CMD)
 clean:
 	rm -f $(ALL_BINS)
 	rm -rf dir.* *.check.output *.check.valgrind

src/tests/loader-dup-test.c

 /* -*- mode: C; c-basic-offset: 4 -*- */
 #ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
-#ident "$Id $"
+#ident "$Id$"
 
 #include "test.h"
 #include "toku_pthread.h"

src/tests/maxsize-for-loader.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "Copyright (c) 2010 Tokutek Inc.  All rights reserved."
+#ident "$Id$"
+
+#include "test.h"
+#include "toku_pthread.h"
+#include <db.h>
+#include <sys/stat.h>
+
+bool fast = false;
+
+DB_ENV *env;
+enum {NUM_DBS=2};
+int USE_PUTS=0;
+
+uint32_t num_rows = 1;
+uint32_t which_db_to_fail = -1;
+uint32_t which_row_to_fail = -1;
+enum how_to_fail { FAIL_NONE, FAIL_KSIZE, FAIL_VSIZE } how_to_fail = FAIL_NONE;
+
+static int put_multiple_generate(DB *dest_db,
+				 DB *src_db __attribute__((__unused__)),
+				 DBT *dest_key, DBT *dest_val,
+				 const DBT *src_key, const DBT *src_val __attribute__((__unused__))) {
+
+    uint32_t which = *(uint32_t*)dest_db->app_private;
+    assert(src_key->size==4);
+    uint32_t rownum = *(uint32_t*)src_key->data;
+
+    uint32_t ksize, vsize;
+    const uint32_t kmax=32*1024, vmax=32*1024*1024;
+    if (which==which_db_to_fail && rownum==which_row_to_fail) {
+	switch (how_to_fail) {
+	case FAIL_NONE:  ksize=kmax;   vsize=vmax;   goto gotsize;
+	case FAIL_KSIZE: ksize=kmax+1; vsize=vmax;   goto gotsize;
+	case FAIL_VSIZE: ksize=kmax;   vsize=vmax+1; goto gotsize;
+	}
+	assert(0);
+    gotsize:;
+    } else {
+	ksize=4; vsize=100;
+    }
+    assert(dest_key->flags==DB_DBT_REALLOC);
+    if (dest_key->ulen < ksize) {
+	dest_key->data = toku_xrealloc(dest_key->data, ksize);
+	dest_key->ulen = ksize;
+    }
+    assert(dest_val->flags==DB_DBT_REALLOC);
+    if (dest_val->ulen < vsize) {
+	dest_val->data = toku_xrealloc(dest_val->data, vsize);
+	dest_val->ulen = vsize;
+    }
+    assert(ksize>=sizeof(uint32_t));
+    for (uint32_t i=0; i<ksize; i++) ((char*)dest_key->data)[i] = random();
+    for (uint32_t i=0; i<vsize; i++) ((char*)dest_val->data)[i] = random();
+    *(uint32_t*)dest_key->data = rownum;
+    dest_key->size = ksize;
+    dest_val->size = vsize;
+
+    return 0;
+}
+
+struct error_extra {
+    int bad_i;
+    int error_count;
+};
+
+static void error_callback (DB *db __attribute__((__unused__)), int which_db, int err, DBT *key __attribute__((__unused__)), DBT *val __attribute__((__unused__)), void *extra) {
+    struct error_extra *e =(struct error_extra *)extra;
+    assert(which_db==(int)which_db_to_fail);
+    assert(err==EINVAL);
+    assert(e->error_count==0);
+    e->error_count++;
+}
+
+static void test_loader_maxsize(DB **dbs)
+{
+    int r;
+    DB_TXN    *txn;
+    DB_LOADER *loader;
+    uint32_t db_flags[NUM_DBS];
+    uint32_t dbt_flags[NUM_DBS];
+    for(int i=0;i<NUM_DBS;i++) { 
+        db_flags[i] = DB_NOOVERWRITE; 
+        dbt_flags[i] = 0;
+    }
+    uint32_t loader_flags = USE_PUTS; // set with -p option
+
+    // create and initialize loader
+    r = env->txn_begin(env, NULL, &txn, 0);                                                               
+    CKERR(r);
+    r = env->create_loader(env, txn, &loader, dbs[0], NUM_DBS, dbs, db_flags, dbt_flags, loader_flags);
+    CKERR(r);
+    struct error_extra error_extra = {.error_count=0};
+    r = loader->set_error_callback(loader, error_callback, (void*)&error_extra);
+    CKERR(r);
+    r = loader->set_poll_function(loader, NULL, NULL);
+    CKERR(r);
+
+    // using loader->put, put values into DB
+    DBT key, val;
+    unsigned int k, v;
+    for(uint32_t i=0;i<num_rows;i++) {
+        k = i;
+        v = i;
+        dbt_init(&key, &k, sizeof(unsigned int));
+        dbt_init(&val, &v, sizeof(unsigned int));
+        r = loader->put(loader, &key, &val);
+        if (USE_PUTS) {
+            //PUT loader can return -1 if it finds an error during the puts.
+            CKERR2s(r, 0,-1);
+        }
+        else {
+            CKERR(r);
+        }
+    }
+
+    // close the loader
+    if (verbose) { printf("closing"); fflush(stdout); }
+    r = loader->close(loader);
+    if (verbose) {  printf(" done\n"); }
+    switch(how_to_fail) {
+    case FAIL_NONE:  assert(r==0);      assert(error_extra.error_count==0); goto checked;
+    case FAIL_KSIZE: assert(r==EINVAL); assert(error_extra.error_count==1); goto checked;
+    case FAIL_VSIZE: assert(r==EINVAL); assert(error_extra.error_count==1); goto checked;
+    }
+    assert(0);
+ checked:
+
+    r = txn->commit(txn, 0);
+    CKERR(r);
+}
+
+char *free_me = NULL;
+char *env_dir = ENVDIR; // the default env_dir
+
+static void run_test(uint32_t nr, uint32_t wdb, uint32_t wrow, enum how_to_fail htf) {
+    num_rows = nr; which_db_to_fail = wdb; which_row_to_fail = wrow; how_to_fail = htf;
+    int r;
+    {
+	int len = strlen(env_dir) + 20;
+	char syscmd[len];
+	r = snprintf(syscmd, len, "rm -rf %s", env_dir);
+	assert(r<len);
+	r = system(syscmd);                                                                                   CKERR(r);
+    }
+    r = toku_os_mkdir(env_dir, S_IRWXU+S_IRWXG+S_IRWXO);                                                       CKERR(r);
+
+    r = db_env_create(&env, 0);                                                                               CKERR(r);
+    r = env->set_default_bt_compare(env, uint_dbt_cmp);                                                       CKERR(r);
+    r = env->set_generate_row_callback_for_put(env, put_multiple_generate);
+    CKERR(r);
+    int envflags = DB_INIT_LOCK | DB_INIT_MPOOL | DB_INIT_TXN | DB_INIT_LOG | DB_CREATE | DB_PRIVATE;
+    r = env->open(env, env_dir, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                                            CKERR(r);
+    env->set_errfile(env, stderr);
+    //Disable auto-checkpointing
+    r = env->checkpointing_set_period(env, 0);                                                                CKERR(r);
+
+    DBT desc;
+    dbt_init(&desc, "foo", sizeof("foo"));
+    enum {MAX_NAME=128};
+    char name[MAX_NAME*2];
+
+    DB **dbs = (DB**)toku_malloc(sizeof(DB*) * NUM_DBS);
+    assert(dbs != NULL);
+    int idx[NUM_DBS];
+    for(int i=0;i<NUM_DBS;i++) {
+        idx[i] = i;
+        r = db_create(&dbs[i], env, 0);                                                                       CKERR(r);
+        dbs[i]->app_private = &idx[i];
+        snprintf(name, sizeof(name), "db_%04x", i);
+        r = dbs[i]->open(dbs[i], NULL, name, NULL, DB_BTREE, DB_CREATE, 0666);                                CKERR(r);
+        IN_TXN_COMMIT(env, NULL, txn_desc, 0, {
+            CHK(dbs[i]->change_descriptor(dbs[i], txn_desc, &desc, 0));
+        });
+    }
+
+    if (verbose) printf("running test_loader()\n");
+    // -------------------------- //
+    test_loader_maxsize(dbs);
+    // -------------------------- //
+    if (verbose) printf("done    test_loader()\n");
+
+    for(int i=0;i<NUM_DBS;i++) {
+        dbs[i]->close(dbs[i], 0);                                                                             CKERR(r);
+        dbs[i] = NULL;
+    }
+    r = env->close(env, 0);                                                                                   CKERR(r);
+    toku_free(dbs);
+}
+
+// ------------ infrastructure ----------
+static void do_args(int argc, char * const argv[]);
+
+int num_rows_set = FALSE;
+
+int test_main(int argc, char * const *argv) {
+    do_args(argc, argv);
+
+    run_test(1, -1, -1, FAIL_NONE);
+    run_test(1,  0,  0, FAIL_NONE);
+    run_test(1,  0,  0, FAIL_KSIZE);
+    run_test(1,  0,  0, FAIL_VSIZE);
+    if (!fast) {
+	run_test(1000000, 0, 500000, FAIL_KSIZE);
+	run_test(1000000, 0, 500000, FAIL_VSIZE);
+    }
+    toku_free(free_me);
+    return 0;
+}
+
+static void do_args(int argc, char * const argv[]) {
+    int resultcode;
+    char *cmd = argv[0];
+    argc--; argv++;
+    while (argc>0) {
+        if (strcmp(argv[0], "-h")==0) {
+            resultcode=0;
+	do_usage:
+	    fprintf(stderr, "Usage: %s [-h] [-v] [-q] [-p] [-f] [ -e <envdir> ]\n", cmd);
+	    fprintf(stderr, " where -e <env>         uses <env> to construct the directory (so that different tests can run concurrently)\n");
+	    fprintf(stderr, "       -h               help\n");
+	    fprintf(stderr, "       -v               verbose\n");
+	    fprintf(stderr, "       -q               quiet\n");
+	    fprintf(stderr, "       -p               use DB->put\n");
+	    fprintf(stderr, "       -f               fast (suitable for vgrind)\n");
+	    exit(resultcode);
+	} else if (strcmp(argv[0], "-e")==0) {
+            argc--; argv++;
+	    if (free_me) toku_free(free_me);
+	    int len = strlen(ENVDIR) + strlen(argv[0]) + 2;
+	    char full_env_dir[len];
+	    int r = snprintf(full_env_dir, len, "%s.%s", ENVDIR, argv[0]);
+	    assert(r<len);
+	    free_me = env_dir = toku_strdup(full_env_dir);
+	} else if (strcmp(argv[0], "-v")==0) {
+	    verbose++;
+	} else if (strcmp(argv[0],"-q")==0) {
+	    verbose--;
+	    if (verbose<0) verbose=0;
+        } else if (strcmp(argv[0], "-p")==0) {
+            USE_PUTS = 1;
+        } else if (strcmp(argv[0], "-f")==0) {
+	    fast     = true;
+	} else {
+	    fprintf(stderr, "Unknown arg: %s\n", argv[0]);
+	    resultcode=1;
+	    goto do_usage;
+	}
+	argc--;
+	argv++;
+    }
+}

src/tests/rowsize.c

+/* -*- mode: C; c-basic-offset: 4 -*- */
+#ident "$Id: brt.c 38079 2011-12-21 20:23:22Z leifwalsh $"
+#ident "Copyright (c) 2011 Tokutek Inc.  All rights reserved."
+
+#include "test.h"
+
+static DB_ENV *env = NULL;
+static DB *db = NULL;
+static char *envdir = ENVDIR;
+
+static void setup_env (void) {
+    const int len = strlen(envdir)+100;
+    char cmd[len];
+    snprintf(cmd, len, "rm -rf %s", envdir);
+    system(cmd);
+    {int r = toku_os_mkdir(envdir, S_IRWXU+S_IRWXG+S_IRWXO);                                                                            CKERR(r); }
+    {int r = db_env_create(&env, 0);                                                                                                    CKERR(r); }
+    //env->set_errfile(env, stderr);
+#ifdef TOKUDB
+    CKERR(env->set_redzone(env, 0));
+#endif
+    { int 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); }
+    { int r = db_create(&db, env, 0);                                                                                                   CKERR(r); }
+    { int r = db->open(db, NULL, "foo.db", 0, DB_BTREE, DB_CREATE | DB_AUTO_COMMIT, S_IRWXU+S_IRWXG+S_IRWXO);                           CKERR(r); }
+}
+
+static void shutdown_env (void) {
+    { int r = db->close(db, 0);   CKERR(r); }
+    { int r = env->close(env, 0); CKERR(r); }
+}
+
+static void put (const char *keystring, int size, bool should_work) {
+    DBT k = {.size = 1+strlen(keystring),
+	     .data = (void*)keystring};
+    DBT v = {.size = size,
+	     .data = toku_xmalloc(size)};
+    memset(v.data, 0, size);
+#ifdef USE_BDB
+#define DB_YES_OVERWRITE 0
+#endif
+    static DB_TXN *txn = NULL;
+    { int r = env->txn_begin(env, 0, &txn, 0); CKERR(r); }
+    {
+	int r = db->put(db, NULL, &k, &v, 0);
+	if (!IS_TDB || should_work) {
+	    CKERR(r);
+	} else {
+	    assert(r!=0);
+	}
+    }
+    { int r = txn->commit(txn, 0); CKERR(r); }
+    toku_free(v.data);
+}
+
+int test_main (int argc, char *const argv[]) {
+    if (0) parse_args(argc, argv);
+    setup_env();
+    if (0) put("foo", 32, true);
+    put("foo", 32*1024*1024, true);
+    put("bar", 32*1024*1024+1, false);
+    shutdown_env();
+    
+    return 0;
+}

src/ydb.c

@@ -4964,17 +4964,13 @@ db_open_iname(DB * db, DB_TXN * txn, const char *iname_in_env, u_int32_t flags,
 //Return non zero otherwise.
 static int
 db_put_check_size_constraints(DB *db, const DBT *key, const DBT *val) {
-    int r;
-
-    //Check limits on size of key and val.
-    unsigned int nodesize;
-    r = toku_brt_get_nodesize(db->i->brt, &nodesize); assert(r == 0);
-
-    u_int32_t limit = nodesize / BRT_FANOUT;
-    if (key->size > limit)
-	r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest key allowed is %u bytes", limit);
-    else if (val->size > nodesize)
-	r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest value allowed is %u bytes", nodesize);
+    unsigned int klimit, vlimit;
+    int r = 0;
+    toku_brt_get_maximum_advised_key_value_lengths(&klimit, &vlimit);
+    if (key->size > klimit)
+	r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest key allowed is %u bytes", klimit);
+    else if (val->size > vlimit)
+	r = toku_ydb_do_error(db->dbenv, EINVAL, "The largest value allowed is %u bytes", vlimit);
     
     return r;
 }

windows/tests/test-pthread-rwlock-rwr.c

@@ -3,39 +3,54 @@
 #include <toku_pthread.h>
 #include <stdio.h>
 #include <unistd.h>
+#include <string.h>
+
+// write a test see if things happen in the right order.
+
+volatile int state = 0;
+int verbose = 0;
 
 static void *f(void *arg) {
     int r;
     toku_pthread_rwlock_t *mylock = arg;
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    sleep(2);
+    assert(state==42); state = 16; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     r = toku_pthread_rwlock_wrlock(mylock); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    assert(state==49); state = 17; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     r = toku_pthread_rwlock_wrunlock(mylock); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    sleep(10);
+    assert(state==52); state = 20; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     return arg;
 }
 
-int test_main(int argc __attribute__((__unused__)), char *const argv[] __attribute__((__unused__))) {
+int test_main(int argc , char *const argv[] ) {
+    assert(argc==1 || argc==2);
+    if (argc==2) {
+	assert(strcmp(argv[1],"-v")==0);
+	verbose = 1;
+    }
     int r;
     toku_pthread_rwlock_t rwlock;
     toku_pthread_t tid;
     void *retptr;
 
     r = toku_pthread_rwlock_init(&rwlock, NULL); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    state = 37; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     r = toku_pthread_rwlock_rdlock(&rwlock); assert(r == 0);
 
     r = toku_pthread_create(&tid, NULL, f, &rwlock); assert(r == 0);
 
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
-    sleep(10);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    assert(state==37); state = 42; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
+    sleep(4);
+    assert(state==16); state = 44; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     r = toku_pthread_rwlock_rdlock(&rwlock); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    assert(state==44); state = 46; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
     r = toku_pthread_rwlock_rdunlock(&rwlock); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    sleep(4);
+    assert(state==46); state=49; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__); // still have a read lock
     r = toku_pthread_rwlock_rdunlock(&rwlock); assert(r == 0);
-    printf("%s:%d\n", __FUNCTION__, __LINE__); fflush(stdout);
+    sleep(6);
+    assert(state==17); state=52; if (verbose) printf("%s:%d\n", __FUNCTION__, __LINE__);
 
     r = toku_pthread_join(tid, &retptr); assert(r == 0);