[t:2506] [t:2507] Fix bug in toku_cachetable_unpin_and_remove

Correctly handles a blocked writer (checkpoint).  it checked state before state was set,
so incorrectly thought that there was never a blocked writer

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

newbrt/cachetable.c

@@ -1878,23 +1878,36 @@ int toku_cachetable_unpin_and_remove (CACHEFILE cachefile, CACHEKEY key) {
 	if (p->key.b==key.b && p->cachefile==cachefile) {
 	    p->dirty = CACHETABLE_CLEAN; // clear the dirty bit.  We're just supposed to remove it.
 	    assert(rwlock_readers(&p->rwlock)==1);
-            assert(rwlock_users(&p->rwlock) == 1); //Debug test 1
             rwlock_read_unlock(&p->rwlock);
+            if (rwlock_blocked_writers(&p->rwlock)>0) {
+                //Someone (checkpoint thread) 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 checkpoint thread finish up with
+                //this pair.
+                assert(rwlock_blocked_writers(&p->rwlock)==1); //Only one checkpoint thread.
+                //  If anyone is waiting on write lock, let them finish.
                 struct workqueue cq;
                 workqueue_init(&cq);
                 p->cq = &cq;
-            if (p->state == CTPAIR_IDLE)
-                flush_and_maybe_remove(ct, p, FALSE);
-            cachetable_unlock(ct);
                 WORKITEM wi = 0;
-            r = workqueue_deq(&cq, &wi, 1);
-            cachetable_lock(ct);
-            assert(rwlock_writers(&p->rwlock) == 1); //Debug test 2
-            assert(rwlock_users(&p->rwlock) == 1); //Debug test 2
+                r = workqueue_deq(&cq, &wi, 0);
+                //Writer is now done.
+                assert(r == 0);
                 PAIR pp = workitem_arg(wi);
-            assert(r == 0 && pp == p);
+                assert(pp == p);
+                //We are holding the write lock on the pair
+                assert(rwlock_writers(&p->rwlock) == 1);
+                assert(rwlock_users(&p->rwlock) == 1);
                 cachetable_complete_write_pair(ct, p, TRUE);
                 workqueue_destroy(&cq);
+            }
+            else {
+                //Remove pair.
+                cachetable_maybe_remove_and_free_pair(ct, p);
+            }
             r = 0;
 	    goto done;
 	}

newbrt/rwlock.h

@@ -149,6 +149,12 @@ static inline int rwlock_readers(RWLOCK rwlock) {
 
 // returns: the number of writers
 
+static inline int rwlock_blocked_writers(RWLOCK rwlock) {
+    return rwlock->want_write;
+}
+
+// returns: the number of writers
+
 static inline int rwlock_writers(RWLOCK rwlock) {
     return rwlock->writer;
 }