ptquery.c added to db-benchmark-test dir #3726 closes [t:3726]

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

db-benchmark-test/Makefile

@@ -23,11 +23,13 @@ TDB_DLINK_FILES = $(TOKUROOT)lib/$(TOKUDB)
 
 TARGET_BDB = db-benchmark-test-bdb$(BINSUF)
 SCANSCAN_BDB = scanscan-bdb$(BINSUF)
+PTQUERY_BDB = ptquery-bdb$(BINSUF)
 TARGET_TDB = db-benchmark-test-tokudb$(BINSUF)
 SCANSCAN_TDB = scanscan-tokudb$(BINSUF)
+PTQUERY_TDB = ptquery-tokudb$(BINSUF)
 SCANRACE_TDB = scanrace-tokudb$(BINSUF)
 MULTIBENCH_TDB = multi-bench-tokudb$(BINSUF)
-TARGETS = $(TARGET_BDB) $(SCANSCAN_BDB) $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB)
+TARGETS = $(TARGET_BDB) $(SCANSCAN_BDB) $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB) $(PTQUERY_BDB)
 TARGETS_BDB = $(TARGET_BDB) $(SCANSCAN_BDB)
 
 ifeq ($(OS_CHOICE),windows)
@@ -68,7 +70,7 @@ build: build.tdb build.bdb;
 endif
 
 build.bdb: $(TARGET_BDB) $(SCANSCAN_BDB) $(WINDOWS_BDB_LIB_NAME)
-build.tdb: $(TARGET_TDB) $(SCANSCAN_TDB) $(MULTIBENCH_TDB)
+build.tdb: $(TARGET_TDB) $(SCANSCAN_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB)
 
 check: check-default check-rowsize check-xfast check-x check-no-rollback check-4G child.benchmark.dir
 
@@ -115,7 +117,7 @@ clean:
 	rm -f $(TARGETS) 4g.out
 	rm -rf *.dir $(BENCHDBS)
 
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): BIN_FROM_C_FLAGS+=-DDIRSUF=tokudb -I$(PORTABILITY_HEADERS) -I$(TOKUROOT)toku_include -I$(TOKUROOT)include
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB): BIN_FROM_C_FLAGS+=-DDIRSUF=tokudb -I$(PORTABILITY_HEADERS) -I$(TOKUROOT)toku_include -I$(TOKUROOT)include
 
 ifneq ($(PROF),)
     USE_STATIC=1
@@ -123,7 +125,7 @@ endif
 
 # A hack to make gprof work.  See #515.
 ifneq ($(USE_STATIC),)
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): DLINK_FILES=
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB): DLINK_FILES=
 
 OFILES = \
 	$(TOKUROOT)src/ydb.lib \
@@ -137,6 +139,8 @@ $(TARGET_TDB): db-benchmark-test.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(OFILES) $(LINK_MUST_BE_LAST)
 $(SCANSCAN_TDB): scanscan.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(OFILES) $(LINK_MUST_BE_LAST)
+$(PTQUERY_TDB): ptquery.c  $(PTHREAD_LOCAL)
+	$(CC) $< $(BIN_FROM_C_FLAGS) $(OFILES) $(LINK_MUST_BE_LAST)
 $(SCANRACE_TDB): scanrace.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(OFILES) $(LINK_MUST_BE_LAST)
 $(MULTIBENCH_TDB): multi-bench.c $(PTHREAD_LOCAL)
@@ -147,14 +151,16 @@ ifeq ($(OS_CHOICE),windows)
 $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB): $(WIN_YDB) $(PTHREAD_LOCAL)
 $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB): LINK_FILES+=$(WIN_YDB)
 else
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): DLINK_FILES=$(TDB_DLINK_FILES)
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): RPATH_DIRS=$(dir $(TDB_DLINK_FILES))
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB): DLINK_FILES=$(TDB_DLINK_FILES)
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB): RPATH_DIRS=$(dir $(TDB_DLINK_FILES))
 endif
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): LDLIBS+=-ltokuportability
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB) $(PTQUERY_TDB): LDLIBS+=-ltokuportability
 $(TARGET_TDB): db-benchmark-test.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
 $(SCANSCAN_TDB): scanscan.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
+$(PTQUERY_TDB): ptquery.c $(PTHREAD_LOCAL)
+	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
 $(SCANRACE_TDB): scanrace.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
 $(MULTIBENCH_TDB): multi-bench.c $(PTHREAD_LOCAL)

db-benchmark-test/ptquery.c

