closes #5211, closes #5215, closes 5224, merge to main

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

ft/CMakeLists.txt

@@ -23,6 +23,7 @@ add_custom_target(
   )
 
 set(FT_SOURCES
+  background_job_manager.c
   block_allocator.c
   block_table.c
   cachetable.c

ft/background_job_manager.c

@@ -0,0 +1,73 @@
+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+#ident "Copyright (c) 2011 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."
+
+#include <config.h>
+#include <stdbool.h>
+#include <toku_pthread.h>
+#include "kibbutz.h"
+#include "background_job_manager.h"
+#include "includes.h"
+
+struct background_job_manager_struct {
+    bool accepting_jobs;
+    u_int32_t num_jobs;
+    toku_cond_t jobs_wait;
+    toku_mutex_t jobs_lock;
+};
+
+void bjm_init(BACKGROUND_JOB_MANAGER* pbjm) {
+    BACKGROUND_JOB_MANAGER XCALLOC(bjm);
+    toku_mutex_init(&bjm->jobs_lock, 0);    
+    toku_cond_init(&bjm->jobs_wait, NULL);
+    bjm->accepting_jobs = true;
+    bjm->num_jobs = 0;
+    *pbjm = bjm;
+}
+
+void bjm_destroy(BACKGROUND_JOB_MANAGER bjm) {
+    assert(bjm->num_jobs == 0);
+    toku_cond_destroy(&bjm->jobs_wait);
+    toku_mutex_destroy(&bjm->jobs_lock);
+    toku_free(bjm);
+}
+
+void bjm_reset(BACKGROUND_JOB_MANAGER bjm) {
+    assert(bjm->num_jobs == 0);
+    bjm->accepting_jobs = true;
+}
+
+int bjm_add_background_job(BACKGROUND_JOB_MANAGER bjm) {
+    int ret_val;
+    toku_mutex_lock(&bjm->jobs_lock);
+    if (bjm->accepting_jobs) {
+        bjm->num_jobs++;
+        ret_val = 0;
+    }
+    else {
+        ret_val = -1;
+    }
+    toku_mutex_unlock(&bjm->jobs_lock);
+    return ret_val;
+}
+void bjm_remove_background_job(BACKGROUND_JOB_MANAGER bjm){
+    toku_mutex_lock(&bjm->jobs_lock);
+    assert(bjm->num_jobs > 0);
+    bjm->num_jobs--;
+    if (bjm->num_jobs == 0 && !bjm->accepting_jobs) {
+        toku_cond_broadcast(&bjm->jobs_wait);
+    }
+    toku_mutex_unlock(&bjm->jobs_lock);
+}
+
+void bjm_wait_for_jobs_to_finish(BACKGROUND_JOB_MANAGER bjm) {
+    toku_mutex_lock(&bjm->jobs_lock);
+    bjm->accepting_jobs = false;
+    while (bjm->num_jobs > 0) {
+        toku_cond_wait(&bjm->jobs_wait, &bjm->jobs_lock);
+    }
+    toku_mutex_unlock(&bjm->jobs_lock);
+}
+

ft/background_job_manager.h

@@ -0,0 +1,20 @@
+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ifndef BACKGROUND_JOB_MANAGER_H
+#define BACKGROUND_JOB_MANAGER_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."
+
+
+typedef struct background_job_manager_struct *BACKGROUND_JOB_MANAGER;
+
+void bjm_init(BACKGROUND_JOB_MANAGER* bjm);
+void bjm_destroy(BACKGROUND_JOB_MANAGER bjm);
+void bjm_reset(BACKGROUND_JOB_MANAGER bjm);
+
+int bjm_add_background_job(BACKGROUND_JOB_MANAGER bjm);
+void bjm_remove_background_job(BACKGROUND_JOB_MANAGER bjm);
+void bjm_wait_for_jobs_to_finish(BACKGROUND_JOB_MANAGER bjm);
+
+#endif

ft/cachetable.h

@@ -74,9 +74,6 @@ int toku_cachetable_end_checkpoint(CACHETABLE ct, TOKULOGGER logger,
 // Requires no locks be held that are taken by the checkpoint function
 void toku_cachetable_minicron_shutdown(CACHETABLE ct);
 
-// Wait for the cachefile's background work to finish.
-void toku_cachefile_wait_for_background_work_to_quiesce(CACHEFILE cf);
-
 // Close the cachetable.
 // Effects: All of the memory objects are flushed to disk, and the cachetable is destroyed.
 int toku_cachetable_close (CACHETABLE*); /* Flushes everything to disk, and destroys the cachetable. */
@@ -477,8 +474,6 @@ typedef enum {
     CT_MISSTIME,               // how many usec spent waiting for disk read because of cache miss
     CT_PUTS,                   // how many times has a newly created node been put into the cachetable?
     CT_PREFETCHES,             // how many times has a block been prefetched into the cachetable?
-    CT_MAYBE_GET_AND_PINS,     // how many times has maybe_get_and_pin(_clean) been called?
-    CT_MAYBE_GET_AND_PIN_HITS, // how many times has maybe_get_and_pin(_clean) returned with a node?
     CT_SIZE_CURRENT,           // the sum of the sizes of the nodes represented in the cachetable
     CT_SIZE_LIMIT,             // the limit to the sum of the node sizes
     CT_SIZE_MAX,               // high water mark of size_current (max value size_current ever had)
@@ -506,12 +501,9 @@ char * toku_construct_full_name(int count, ...);
 char * toku_cachetable_get_fname_in_cwd(CACHETABLE ct, const char * fname_in_env);
 
 void cachefile_kibbutz_enq (CACHEFILE cf, void (*f)(void*), void *extra);
-// Effect: Add a job to the cachetable's collection of work to do.  Note that function f must call remove_background_job()
+// Effect: Add a job to the cachetable's collection of work to do.  Note that function f must call remove_background_job_from_cf()
 
-void add_background_job (CACHEFILE cf, bool already_locked);
-// Effect: When a kibbutz job or cleaner thread starts working, the
-// cachefile must be notified (so during a close it can wait);
-void remove_background_job (CACHEFILE cf, bool already_locked);
+void remove_background_job_from_cf (CACHEFILE cf);
 // Effect: When a kibbutz job or cleaner thread finishes in a cachefile,
 // the cachetable must be notified.
 

ft/ft-flusher.c

@@ -1758,7 +1758,7 @@ static void flush_node_fun(void *fe_v)
         // It is the responsibility of flush_some_child to unlock the node
         flush_some_child(fe->h, fe->node, &fa);
     }
-    remove_background_job(fe->h->cf, false);
+    remove_background_job_from_cf(fe->h->cf);
     toku_free(fe);
 }
 

ft/kibbutz.h

@@ -1,6 +1,7 @@
 /* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
 // vim: expandtab:ts=8:sw=4:softtabstop=4:
 #ifndef KIBBUTZ_H
+#define KIBBUTZ_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."

ft/tests/cachetable-kibbutz_and_flush_cachefile.c

@@ -15,9 +15,12 @@ static void kibbutz_work(void *fe_v)
     CACHEFILE f1 = fe_v;
     sleep(2);
     foo = TRUE;
-    int r = toku_cachetable_unpin(f1, make_blocknum(1), 1, CACHETABLE_CLEAN, make_pair_attr(8));
+    // note that we make the size 16 to induce an eviction
+    // once evictions are moved to their own thread, we need
+    // to modify this test
+    int r = toku_cachetable_unpin(f1, make_blocknum(1), 1, CACHETABLE_CLEAN, make_pair_attr(16));
     assert(r==0);
-    remove_background_job(f1, false);    
+    remove_background_job_from_cf(f1);    
 }
 
 

ft/tests/cachetable-simple-pin-nonblocking.c

@@ -52,7 +52,7 @@ static void kibbutz_work(void *fe_v)
     foo = TRUE;
     int r = toku_cachetable_unpin(f1, make_blocknum(1), 1, CACHETABLE_CLEAN, make_pair_attr(8));
     assert(r==0);
-    remove_background_job(f1, false);    
+    remove_background_job_from_cf(f1);    
 }
 
 

ft/tests/cachetable-simple-pin.c

@@ -45,7 +45,7 @@ static void kibbutz_work(void *fe_v)
     foo = TRUE;
     int r = toku_cachetable_unpin(f1, make_blocknum(1), 1, CACHETABLE_CLEAN, make_pair_attr(8));
     assert(r==0);
-    remove_background_job(f1, false);    
+    remove_background_job_from_cf(f1);    
 }
 
 static void

ft/tests/test-bjm.c

@@ -0,0 +1,68 @@
+/* -*- mode: C; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id: test-kibbutz2.c 43762 2012-05-22 16:17:53Z yfogel $"
+#ident "Copyright (c) 2011 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."
+
+#include "background_job_manager.h"
+#include "includes.h"
+#include "test.h"
+
+BACKGROUND_JOB_MANAGER bjm; 
+
+static void *finish_bjm(void *arg) {
+    bjm_wait_for_jobs_to_finish(bjm);
+    return arg;
+}
+
+
+static void bjm_test(void) {
+    int r = 0;
+    bjm = NULL;
+    bjm_init(&bjm);
+    // test simple add/remove of background job works
+    r = bjm_add_background_job(bjm);
+    assert_zero(r);
+    bjm_remove_background_job(bjm);
+    bjm_wait_for_jobs_to_finish(bjm);
+    // assert that you cannot add a background job
+    // without resetting bjm after waiting 
+    // for finish
+    r = bjm_add_background_job(bjm);
+    assert(r != 0);
+    // test that after a reset, we can resume adding background jobs
+    bjm_reset(bjm);
+    r = bjm_add_background_job(bjm);
+    assert_zero(r);
+    bjm_remove_background_job(bjm);    
+    bjm_wait_for_jobs_to_finish(bjm);
+
+    bjm_reset(bjm);
+    r = bjm_add_background_job(bjm);
+    assert_zero(r);        
+    toku_pthread_t tid;    
+    r = toku_pthread_create(&tid, NULL, finish_bjm, NULL); 
+    assert_zero(r);
+    usleep(2*1024*1024);
+    // should return non-zero because tid is waiting 
+    // for background jobs to finish
+    r = bjm_add_background_job(bjm);
+    assert(r != 0);
+    bjm_remove_background_job(bjm);
+    void *ret;
+    r = toku_pthread_join(tid, &ret); 
+    assert_zero(r);
+    
+    bjm_destroy(bjm);
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+    
+    bjm_test();
+    if (verbose) printf("test ok\n");
+    return 0;
+}
+
+