#2907 closes[t:2907] merge single insert per txn profiler to main

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

db-benchmark-test/Makefile

@@ -26,7 +26,8 @@ SCANSCAN_BDB = scanscan-bdb$(BINSUF)
 TARGET_TDB = db-benchmark-test-tokudb$(BINSUF)
 SCANSCAN_TDB = scanscan-tokudb$(BINSUF)
 SCANRACE_TDB = scanrace-tokudb$(BINSUF)
-TARGETS = $(TARGET_BDB) $(SCANSCAN_BDB) $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB)
+MULTIBENCH_TDB = multi-bench-tokudb$(BINSUF)
+TARGETS = $(TARGET_BDB) $(SCANSCAN_BDB) $(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB)
 TARGETS_BDB = $(TARGET_BDB) $(SCANSCAN_BDB)
 
 ifeq ($(OS_CHOICE),windows)
@@ -67,7 +68,7 @@ build: build.tdb build.bdb;
 endif
 
 build.bdb: $(TARGET_BDB) $(SCANSCAN_BDB) $(WINDOWS_BDB_LIB_NAME)
-build.tdb: $(TARGET_TDB) $(SCANSCAN_TDB)
+build.tdb: $(TARGET_TDB) $(SCANSCAN_TDB) $(MULTIBENCH_TDB)
 
 check: check-default check-rowsize check-xfast check-x check-no-rollback check-4G child.benchmark.dir
 
@@ -114,7 +115,7 @@ clean:
 	rm -f $(TARGETS) 4g.out
 	rm -rf *.dir $(BENCHDBS)
 
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_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): BIN_FROM_C_FLAGS+=-DDIRSUF=tokudb -I$(PORTABILITY_HEADERS) -I$(TOKUROOT)toku_include -I$(TOKUROOT)include
 
 ifneq ($(PROF),)
     USE_STATIC=1
@@ -122,7 +123,7 @@ endif
 
 # A hack to make gprof work.  See #515.
 ifneq ($(USE_STATIC),)
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB): DLINK_FILES=
+$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB) $(MULTIBENCH_TDB): DLINK_FILES=
 
 OFILES = \
 	$(TOKUROOT)src/ydb.lib \
@@ -138,14 +139,16 @@ $(SCANSCAN_TDB): scanscan.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)
+	$(CC) $< $(BIN_FROM_C_FLAGS) $(OFILES) $(LINK_MUST_BE_LAST)
 
 else
 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): DLINK_FILES=$(TDB_DLINK_FILES)
-$(TARGET_TDB) $(SCANSCAN_TDB) $(SCANRACE_TDB): RPATH_DIRS=$(dir $(TDB_DLINK_FILES))
+$(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))
 endif
 $(TARGET_TDB): db-benchmark-test.c $(PTHREAD_LOCAL)
 	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
@@ -153,6 +156,8 @@ $(SCANSCAN_TDB): scanscan.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)
+	$(CC) $< $(BIN_FROM_C_FLAGS) $(LINK_MUST_BE_LAST)
 endif
 
 $(TARGETS_BDB): CPPFLAGS+=-DDIRSUF=bdb  -DTOKU_ALLOW_DEPRECATED

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

@@ -12,6 +12,9 @@
 #include <stdlib.h>
 #include <sys/stat.h>
 #include <unistd.h>
+#if defined(TOKUDB)
+#include "trace_mem.h"
+#endif
 
 #if !defined(DB_YESOVERWRITE)
 #define DB_YESOVERWRITE 0
@@ -362,6 +365,12 @@ static void fill_array (unsigned char *data, int size) {
     }
 }
 
