merge parallel block compressor from 3.0.4 to main closes[t:2371]

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

newbrt/backwards_10.c

@@ -493,6 +493,55 @@ enum { uncompressed_magic_len_10 = (8 // tokuleaf or tokunode
 				 ) 
 };
 
+#define DO_DECOMPRESS_WORKER 1
+
+struct decompress_work_10 {
+    toku_pthread_t id;
+    void *compress_ptr;
+    void *uncompress_ptr;
+    u_int32_t compress_size;
+    u_int32_t uncompress_size;
+};
+
+// initialize the decompression work
+static void init_decompress_work_10(struct decompress_work_10 *w,
+                                 void *compress_ptr, u_int32_t compress_size,
+                                 void *uncompress_ptr, u_int32_t uncompress_size) {
+    memset(&w->id, 0, sizeof(w->id));
+    w->compress_ptr = compress_ptr; w->compress_size = compress_size;
+    w->uncompress_ptr = uncompress_ptr; w->uncompress_size = uncompress_size;
+}
+
+// do the decompression work
+static void do_decompress_work_10(struct decompress_work_10 *w) {
+    uLongf destlen = w->uncompress_size;
+    int r = uncompress(w->uncompress_ptr, &destlen,
+                       w->compress_ptr, w->compress_size);
+    assert(destlen==w->uncompress_size);
+    assert(r==Z_OK);
+}
+
+#if DO_DECOMPRESS_WORKER
+
+static void *decompress_worker_10(void *);
+
+static void start_decompress_work_10(struct decompress_work_10 *w) {
+    int r = toku_pthread_create(&w->id, NULL, decompress_worker_10, w); assert(r == 0);
+}
+
+static void wait_decompress_work_10(struct decompress_work_10 *w) {
+    void *ret;
+    int r = toku_pthread_join(w->id, &ret); assert(r == 0);
+}
+
+static void *decompress_worker_10(void *arg) {
+    struct decompress_work_10 *w = (struct decompress_work_10 *) arg;
+    do_decompress_work_10(w);
+    return arg;
+}
+
+#endif
+
 static int
 decompress_brtnode_from_raw_block_into_rbuf_10(u_int8_t *raw_block, struct rbuf *rb, BLOCKNUM blocknum) {
     int r;
@@ -534,24 +583,24 @@ decompress_brtnode_from_raw_block_into_rbuf_10(u_int8_t *raw_block, struct rbuf
 
     // decompress the sub blocks
     unsigned char *uncompressed_data = rb->buf+uncompressed_magic_len_10;
-    struct decompress_work decompress_work[n_sub_blocks];
+    struct decompress_work_10 decompress_work[n_sub_blocks];
 
     for (i=0; i<n_sub_blocks; i++) {
-        init_decompress_work(&decompress_work[i], compressed_data, sub_block_sizes[i].compressed_size, uncompressed_data, sub_block_sizes[i].uncompressed_size);
+        init_decompress_work_10(&decompress_work[i], compressed_data, sub_block_sizes[i].compressed_size, uncompressed_data, sub_block_sizes[i].uncompressed_size);
         if (i>0) {
 #if DO_DECOMPRESS_WORKER
-            start_decompress_work(&decompress_work[i]);
+            start_decompress_work_10(&decompress_work[i]);
 #else
-            do_decompress_work(&decompress_work[i]);
+            do_decompress_work_10(&decompress_work[i]);
 #endif
         }
         uncompressed_data += sub_block_sizes[i].uncompressed_size;
         compressed_data += sub_block_sizes[i].compressed_size;
     }
-    do_decompress_work(&decompress_work[0]);
+    do_decompress_work_10(&decompress_work[0]);
 #if DO_DECOMPRESS_WORKER
     for (i=1; i<n_sub_blocks; i++)
-        wait_decompress_work(&decompress_work[i]);
+        wait_decompress_work_10(&decompress_work[i]);
 #endif
     toku_trace("decompress done");
 

newbrt/brt.c

@@ -5175,11 +5175,15 @@ int toku_brt_init(void (*ydb_lock_callback)(void), void (*ydb_unlock_callback)(v
         r = toku_brt_lock_init();
     if (r==0) 
         r = toku_checkpoint_init(ydb_lock_callback, ydb_unlock_callback);
+    if (r == 0)
+        r = toku_brt_serialize_init();
     return r;
 }
 
 int toku_brt_destroy(void) {
     int r = 0;
+    if (r == 0)
+        r = toku_brt_serialize_destroy();
     if (r==0) 
         r = toku_brt_lock_destroy();
     if (r==0)

newbrt/brt.h

@@ -189,6 +189,8 @@ int toku_brt_init(void (*ydb_lock_callback)(void), void (*ydb_unlock_callback)(v
 int toku_brt_destroy(void);
 int toku_pwrite_lock_init(void);
 int toku_pwrite_lock_destroy(void);
+int toku_brt_serialize_init(void);
+int toku_brt_serialize_destroy(void);
 
 void toku_maybe_truncate_cachefile (CACHEFILE cf, u_int64_t size_used);
 // Effect: truncate file if overallocated by at least 32MiB

newbrt/cachetable.c

@@ -1325,6 +1325,7 @@ int toku_cachetable_get_and_pin(CACHEFILE cachefile, CACHEKEY key, u_int32_t ful
             }
             if (do_wait_time)
 	        t0 = get_tnow();
+
             if (p->checkpoint_pending) {
 		get_and_pin_footprint = 4;		
 		write_pair_for_checkpoint(ct, p);

newbrt/workset.h

+#ifndef _TOKU_WORKSET_H
+#define _TOKU_WORKSET_H
+
+#include <toku_list.h>
+#include <toku_pthread.h>
+
+// the work struct is the base class for work to be done by some threads
+struct work {
+    struct toku_list next;
+};
+
+// the workset struct contains the set of work to be done by some threads
+// the lock protects the work list
+struct workset {
+    pthread_mutex_t lock;
+    struct toku_list worklist;
+};
+
+static inline void workset_init(struct workset *ws) {
+    int r = toku_pthread_mutex_init(&ws->lock, NULL); assert(r == 0);
+    toku_list_init(&ws->worklist);
+};
+
+static inline void workset_destroy(struct workset *ws) {
+    assert(toku_list_empty(&ws->worklist));
+    int r = toku_pthread_mutex_destroy(&ws->lock); assert(r == 0);
+}
+
+static inline void workset_lock(struct workset *ws) {
+    int r = toku_pthread_mutex_lock(&ws->lock); assert(r == 0);
+}
+        
+static inline void workset_unlock(struct workset *ws) {
+    int r = toku_pthread_mutex_unlock(&ws->lock); assert(r == 0);
+}
+
+// put work in the workset 
+static inline void workset_put(struct workset *ws, struct work *w) {
+    workset_lock(ws);
+    toku_list_push(&ws->worklist, &w->next);
+    workset_unlock(ws);
+}
+
+// put work in the workset.  assume already locked.
+static inline void workset_put_locked(struct workset *ws, struct work *w) {
+    toku_list_push(&ws->worklist, &w->next);
+}
+
+// get work from the workset
+static inline struct work *workset_get(struct workset *ws) {
+    workset_lock(ws);
+    struct work *w = NULL;
+    if (!toku_list_empty(&ws->worklist)) {
+        struct toku_list *l = toku_list_pop_head(&ws->worklist);
+        w = toku_list_struct(l, struct work, next);
+    }
+    workset_unlock(ws);
+    return w;
+}
+
+// create a set of threads to run a given function
+// tids will contain the thread id's of the created threads
+// *ntids on input contains the number of threads requested, on output contains the number of threads created
+static inline void threadset_create(toku_pthread_t tids[], int *ntids, void *(*f)(void *arg), void *arg) {
+    int n = *ntids;
+    int i;
+    for (i = 0; i < n; i++) {
+        int r = toku_pthread_create(&tids[i], NULL, f, arg); 
+        if (r != 0)
+            break;
+    }
+    *ntids = i;
+}
+
+// join with a set of threads
+static inline void threadset_join(toku_pthread_t tids[], int ntids) {
+    for (int i = 0; i < ntids; i++) {
+        void *ret;
+        int r = toku_pthread_join(tids[i], &ret); assert(r == 0);
+    }
+}
+
+#endif