refs #5241, make completion queues in cachetable unnecessary

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

ft/cachetable.c

@@ -2724,56 +2724,25 @@ int toku_cachetable_unpin_and_remove (
             // we don't want any NEW threads to try to grab the PAIR
             // lock.
             //
-            // Because we call cachetable_remove_pair and setup a completion queue,
+            // Because we call cachetable_remove_pair and wait,
             // the threads that may be waiting
-            // on this PAIR lock must be careful to do NOTHING with the PAIR because
-            // it notices a completion queue. As per our analysis above, we only need
+            // on this PAIR lock must be careful to do NOTHING with the PAIR 
+            // As per our analysis above, we only need
             // to make sure the checkpoint thread and get_and_pin_nonblocking do
             // nothing, and looking at those functions, it is clear they do nothing.
             // 
             cachetable_remove_pair(ct, p);
             if (nb_mutex_blocked_writers(&p->value_nb_mutex)>0) {
-                struct workqueue cq;
-                workqueue_init(&cq);
-                while (nb_mutex_blocked_writers(&p->value_nb_mutex)>0) {
-                    //Someone (one or more checkpoint threads) is waiting for a write lock
-                    //on this pair.
-                    //They are still blocked because we have not released the
-                    //cachetable lock.
-                    //If we freed the memory for the pair we would have dangling
-                    //pointers.  We need to let the other threads finish up with
-                    //this pair.
-
-                    p->cq = &cq;
-
-                    //  If anyone is waiting on write lock, let them finish.
-                    cachetable_unlock(ct);
-
-                    WORKITEM wi = NULL;
-                    r = workqueue_deq(&cq, &wi, 1);
-                    //Writer is now done.
-                    assert(r == 0);
-                    PAIR pp = (PAIR) workitem_arg(wi);
-                    assert(pp == p);
-
-                    //We are holding the write lock on the pair
-                    cachetable_lock(ct);
-                    assert(nb_mutex_writers(&p->value_nb_mutex) == 1);
-                    // let's also assert that this PAIR was not somehow marked 
-                    // as pending a checkpoint. Above, when calling
-                    // remove_key(), we cleared the dirty bit so that
-                    // this PAIR cannot be marked for checkpoint, so let's
-                    // make sure that our assumption is valid.
-                    assert(!p->checkpoint_pending);
-                    assert(p->attr.cache_pressure_size == 0);
-                    nb_mutex_unlock(&p->value_nb_mutex);
-                    // Because we assume it is just the checkpoint thread
-                    // that may have been blocked (as argued above),
-                    // it is safe to simply remove the PAIR from the 
-                    // cachetable. We don't need to write anything out.
-                }
-                p->cq = NULL;
-                workqueue_destroy(&cq);
+                toku_cond_t cond;
+                toku_cond_init(&cond, NULL);
+                nb_mutex_wait_for_users(
+                    &p->value_nb_mutex,
+                    ct->mutex,
+                    &cond
+                    );
+                toku_cond_destroy(&cond);
+                assert(!p->checkpoint_pending);
+                assert(p->attr.cache_pressure_size == 0);
             }
             // just a sanity check
             assert(nb_mutex_users(&p->disk_nb_mutex) == 0);

ft/nonblocking_mutex.h

@@ -56,6 +56,15 @@ static inline void nb_mutex_unlock(NB_MUTEX nb_mutex) {
     rwlock_write_unlock(&nb_mutex->lock);
 }
 
+static inline void nb_mutex_wait_for_users(
+    NB_MUTEX nb_mutex, 
+    toku_mutex_t *mutex, 
+    toku_cond_t* cond
+    ) 
+{
+    rwlock_wait_for_users(&nb_mutex->lock, mutex, cond);
+}
+
 // returns: the number of writers who are waiting for the lock
 
 static inline int nb_mutex_blocked_writers(NB_MUTEX nb_mutex) {

ft/rwlock.h

@@ -36,8 +36,16 @@ struct rwlock {
     int writer;                  // the number of writers
     int want_write;              // the number of blocked writers
     toku_cond_t wait_write;
+    toku_cond_t* wait_users_go_to_zero;
 };
 
+// returns: the sum of the number of readers, pending readers, writers, and
+// pending writers
+
+static inline int rwlock_users(RWLOCK rwlock) {
+    return rwlock->reader + rwlock->want_read + rwlock->writer + rwlock->want_write;
+}
+
 // initialize a read write lock
 
 static __attribute__((__unused__))
@@ -47,6 +55,7 @@ rwlock_init(RWLOCK rwlock) {
     toku_cond_init(&rwlock->wait_read, 0);
     rwlock->writer = rwlock->want_write = 0;
     toku_cond_init(&rwlock->wait_write, 0);
+    rwlock->wait_users_go_to_zero = NULL;
 }
 
 // destroy a read write lock
@@ -64,6 +73,7 @@ rwlock_destroy(RWLOCK rwlock) {
 // expects: mutex is locked
 
 static inline void rwlock_read_lock(RWLOCK rwlock, toku_mutex_t *mutex) {
+    assert(!rwlock->wait_users_go_to_zero);
     if (rwlock->writer || rwlock->want_write) {
         rwlock->want_read++;
         while (rwlock->writer || rwlock->want_write) {
@@ -84,12 +94,16 @@ static inline void rwlock_read_unlock(RWLOCK rwlock) {
     if (rwlock->reader == 0 && rwlock->want_write) {
         toku_cond_signal(&rwlock->wait_write);
     }
+    if (rwlock->wait_users_go_to_zero && rwlock_users(rwlock) == 0) {
+        toku_cond_signal(rwlock->wait_users_go_to_zero);
+    }
 }
 
 // obtain a write lock
 // expects: mutex is locked
 
 static inline void rwlock_write_lock(RWLOCK rwlock, toku_mutex_t *mutex) {
+    assert(!rwlock->wait_users_go_to_zero);
     if (rwlock->reader || rwlock->writer) {
         rwlock->want_write++;
         while (rwlock->reader || rwlock->writer) {
@@ -111,6 +125,9 @@ static inline void rwlock_write_unlock(RWLOCK rwlock) {
         toku_cond_signal(&rwlock->wait_write);
     } else if (rwlock->want_read) {
         toku_cond_broadcast(&rwlock->wait_read);
+    }    
+    if (rwlock->wait_users_go_to_zero && rwlock_users(rwlock) == 0) {
+        toku_cond_signal(rwlock->wait_users_go_to_zero);
     }
 }
 
@@ -138,11 +155,18 @@ static inline int rwlock_writers(RWLOCK rwlock) {
     return rwlock->writer;
 }
 
-// returns: the sum of the number of readers, pending readers, writers, and
-// pending writers
-
-static inline int rwlock_users(RWLOCK rwlock) {
-    return rwlock->reader + rwlock->want_read + rwlock->writer + rwlock->want_write;
+static inline void rwlock_wait_for_users(
+    RWLOCK rwlock, 
+    toku_mutex_t *mutex, 
+    toku_cond_t* cond
+    ) 
+{
+    assert(!rwlock->wait_users_go_to_zero);
+    while (rwlock_users(rwlock) > 0) {
+        rwlock->wait_users_go_to_zero = cond;
+        toku_cond_wait(cond, mutex);
+    }
+    rwlock->wait_users_go_to_zero = NULL;
 }
 
 #if defined(__cplusplus) || defined(__cilkplusplus)