+static void trace(const char *UU(s), int UU(n)) {
+#if defined(TOKUDB)
+    toku_add_trace_mem(s, n);
+#endif
+}
+
 static void insert (long long v) {
     int r;
     unsigned char kc[keysize];
@@ -384,16 +393,22 @@ static void insert (long long v) {
     else {
         for (which = 0; which < num_dbs; which++) {
             DB *db = dbs[which];
+            trace("pstart", __LINE__);
             r = db->put(db, tid, &kt, &vt, put_flags);
+            trace("pdone", __LINE__);
             CKERR(r);
         }
     }
     if (do_transactions) {
 	if (n_insertions_since_txn_began>=items_per_transaction && !singlex) {
 	    n_insertions_since_txn_began=0;
+            trace("cstart", __LINE__);
 	    r = tid->commit(tid, commitflags); assert(r==0);
+            trace("cdone", __LINE__);
             tid = NULL;
+            trace("txn", __LINE__);
 	    r=dbenv->txn_begin(dbenv, 0, &tid, 0); assert(r==0);
+            trace("txndone", __LINE__);
             for (which = 0; which < num_dbs; which++) {
                 DB *db = dbs[which];
                 do_prelock(db, tid);
@@ -727,6 +742,10 @@ static int test_main (int argc, char *const argv[]) {
 	print_hash_histogram();
     }
 #endif
+#if defined(TOKUDB)
+    // if tokudb has tracing enabled (see trace_mem.h) then this will dump the trace data
+    if (1) toku_print_trace_mem(stderr);
+#endif
 
     return 0;
 }

db-benchmark-test/multi-bench.c

+#include <stdio.h>
+#include <assert.h>
+#include <stdint.h>
+#include <inttypes.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <pthread.h>
+#include <sys/stat.h>
+#include <db.h>
+#include <byteswap.h>
+
+static int verbose = 0;
+static int commit_flag = 0;
+static int do_verify = 0;
+static int nthread = 1;
+static int maxk = 1000000;
+static int do_multidb = 0;
+
+static int usage(const char *prog) {
+    fprintf(stderr, "%s: run multi client single row insertions\n", prog);
+    fprintf(stderr, "[--n %d]\n", maxk);
+    fprintf(stderr, "[--nthread %d]\n", nthread);
+    fprintf(stderr, "[--verify] (%d)\n", do_verify);
+    fprintf(stderr, "[--nosync] (%d)\n", commit_flag);
+    fprintf(stderr, "[--multidb] (%d)\n", do_multidb);
+    return 1;
+}
+
+struct keygen {
+    pthread_mutex_t lock;
+    uint64_t k;
+    uint64_t maxk;
+};
+
+static void keygen_init(struct keygen *key, uint64_t _maxk) {
+    pthread_mutex_init(&key->lock, NULL);
+    key->k = 0;
+    key->maxk = _maxk;
+}
+
+static void keygen_destroy(struct keygen *key) {
+    pthread_mutex_destroy(&key->lock);
+}
+
+static int keygen_next(struct keygen *key, uint64_t *k) {
+    int r;
+    pthread_mutex_lock(&key->lock);
+    if (key->k >= key->maxk)
+        r = 1;
+    else {
+        *k = key->k++;
+        r = 0;
+    }
+    pthread_mutex_unlock(&key->lock);
+    return r;
+}
+
+struct inserter_arg {
+    struct keygen *keygen;
+    DB_ENV *env;
+    DB *db;
+};
+
+static int inserter(struct keygen *keygen, DB_ENV *env, DB *db) {
+    if (verbose) printf("%p %p %p\n", keygen, env, db);
+    while (1) {
+        uint64_t k;
+        int r = keygen_next(keygen, &k);
+        if (r != 0)
+            break;
+        
+        if (verbose) printf("%d: %"PRIu64"\n", (int) pthread_self(), k);
+
+        DB_TXN *txn;
+        r = env->txn_begin(env, NULL, &txn, 0);
+        assert(r == 0);
+
+        uint64_t kk = bswap_64(k);
+        DBT key = { .data = &kk, .size = sizeof kk };
+        DBT val = { .data = &k, .size = sizeof k };
+        r = db->put(db, txn, &key, &val, DB_YESOVERWRITE);
+        assert(r == 0);
+
+        r = txn->commit(txn, commit_flag);
+        assert(r == 0);
+    }
+    return 0;
+}
+
+static void *inserter_wrap(void *arg) {
+    if (verbose) printf("%d\n", (int) pthread_self());
+    struct inserter_arg *inserter_arg = (struct inserter_arg *) arg;
+    int r = inserter(inserter_arg->keygen, inserter_arg->env, inserter_arg->db);
+    assert(r == 0);
+    return arg;
+}
+
+static int verify(DB_ENV *env, DB *db, uint64_t _maxk) {
+    int r;
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0);
+    assert(r == 0);
+
+    DBC *cursor;
+    r = db->cursor(db, txn, &cursor, 0);
+    assert(r == 0);
+
+    DBT key; memset(&key, 0, sizeof key);
+    DBT val; memset(&val, 0, sizeof val);
+    uint64_t i;
+    for (i=0; 1; i++) {
+        r = cursor->c_get(cursor, &key, &val, DB_NEXT);
+        if (r != 0)
+            break;
+        uint64_t k, v;
+        assert(key.size == sizeof k);
+        assert(val.size == sizeof v);
+        memcpy(&k, key.data, key.size);
+        k = bswap_64(k);
+        assert(i == k);
+        memcpy(&v, val.data, val.size);
+        assert(v == i);
+    }
+    assert(i == _maxk);
+
+    r = cursor->c_close(cursor);
+    assert(r == 0);
+    r = txn->commit(txn, 0);
+    assert(r == 0);
+    return 0;
+}
+
+static int env_init(DB_ENV **envptr, const char *envdir) {
+    int r;
+    DB_ENV *env;
+
+    r = db_env_create(&env, 0);
+    if (r == 0) {
+        // env setup
+
+        // env open
+        r = env->open(env, envdir, DB_CREATE+DB_PRIVATE+DB_INIT_LOCK+DB_INIT_LOG+DB_INIT_MPOOL+DB_INIT_TXN, 0777);
+    }
+    if (r == 0)
+        *envptr = env;
+    return r;
+}
+
+static int db_init(DB_ENV *env, const char *dbname, DB **dbptr) {
+    int r;
+    DB *db;
+
+    r = db_create(&db, env, 0);
+    if (r == 0) {
+        // db create
+        r = db->open(db, NULL, dbname, NULL, DB_BTREE, DB_CREATE, 0777);
+        if (r != 0) {
+            r = db->close(db, 0);
+            assert(r == 0);
+        }
+    }
+    if (r == 0)
+        *dbptr = db;
+    return r;
+}   
+
+int main(int argc, char *argv[]) {
+    int r;
+
+    for (int i = 1; i < argc; i++) {
+        char *arg = argv[i];
+        if (strcmp(arg, "--nthread") == 0 && i+1 < argc) {
+            nthread = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--n") == 0 && i+1 < argc) {
+            maxk = atoi(argv[++i]);
+            continue;
+        }
+        if (strcmp(arg, "--verbose") == 0) {
+            verbose++;
+            continue;
+        }
+        if (strcmp(arg, "--nosync") == 0) {
+            commit_flag = DB_TXN_NOSYNC;
+            continue;
+        }
+        if (strcmp(arg, "--verify") == 0) {
+            do_verify++;
+            continue;
+        }
+        if (strcmp(arg, "--multidb") == 0) {
+            do_multidb++;
+            continue;
+        }
+
+        return usage(argv[0]);
+    }
+
+    int ndb = 1;
+    int nkeygen = 1;
+    if (do_multidb) {
+        ndb = nthread;
+        nkeygen = nthread;
+    }
+
+    const char *envdir = "dir." __FILE__;
+    r = system("rm -rf dir." __FILE__);
+    assert(r == 0);
+    r = mkdir(envdir, 0777);
+    assert(r == 0);
+
+    struct keygen keygen[nkeygen];
+    for (int i = 0; i < nkeygen; i++) 
+        keygen_init(&keygen[i], maxk);
+
+    DB_ENV *env;
+    r = env_init(&env, envdir);
+    assert(r == 0);
+
+    DB *db[ndb];
+    for (int i = 0 ; i < ndb; i++) {
+        char dbname[32]; sprintf(dbname, "db%d", i);
+        r = db_init(env, dbname, &db[i]);
+        assert(r == 0);
+    }
+
+    pthread_t tids[nthread];
+    struct inserter_arg args[nthread];
+    for (int i = 0; i < nthread; i++) {
+        struct inserter_arg this_arg = { .keygen = &keygen[i % nkeygen], .env = env, .db = db[i % ndb] };
+        args[i] = this_arg;
+        pthread_create(&tids[i], NULL, inserter_wrap, &args[i]);
+    }
+    for (int i = 0; i < nthread; i++) {
+        void *retptr;
+        pthread_join(tids[i], &retptr);
+    }
+
+    for (int i = 0; i < ndb; i++) {
+        if (do_verify) 
+            verify(env, db[i], maxk);
+        r = db[i]->close(db[i], 0);
+        assert(r == 0);
+    }
+
+    r = env->close(env, 0);
+    assert(r == 0);
+
+    for (int i = 0; i < nkeygen; i++)
+        keygen_destroy(&keygen[i]);
+
+    return 0;
+}

db-benchmark-test/scanscan.c

@@ -460,11 +460,8 @@ static int test_main (int argc, char *const argv[]) {
     scanscan_shutdown();
 
 #if defined(TOKUDB)
-    // if tokudb has tracing enabled (see trace_mem.h) then this will dump
-    // the trace data
-    if (0) {
-        toku_print_trace_mem();
-    }
+    // if tokudb has tracing enabled (see trace_mem.h) then this will dump the trace data
+    if (1) toku_print_trace_mem(stderr);
 #endif
 #if defined(__linux__) && __linux__
     if (verbose) {

db-benchmark-test/tracedelta.py

+import sys
+
+def main():
+    ts = None
+    while 1:
+        b = sys.stdin.readline()
+        if b == "": break
+        f = b.split()
+        if len(f) != 2: continue
+        newts = int(f[0])
+        event = f[1]
+        if ts is None:
+            ts = int(f[0])
+        else:
+            print "%8d %s" % (newts - ts, event)
+            ts = newts
+    return 0
+sys.exit(main())

newbrt/logger.c

@@ -4,6 +4,7 @@
 #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 "includes.h"
+#include "trace_mem.h"
 
 static const int log_format_version=TOKU_LOG_VERSION;
 
@@ -735,11 +736,13 @@ int toku_logger_maybe_fsync (TOKULOGGER logger, LSN lsn, int do_fsync)
 	write_outbuf_to_logfile(logger, &fsynced_lsn);
 	if (fsynced_lsn.lsn < lsn.lsn) {
 	    // it may have gotten fsynced by the write_outbuf_to_logfile.
+            toku_add_trace_mem("fsync", __LINE__);
 	    r = toku_file_fsync_without_accounting(logger->fd);
 	    if (r!=0) {
 		toku_logger_panic(logger, r);
 		return r;
 	    }
+            toku_add_trace_mem("fsyncdone", __LINE__);
 	    assert(fsynced_lsn.lsn <= logger->written_lsn.lsn);
 	    fsynced_lsn = logger->written_lsn;
 	}

newbrt/trace_mem.c

@@ -9,6 +9,40 @@
 #include "rdtsc.h"
 #include "trace_mem.h"
 
+// customize this as required
+#define NTRACE 0
+#if NTRACE
+static struct toku_trace {
+    const char *str;
+    int n;
+    unsigned long long ts;
+} toku_trace[NTRACE];
+
+static int toku_next_trace = 0;
+#endif
+
+void toku_add_trace_mem (const char *UU(str), int UU(n)) {
+#if USE_RDTSC && NTRACE
+    int i = toku_next_trace++;
+    if (toku_next_trace >= NTRACE) toku_next_trace = 0;
+    toku_trace[i].ts = rdtsc();
+    toku_trace[i].str = str;
+    toku_trace[i].n = n;
+#endif
+}
+
+void toku_print_trace_mem(FILE *UU(out)) {
+#if NTRACE
+    int i = toku_next_trace;
+    do {
+        if (toku_trace[i].str)
+            fprintf(out, "%llu %s:%d\n", toku_trace[i].ts, toku_trace[i].str, toku_trace[i].n);
+        i++;
+        if (i >= NTRACE) i = 0;
+    } while (i != toku_next_trace);
+#endif
+}
+
 #if BL_DO_TRACE && BL_SIMPLE_TRACE
 
 static double saved_scale_factor = 1e-9; // approximate until we get the actual scale factor
@@ -57,40 +91,6 @@ void bl_trace_end(void) {
 
 #elif BL_DO_TRACE && !BL_SIMPLE_TRACE
 
-// customize this as required
-#define NTRACE 0
-#if NTRACE
-static struct toku_trace {
-    const char *str;
-    int n;
-    unsigned long long ts;
-} toku_trace[NTRACE];
-
-static int toku_next_trace = 0;
-#endif
-
-void toku_add_trace_mem (const char *str __attribute__((unused)),
-			 int n __attribute__((unused))) {
-#if USE_RDTSC && NTRACE
-    int i = toku_next_trace++;
-    if (toku_next_trace >= NTRACE) toku_next_trace = 0;
-    toku_trace[i].ts = rdtsc();
-    toku_trace[i].str = str;
-    toku_trace[i].n = n;
-#endif
-}
-
-void toku_print_trace_mem(void) {
-#if NTRACE
-    int i = toku_next_trace;
-    do {
-        if (toku_trace[i].str)
-            printf("%llu %s:%d\n", toku_trace[i].ts, toku_trace[i].str, toku_trace[i].n);
-        i++;
-        if (i >= NTRACE) i = 0;
-    } while (i != toku_next_trace);
-#endif
-}
 
 /*
  *   Some trace functions (similar) added for the bulk loader
@@ -163,3 +163,4 @@ void bl_trace_end(void)
 }
 
 #endif
+

newbrt/trace_mem.h

@@ -10,8 +10,12 @@ extern "C" {
 #endif
 
 // Define this even when traces are compiled out so we don't have to recompile things like scanscan.c
+
 // print the trace
-void toku_print_trace_mem(void) __attribute__((__visibility__("default")));
+void toku_print_trace_mem(FILE *) __attribute__((__visibility__("default")));
+
+// add an entry to the end of the trace which consists of a string pointer, a number, and the processor timestamp
+void toku_add_trace_mem(const char *str, int n) __attribute__((__visibility__("default")));
 
 #define BL_DO_TRACE 0
 // BL_SIMPLE_TRACE 1 is Bradley's in-memory trace analysis.
@@ -29,10 +33,6 @@ void toku_print_trace_mem(void) __attribute__((__visibility__("default")));
 // performance: we trade speed for size by not compressing the trace
 // entries.
 
-// add an entry to the end of the trace which consists of a string
-// pointer, a number, and the processor timestamp
-void toku_add_trace_mem(const char *str, int n) __attribute__((__visibility__("default")));
-
 //  some trace functions added for the bulk loader
 void bl_trace(const char *func __attribute__((unused)), 
               int         line __attribute__ ((unused)),