+//#include <stdbool.h>
+//#include "test.h"
+//#include "toku_pthread.h"
+
+#include <toku_portability.h>
+#include "tokudb_common_funcs.h"
+#include <toku_assert.h>
+#include <db.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <string.h>
+#include <fcntl.h>
+#include <unistd.h>
+#ifdef TOKUDB
+#include "key.h"
+#include "cachetable.h"
+#include "trace_mem.h"
+#endif
+
+DB_ENV *env;
+DB *db;
+DB_TXN *tid=0;
+
+#define STRINGIFY2(s) #s
+#define STRINGIFY(s) STRINGIFY2(s)
+const char *dbdir = "./bench."  STRINGIFY(DIRSUF); /* DIRSUF is passed in as a -D argument to the compiler. */
+int env_open_flags_yesx = DB_CREATE|DB_PRIVATE|DB_INIT_MPOOL|DB_INIT_TXN|DB_INIT_LOG|DB_INIT_LOCK;
+int env_open_flags_nox = DB_CREATE|DB_PRIVATE|DB_INIT_MPOOL;
+char *dbfilename = "bench.db";
+u_int32_t cachesize = 127*1024*1024;
+static int verbose UU() = 0;
+static const char *log_dir = NULL;
+
+static long long set_count = 0;
+
+static void pt_query_setup (void) {
+    int r;
+    r = db_env_create(&env, 0);                                                           assert(r==0);
+    r = env->set_cachesize(env, 0, cachesize, 1);                                         assert(r==0);
+    if (log_dir) {
+        r = env->set_lg_dir(env, log_dir);                                                assert(r==0);
+    }
+    r = env->open(env, dbdir, env_open_flags_yesx, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);       assert(r==0);
+    r = db_create(&db, env, 0);                                                            assert(r==0);
+    r = db->open(db, tid, dbfilename, NULL, DB_BTREE, 0, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH); assert(r==0);
+}
+
+static void pt_query_shutdown (void) {
+    int r;
+    r = db->close(db, 0);                                       assert(r==0);
+    r = env->close(env, 0);                                     assert(r==0);
+    env = NULL;
+}
+
+static unsigned long long maxkey;
+enum { SERIAL_SPACING = 1<<6 };
+static const int keylen=8;
+
+static double gettime (void) {
+    struct timeval tv;
+    int r = gettimeofday(&tv, 0);
+    assert(r==0);
+    return tv.tv_sec + 1e-6*tv.tv_usec;
+}
+
+/* From db-benchmark-test.c */
+static void long_long_to_array (unsigned char *a, int array_size, unsigned long long l) {
+    int i;
+    for (i=0; i<8 && i<array_size; i++)
+	a[i] = (l>>(56-8*i))&0xff;
+}
+
+static void array_to_long_long (unsigned long long *l, unsigned char *a, int array_size) {
+    int i;
+    *l = 0;
+    unsigned long long tmp;
+    for(i=0; i<8 && i<array_size; i++) {
+        tmp =  a[i] & 0xff;
+        *l += tmp << (56-8*i);
+    }
+}
+
+static void test_begin_commit(int nqueries) {
+    int r;
+    unsigned long long k;
+    unsigned char kv[keylen];
+    for (int i = 0; i < nqueries; i++) {
+        DB_TXN *txn = NULL;
+        r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+        DBC *c = NULL;
+        r = db->cursor(db, txn, &c, 0); assert_zero(r);
+
+        k = (random() + (random() << 31)) % maxkey;
+        k = ( k / SERIAL_SPACING ) * SERIAL_SPACING;
+        long_long_to_array(kv, keylen, k);
+        
+        DBT key, val;
+        memset(&key, 0, sizeof key); key.data=kv; key.size=8;
+        memset(&val, 0, sizeof val);
+        r = c->c_get(c, &key, &val, DB_SET); 
+        assert_zero(r);
+        (void) __sync_fetch_and_add(&set_count, 1);
+        r = c->c_close(c); assert_zero(r);
+        r = txn->commit(txn, 0); assert_zero(r);
+    }
+}
+
+// read in the entire DB, warming up the cache
+//  - record maxkey 
+static void warmup(void) {
+    int r;
+    DB_TXN *txn=NULL;
+    DBC *c = NULL;
+    DBT key, val;
+
+    double tstart = gettime();
+    memset(&key, 0, sizeof key);
+    memset(&val, 0, sizeof val);
+    r = env->txn_begin(env, NULL, &txn, 0); assert_zero(r);
+    r = db->cursor(db, txn, &c, 0); assert_zero(r);
+    r = c->c_get(c, &key, &val, DB_FIRST); assert_zero(r);
+    assert(key.size == 8);
+    while ( r != DB_NOTFOUND ) {
+        r = c->c_get(c, &key, &val, DB_NEXT);
+        if ( r != 0 && r != DB_NOTFOUND) assert_zero(r);
+    }
+    memset(&key, 0, sizeof key);
+    memset(&val, 0, sizeof val);
+    r = c->c_get(c, &key, &val, DB_LAST); assert_zero(r);
+    array_to_long_long(&maxkey, key.data, key.size);
+    r = c->c_close(c); assert_zero(r);
+    r = txn->commit(txn, 0); assert_zero(r);
+    double tdiff = gettime() - tstart;
+    unsigned long long rows = maxkey / SERIAL_SPACING;
+    printf("Warmup        : read %12llu rows in %6.1fs %8.0f/s\n", rows, tdiff, rows/tdiff);
+}
+
+
+struct arg {
+//    DB_ENV *env;
+//    DB *db;
+    int nqueries;
+//    int nrows;
+};
+
+static void *test_thread(void *arg) {
+    struct arg *myarg = (struct arg *) arg;
+    test_begin_commit(myarg->nqueries);
+    return arg;
+}
+
+
+int test_main(int argc, char * const argv[]) {
+    int nqueries = 1000000;
+    int nthreads = 1;
+
+    // parse_args(argc, argv);
+    for (int i = 1; i < argc; i++) {
+        char * const arg = argv[i];
+        if (strcmp(arg, "--nqueries") == 0 && i+1 < argc) {
+            nqueries = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--nthreads") == 0 && i+1 < argc) {
+            nthreads = atoi(argv[++i]);
+            continue;
+        }
+        return 1;
+    }
+
+    int r;
+    pt_query_setup();
+
+    warmup();
+    
+    assert(nthreads > 0);
+    struct arg myargs[nthreads-1];
+    toku_pthread_t mytids[nthreads];
+    double tstart = gettime();
+    for (int i = 0; i < nthreads-1; i++) {
+        myargs[i] = (struct arg) { nqueries };
+        r = toku_pthread_create(&mytids[i], NULL, test_thread, &myargs[i]); assert_zero(r);
+    }
+    test_begin_commit(nqueries);
+    for (int i = 0; i < nthreads-1; i++) {
+        void *ret;
+        r = toku_pthread_join(mytids[i], &ret); assert_zero(r);
+    }
+    assert(set_count == nthreads * nqueries);
+    double tdiff = gettime() - tstart;
+    printf("Point Queries : read %12llu rows in %6.1fs %8.0f/s with %d threads\n", set_count, tdiff, set_count/tdiff, nthreads);
+
+    pt_query_shutdown();
+    return 0;
+}

db-benchmark-test/scanscan.c

@@ -17,7 +17,7 @@
 #endif
 
 const char *pname;
-enum run_mode { RUN_HWC, RUN_LWC, RUN_VERIFY, RUN_RANGE, RUN_FLATTEN} run_mode = RUN_HWC;
+enum run_mode { RUN_HWC, RUN_LWC, RUN_VERIFY, RUN_RANGE, RUN_FLATTEN, RUN_PT} run_mode = RUN_HWC;
 int do_txns=1, prelock=0, prelockflag=0;
 u_int32_t lock_flag = 0;
 long limitcount=-1;
@@ -25,6 +25,7 @@ u_int32_t cachesize = 127*1024*1024;
 static int do_mysql = 0;
 static u_int64_t start_range = 0, end_range = 0;
 static int n_experiments = 2;
+static long pt_row_cnt = 1;
 
 static int verbose = 0;
 static const char *log_dir = NULL;
@@ -48,6 +49,7 @@ static int print_usage (const char *argv0) {
     fprintf(stderr, "  --experiments N     run N experiments (default:%d)\n", n_experiments);
     fprintf(stderr, "  --srandom N         srandom(N)\n");
     fprintf(stderr, "  --recover           run recovery\n");
+    fprintf(stderr, "  --ptquery           run point query test\n");
     fprintf(stderr, "  --verbose           print verbose information\n");
     return 1;
 }
@@ -89,6 +91,9 @@ static void parse_args (int argc, char *const argv[]) {
 	} else if (strcmp(*argv, "--hwc")==0)  {
 	    if (specified_run_mode && run_mode!=RUN_VERIFY) goto two_modes;
 	    run_mode = RUN_HWC;
+	} else if (strcmp(*argv, "--ptquery")==0)  {
+	    if (specified_run_mode && run_mode!=RUN_PT) goto two_modes;
+	    run_mode = RUN_PT;
 	} else if (strcmp(*argv, "--prelock")==0) prelock=1;
 #ifdef TOKUDB
         else if (strcmp(*argv, "--prelockflag")==0)      { prelockflag=1; lock_flag = DB_PRELOCKED; }
@@ -144,6 +149,23 @@ static void parse_args (int argc, char *const argv[]) {
     }
 }
 
+/* From db-benchmark-test.c */
+static void long_long_to_array (unsigned char *a, int array_size, unsigned long long l) {
+    int i;
+    for (i=0; i<8 && i<array_size; i++)
+	a[i] = (l>>(56-8*i))&0xff;
+}
+
+static void array_to_long_long (unsigned long long *l, unsigned char *a, int array_size) {
+    int i;
+    *l = 0;
+    unsigned long long tmp;
+    for(i=0; i<8 && i<array_size; i++) {
+        tmp =  a[i] & 0xff;
+        *l += tmp << (56-8*i);
+    }
+}
+
 
 static inline uint64_t mysql_get_bigint(unsigned char *d) {
     uint64_t r = 0;
@@ -165,6 +187,8 @@ static int mysql_key_compare(DB *mydb __attribute__((unused)),
     return 0;
 }
 
+
+
 static void scanscan_setup (void) {
     int r;
     r = db_env_create(&env, 0);                                                           assert(r==0);
@@ -261,6 +285,7 @@ static void scanscan_hwc (void) {
 	double thistime = gettime();
 	double tdiff = thistime-prevtime;
 	printf("Scan    %lld bytes (%d rows) in %9.6fs at %9fMB/s\n", totalbytes, rowcounter, tdiff, 1e-6*totalbytes/tdiff);
+        pt_row_cnt=rowcounter;
 	print_engine_status();
     }
 }
@@ -325,6 +350,67 @@ static void scanscan_flatten (void) {
 }
 #endif
 
+//  To measure point query performance
+//    $ ./db-benchmark-test-tokudb -x --norandom 32
+//    $ ./scanscan-tokudb --prelock --prelockflag --ptquery
+
+static void scanscan_ptquery (void) {
+    int r;
+    // read in the whole database to warm up the cache
+    printf("Warm-up\n");
+    {
+        int tmp_experiments = n_experiments;
+        n_experiments = 1;
+        scanscan_hwc();
+        n_experiments = tmp_experiments;
+    }
+    // perform n_experiments point queries    
+    
+    DBT key, val;
+    DBC *dbc;
+    r = db->cursor(db, tid, &dbc, 0); assert(r==0);
+
+    int const ptqpertxn = 1<<20;
+//    int const ptqpertxn = 1<<8;
+    int counter = 0;
+    unsigned long long k, max_key;
+    unsigned char kv[8];
+    // figure out max index
+    memset(&key, 0, sizeof key); 
+    memset(&val, 0, sizeof val);
+    dbc->c_get(dbc, &key, &val, DB_LAST);
+    array_to_long_long(&max_key, key.data, key.size);
+    
+    // assume stride of 16 (SERIAL_SPACING from db-benchmark-test)
+    int const idx_stride=16;
+
+    printf("Random point queries of %d per batch (with transactions)\n", ptqpertxn);
+    int ex;
+    for (ex=0;ex<n_experiments;ex++) {
+        double tstart = gettime();
+        for (counter = 0; counter < ptqpertxn; counter++) {
+            // set the cursor to the random key
+            k = ( random() + ( random() << 31 ) ) % max_key;
+            k = (k / idx_stride) * idx_stride; // put on even stride
+//        printf("k=%llu\n", k); 
+            long_long_to_array(kv, 8, k);
+
+            memset(&key, 0, sizeof key); key.data=kv; key.size=8;
+            memset(&val, 0, sizeof val);
+            r = dbc->c_get(dbc, &key, &val, DB_SET_RANGE);
+            if ( r != 0 ) {
+                array_to_long_long(&k, key.data, key.size);
+                printf("error k = %llu\n", k);
+                assert(0);
+            }
+        }
+        double tdiff = gettime() - tstart;
+        printf("%d point queries in %fs at %.0f ptq/s\n", counter, tdiff, counter / tdiff); fflush(stdout);
+    }
+    r = dbc->c_close(dbc);                                      
+    assert(r==0);
+}
+
 static void scanscan_range (void) {
     int r;
 
@@ -462,6 +548,7 @@ static int test_main (int argc, char *const argv[]) {
     case RUN_VERIFY: scanscan_verify(); break;
 #endif
     case RUN_RANGE:  scanscan_range();  break;
+    case RUN_PT:     scanscan_ptquery(); break;
     default:         assert(0);         break;
     }
     scanscan_shutdown();