FT-197 Store cachefiles in OMTs only, remove linked list pointers

3 files changed, 150 insertions, 188 deletions

diff --git a/ft/cachetable-internal.h b/ft/cachetable-internal.h
index 0e0d1ad0f64..9eb3ec66568 100644
--- a/ft/cachetable-internal.h
+++ b/ft/cachetable-internal.h
@@ -178,8 +178,6 @@ class pair_list;
 // Maps to a file on disk.
 //
 struct cachefile {
-    CACHEFILE next;
-    CACHEFILE prev;
     // these next two fields are protected by cachetable's list lock
     // they are managed whenever we add or remove a pair from
     // the cachetable. As of Riddler, this linked list is only used to
@@ -439,14 +437,12 @@ public:
     bool evict_some_stale_pair(evictor* ev);
     void free_stale_data(evictor* ev);
     // access to these fields are protected by the lock
-    CACHEFILE m_active_head; // head of CACHEFILEs that are active
-    CACHEFILE m_stale_head; // head of CACHEFILEs that are stale
-    CACHEFILE m_stale_tail; // tail of CACHEFILEs that are stale
     FILENUM m_next_filenum_to_use;
     uint32_t m_next_hash_id_to_use;
     toku_pthread_rwlock_t m_lock; // this field is publoc so we are still POD
     toku::omt<CACHEFILE> m_active_filenum;
     toku::omt<CACHEFILE> m_active_fileid;
+    toku::omt<CACHEFILE> m_stale_fileid;
 private:    
     CACHEFILE find_cachefile_in_list_unlocked(CACHEFILE start, struct fileid* fileid);
 };
diff --git a/ft/cachetable.cc b/ft/cachetable.cc
index d7c734cc5fd..d57592286a4 100644
--- a/ft/cachetable.cc
+++ b/ft/cachetable.cc
@@ -4414,43 +4414,48 @@ void checkpointer::increment_num_txns() {
     m_checkpoint_num_txns++;
 }
 
-//
-// Update the user data in any cachefiles in our checkpoint list.
-//
-void checkpointer::update_cachefiles() {
-    CACHEFILE cf;
-    for(cf = m_cf_list->m_active_head; cf; cf=cf->next) {
+struct iterate_begin_checkpoint {
+    LSN lsn_of_checkpoint_in_progress;
+    iterate_begin_checkpoint(LSN lsn) : lsn_of_checkpoint_in_progress(lsn) { }
+    static int fn(const CACHEFILE &cf, const uint32_t UU(idx), struct iterate_begin_checkpoint *info) {
         assert(cf->begin_checkpoint_userdata);
         if (cf->for_checkpoint) {
-            cf->begin_checkpoint_userdata(m_lsn_of_checkpoint_in_progress,
-                                              cf->userdata);
+            cf->begin_checkpoint_userdata(info->lsn_of_checkpoint_in_progress, cf->userdata);
         }
+        return 0;
     }
+};
+
+//
+// Update the user data in any cachefiles in our checkpoint list.
+//
+void checkpointer::update_cachefiles() {
+    struct iterate_begin_checkpoint iterate(m_lsn_of_checkpoint_in_progress);
+    int r = m_cf_list->m_active_fileid.iterate<struct iterate_begin_checkpoint,
+                                               iterate_begin_checkpoint::fn>(&iterate);
+    assert_zero(r);
 }
 
+struct iterate_note_pin {
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), void **UU(extra)) {
+        assert(cf->note_pin_by_checkpoint);
+        cf->note_pin_by_checkpoint(cf, cf->userdata);
+        cf->for_checkpoint = true;
+        return 0;
+    }
+};
+
 //
 // Sets up and kicks off a checkpoint.
 //
 void checkpointer::begin_checkpoint() {
     // 1. Initialize the accountability counters.
-    m_checkpoint_num_files = 0;
     m_checkpoint_num_txns = 0;
     
     // 2. Make list of cachefiles to be included in the checkpoint.
-    // TODO: <CER> How do we remove the non-lock cachetable reference here?
     m_cf_list->read_lock();
-    for (CACHEFILE cf = m_cf_list->m_active_head; cf; cf = cf->next) {
-        // The caller must serialize open, close, and begin checkpoint.
-        // So we should never see a closing cachefile here.
-        // <CER> Is there an assert we can add here?
-        
-        // Putting this check here so that this method may be called
-        // by cachetable tests.
-        assert(cf->note_pin_by_checkpoint);
-        cf->note_pin_by_checkpoint(cf, cf->userdata);
-        cf->for_checkpoint = true;
-        m_checkpoint_num_files++;
-    }
+    m_cf_list->m_active_fileid.iterate<void *, iterate_note_pin::fn>(nullptr);
+    m_checkpoint_num_files = m_cf_list->m_active_fileid.size();
     m_cf_list->read_unlock();
     
     // 3. Create log entries for this checkpoint.
@@ -4475,6 +4480,14 @@ void checkpointer::begin_checkpoint() {
     m_list->write_pending_exp_unlock();
 }
 
+struct iterate_log_fassociate {
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), void **UU(extra)) {
+        assert(cf->log_fassociate_during_checkpoint);
+        cf->log_fassociate_during_checkpoint(cf, cf->userdata);
+        return 0;
+    }
+};
+
 //
 // Assuming the logger exists, this will write out the folloing 
 // information to the log.
@@ -4498,10 +4511,7 @@ void checkpointer::log_begin_checkpoint() {
     m_lsn_of_checkpoint_in_progress = begin_lsn;
 
     // Log the list of open dictionaries.
-    for (CACHEFILE cf = m_cf_list->m_active_head; cf; cf = cf->next) {
-        assert(cf->log_fassociate_during_checkpoint);
-        cf->log_fassociate_during_checkpoint(cf, cf->userdata);
-    }
+    m_cf_list->m_active_fileid.iterate<void *, iterate_log_fassociate::fn>(nullptr);
 
     // Write open transactions to the log.
     r = toku_txn_manager_iter_over_live_txns(
@@ -4576,17 +4586,29 @@ void checkpointer::end_checkpoint(void (*testcallback_f)(void*),  void* testextr
     toku_free(checkpoint_cfs);
 }
 
-void checkpointer::fill_checkpoint_cfs(CACHEFILE* checkpoint_cfs) {
-    m_cf_list->read_lock();
-    uint32_t curr_index = 0;
-    for (CACHEFILE cf = m_cf_list->m_active_head; cf; cf = cf->next) {
+struct iterate_checkpoint_cfs {
+    CACHEFILE *checkpoint_cfs;
+    uint32_t checkpoint_num_files;
+    uint32_t curr_index;
+    iterate_checkpoint_cfs(CACHEFILE *cfs, uint32_t num_files) :
+        checkpoint_cfs(cfs), checkpoint_num_files(num_files), curr_index(0) {
+    }
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), struct iterate_checkpoint_cfs *info) {
         if (cf->for_checkpoint) {
-            assert(curr_index < m_checkpoint_num_files);
-            checkpoint_cfs[curr_index] = cf;
-            curr_index++;
+            assert(info->curr_index < info->checkpoint_num_files);
+            info->checkpoint_cfs[info->curr_index] = cf;
+            info->curr_index++;
         }
+        return 0;
     }
-    assert(curr_index == m_checkpoint_num_files);
+};
+
+void checkpointer::fill_checkpoint_cfs(CACHEFILE* checkpoint_cfs) {
+    struct iterate_checkpoint_cfs iterate(checkpoint_cfs, m_checkpoint_num_files);
+
+    m_cf_list->read_lock();
+    m_cf_list->m_active_fileid.iterate<struct iterate_checkpoint_cfs, iterate_checkpoint_cfs::fn>(&iterate);
+    assert(iterate.curr_index == m_checkpoint_num_files);
     m_cf_list->read_unlock();
 }
 
@@ -4671,19 +4693,18 @@ void checkpointer::remove_cachefiles(CACHEFILE* checkpoint_cfs) {
 static_assert(std::is_pod<cachefile_list>::value, "cachefile_list isn't POD");
 
 void cachefile_list::init() {
-    m_active_head = NULL;
-    m_stale_head = NULL;
-    m_stale_tail = NULL;
     m_next_filenum_to_use.fileid = 0;
     m_next_hash_id_to_use = 0;
     toku_pthread_rwlock_init(&m_lock, NULL);
     m_active_filenum.create();
     m_active_fileid.create();
+    m_stale_fileid.create();
 }
 
 void cachefile_list::destroy() {
     m_active_filenum.destroy();
     m_active_fileid.destroy();
+    m_stale_fileid.destroy();
     toku_pthread_rwlock_destroy(&m_lock);
 }
 
@@ -4702,34 +4723,31 @@ void cachefile_list::write_lock() {
 void cachefile_list::write_unlock() {
     toku_pthread_rwlock_wrunlock(&m_lock);
 }
-int cachefile_list::cachefile_of_iname_in_env(const char *iname_in_env, CACHEFILE *cf) {
-    read_lock();
-    CACHEFILE extant;
-    int r;
-    r = ENOENT;
-    for (extant = m_active_head; extant; extant = extant->next) {
-        if (extant->fname_in_env &&
-            !strcmp(extant->fname_in_env, iname_in_env)) {
-            *cf = extant;
-            r = 0;
-            break;
+
+struct iterate_find_iname {
+    const char *iname_in_env;
+    CACHEFILE found_cf;
+    iterate_find_iname(const char *iname) : iname_in_env(iname), found_cf(nullptr) { }
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), struct iterate_find_iname *info) {
+        if (cf->fname_in_env && strcmp(cf->fname_in_env, info->iname_in_env) == 0) {
+            info->found_cf = cf;
+            return -1;
         }
+        return 0;
     }
-    read_unlock();
-    return r;
-}
+};
+
+int cachefile_list::cachefile_of_iname_in_env(const char *iname_in_env, CACHEFILE *cf) {
+    struct iterate_find_iname iterate(iname_in_env);
 
-int cachefile_list::cachefile_of_filenum(FILENUM filenum, CACHEFILE *cf) {
     read_lock();
-    CACHEFILE extant;
-    int r = ENOENT;
-    *cf = NULL;
-    for (extant = m_active_head; extant; extant = extant->next) {
-        if (extant->filenum.fileid==filenum.fileid) {
-            *cf = extant;
-            r = 0;
-            break;
-        }
+    int r = m_active_fileid.iterate<iterate_find_iname, iterate_find_iname::fn>(&iterate);
+    if (iterate.found_cf != nullptr) {
+        assert(strcmp(iterate.found_cf->fname_in_env, iname_in_env) == 0);
+        *cf = iterate.found_cf;
+        r = 0;
+    } else {
+        r = ENOENT;
     }
     read_unlock();
     return r;
@@ -4746,20 +4764,23 @@ static int cachefile_find_by_filenum(const CACHEFILE &a_cf, const FILENUM &b) {
     }
 }
 
+int cachefile_list::cachefile_of_filenum(FILENUM filenum, CACHEFILE *cf) {
+    read_lock();
+    int r = m_active_filenum.find_zero<FILENUM, cachefile_find_by_filenum>(filenum, cf, nullptr);
+    if (r == DB_NOTFOUND) {
+        r = ENOENT;
+    } else {
+        invariant_zero(r);
+    }
+    read_unlock();
+    return r;
+}
+
 static int cachefile_find_by_fileid(const CACHEFILE &a_cf, const struct fileid &b) {
     return toku_fileid_cmp(a_cf->fileid, b);
 }
 
 void cachefile_list::add_cf_unlocked(CACHEFILE cf) {
-    invariant(cf->next == NULL);
-    invariant(cf->prev == NULL);
-    cf->next = m_active_head;
-    cf->prev = NULL;
-    if (m_active_head) {
-        m_active_head->prev = cf;
-    }
-    m_active_head = cf;
-
     int r;
     r = m_active_filenum.insert<FILENUM, cachefile_find_by_filenum>(cf, cf->filenum, nullptr);
     assert_zero(r);
@@ -4769,36 +4790,13 @@ void cachefile_list::add_cf_unlocked(CACHEFILE cf) {
 
 void cachefile_list::add_stale_cf(CACHEFILE cf) {
     write_lock();
-    invariant(cf->next == NULL);
-    invariant(cf->prev == NULL);
-
-    cf->next = m_stale_head;
-    cf->prev = NULL;
-    if (m_stale_head) {
-        m_stale_head->prev = cf;
-    }
-    m_stale_head = cf;
-    if (m_stale_tail == NULL) {
-        m_stale_tail = cf;
-    }
+    int r = m_stale_fileid.insert<struct fileid, cachefile_find_by_fileid>(cf, cf->fileid, nullptr);
+    assert_zero(r);
     write_unlock();
 }
 
 void cachefile_list::remove_cf(CACHEFILE cf) {
     write_lock();
-    invariant(m_active_head != NULL);
-    if (cf->next) {
-        cf->next->prev = cf->prev;
-    }
-    if (cf->prev) {
-        cf->prev->next = cf->next;
-    }
-    if (cf == m_active_head) {
-        invariant(cf->prev == NULL);
-        m_active_head = cf->next;
-    }
-    cf->prev = NULL;
-    cf->next = NULL;
 
     uint32_t idx;
     int r;
@@ -4816,24 +4814,12 @@ void cachefile_list::remove_cf(CACHEFILE cf) {
 }
 
 void cachefile_list::remove_stale_cf_unlocked(CACHEFILE cf) {
-    invariant(m_stale_head != NULL);
-    invariant(m_stale_tail != NULL);
-    if (cf->next) {
-        cf->next->prev = cf->prev;
-    }
-    if (cf->prev) {
-        cf->prev->next = cf->next;
-    }
-    if (cf == m_stale_head) {
-        invariant(cf->prev == NULL);
-        m_stale_head = cf->next;
-    }
-    if (cf == m_stale_tail) {
-        invariant(cf->next == NULL);
-        m_stale_tail = cf->prev;
-    }
-    cf->prev = NULL;
-    cf->next = NULL;
+    uint32_t idx;
+    int r;
+    r = m_stale_fileid.find_zero<struct fileid, cachefile_find_by_fileid>(cf->fileid, nullptr, &idx);
+    assert_zero(r);
+    r = m_stale_fileid.delete_at(idx);
+    assert_zero(r);
 }
 
 FILENUM cachefile_list::reserve_filenum() {
@@ -4849,11 +4835,6 @@ FILENUM cachefile_list::reserve_filenum() {
         break;
     }
     FILENUM filenum = m_next_filenum_to_use;
-#if TOKU_DEBUG_PARANOID
-    for (CACHEFILE extant = m_active_head; extant; extant = extant->next) {
-        assert(filenum.fileid != extant->filenum.fileid);
-    }
-#endif
     m_next_filenum_to_use.fileid++;
     write_unlock();
     return filenum;
@@ -4865,91 +4846,77 @@ uint32_t cachefile_list::get_new_hash_id_unlocked() {
     return retval;
 }
 
-CACHEFILE cachefile_list::find_cachefile_in_list_unlocked(
-    CACHEFILE start,
-    struct fileid* fileid
-    )
-{
-    CACHEFILE retval = NULL;
-    for (CACHEFILE extant = start; extant; extant = extant->next) {
-        if (toku_fileids_are_equal(&extant->fileid, fileid)) {
-            // Clients must serialize cachefile open, close, and unlink
-            // So, during open, we should never see a closing cachefile 
-            // or one that has been marked as unlink on close.
-            assert(!extant->unlink_on_close);
-            retval = extant;
-            goto exit;
-        }
-    }
-exit:
-    return retval;
-}
-
 CACHEFILE cachefile_list::find_cachefile_unlocked(struct fileid* fileid) {
     CACHEFILE cf = nullptr;
     int r = m_active_fileid.find_zero<struct fileid, cachefile_find_by_fileid>(*fileid, &cf, nullptr);
     if (r == 0) {
         assert(!cf->unlink_on_close);
     }
-#if TOKU_DEBUG_PARANOID
-    assert(cf == find_cachefile_in_list_unlocked(m_active_head, fileid));
-#endif
     return cf;
 }
 
 CACHEFILE cachefile_list::find_stale_cachefile_unlocked(struct fileid* fileid) {
-    return find_cachefile_in_list_unlocked(m_stale_head, fileid);
+    CACHEFILE cf = nullptr;
+    int r = m_stale_fileid.find_zero<struct fileid, cachefile_find_by_fileid>(*fileid, &cf, nullptr);
+    if (r == 0) {
+        assert(!cf->unlink_on_close);
+    }
+    return cf;
 }
 
 void cachefile_list::verify_unused_filenum(FILENUM filenum) {
     int r = m_active_filenum.find_zero<FILENUM, cachefile_find_by_filenum>(filenum, nullptr, nullptr);
     assert(r == DB_NOTFOUND);
-#if TOKU_DEBUG_PARANOID
-    for (CACHEFILE extant = m_active_head; extant; extant = extant->next) {
-        invariant(extant->filenum.fileid != filenum.fileid);
-    }
-#endif
 }
 
 // returns true if some eviction ran, false otherwise
 bool cachefile_list::evict_some_stale_pair(evictor* ev) {
-    PAIR p = NULL;
-    CACHEFILE cf_to_destroy = NULL;
     write_lock();
-    if (m_stale_tail == NULL) {
+    if (m_stale_fileid.size() == 0) {
         write_unlock();
         return false;
     }
-    p = m_stale_tail->cf_head;
+
+    CACHEFILE stale_cf = nullptr;
+    int r = m_stale_fileid.fetch(0, &stale_cf);
+    assert_zero(r);
+
     // we should not have a cf in the stale list
     // that does not have any pairs
+    PAIR p = stale_cf->cf_head;
     paranoid_invariant(p != NULL);
-
     evict_pair_from_cachefile(p);
 
     // now that we have evicted something,
     // let's check if the cachefile is needed anymore
-    if (m_stale_tail->cf_head == NULL) {
-        cf_to_destroy = m_stale_tail;
-        remove_stale_cf_unlocked(m_stale_tail);
+    //
+    // it is not needed if the latest eviction caused
+    // the cf_head for that cf to become null
+    bool destroy_cf = stale_cf->cf_head == nullptr;
+    if (destroy_cf) {
+        remove_stale_cf_unlocked(stale_cf);
     }
 
     write_unlock();
     
     ev->remove_pair_attr(p->attr);
     cachetable_free_pair(p);
-    if (cf_to_destroy) {
-        cachefile_destroy(cf_to_destroy);
+    if (destroy_cf) {
+        cachefile_destroy(stale_cf);
     }
     return true;
 }
 
 void cachefile_list::free_stale_data(evictor* ev) {
     write_lock();
-    while (m_stale_tail != NULL) {
-        PAIR p = m_stale_tail->cf_head;
+    while (m_stale_fileid.size() != 0) {
+        CACHEFILE stale_cf = nullptr;
+        int r = m_stale_fileid.fetch(0, &stale_cf); 
+        assert_zero(r);
+
         // we should not have a cf in the stale list
         // that does not have any pairs
+        PAIR p = stale_cf->cf_head;
         paranoid_invariant(p != NULL);
         
         evict_pair_from_cachefile(p);
@@ -4958,10 +4925,9 @@ void cachefile_list::free_stale_data(evictor* ev) {
         
         // now that we have evicted something,
         // let's check if the cachefile is needed anymore
-        if (m_stale_tail->cf_head == NULL) {
-            CACHEFILE cf_to_destroy = m_stale_tail;
-            remove_stale_cf_unlocked(m_stale_tail);
-            cachefile_destroy(cf_to_destroy);
+        if (stale_cf->cf_head == NULL) {
+            remove_stale_cf_unlocked(stale_cf);
+            cachefile_destroy(stale_cf);
         }
     }
     write_unlock();
diff --git a/ft/tests/cachetable-checkpointer-class.cc b/ft/tests/cachetable-checkpointer-class.cc
index c2adc202fb5..fa950d3972a 100644
--- a/ft/tests/cachetable-checkpointer-class.cc
+++ b/ft/tests/cachetable-checkpointer-class.cc
@@ -112,6 +112,14 @@ struct checkpointer_test {
     uint32_t k);
 };
 
+static void init_cachefile(CACHEFILE cf, int which_cf, bool for_checkpoint) {
+    memset(cf, 0, sizeof(*cf));
+    create_dummy_functions(cf);
+    cf->fileid = { 0, (unsigned) which_cf };
+    cf->filenum = { (unsigned) which_cf };
+    cf->for_checkpoint = for_checkpoint;
+}
+
 //------------------------------------------------------------------------------
 // test_begin_checkpoint() -
 //
@@ -135,33 +143,28 @@ void checkpointer_test::test_begin_checkpoint() {
     // 2. Call checkpoint with ONE cachefile.
     //cachefile cf;
     struct cachefile cf;
-    cf.next = NULL;
-    cf.for_checkpoint = false;
-    m_cp.m_cf_list->m_active_head = &cf;
-    create_dummy_functions(&cf);
+    init_cachefile(&cf, 0, false);
+    m_cp.m_cf_list->add_cf_unlocked(&cf);
 
     m_cp.begin_checkpoint();
     assert(m_cp.m_checkpoint_num_files == 1);
     assert(cf.for_checkpoint == true);
+    m_cp.m_cf_list->remove_cf(&cf);
 
     // 3. Call checkpoint with MANY cachefiles.
     const uint32_t count = 3;
     struct cachefile cfs[count];
-    m_cp.m_cf_list->m_active_head = &cfs[0];
     for (uint32_t i = 0; i < count; ++i) {
-        cfs[i].for_checkpoint = false;
+        init_cachefile(&cfs[i], i, false);
         create_dummy_functions(&cfs[i]);
-        if (i == count - 1) {
-            cfs[i].next = NULL;
-        } else {
-            cfs[i].next = &cfs[i + 1];
-        }
+        m_cp.m_cf_list->add_cf_unlocked(&cfs[i]);
     }
 
     m_cp.begin_checkpoint();
     assert(m_cp.m_checkpoint_num_files == count);
     for (uint32_t i = 0; i < count; ++i) {
         assert(cfs[i].for_checkpoint == true);
+        cfl.remove_cf(&cfs[i]);
     }
     ctbl.list.destroy();
     m_cp.destroy();
@@ -195,10 +198,8 @@ void checkpointer_test::test_pending_bits() {
     //
     struct cachefile cf;
     cf.cachetable = &ctbl;
-    memset(&cf, 0, sizeof(cf));
-    cf.next = NULL;
-    cf.for_checkpoint = true;
-    m_cp.m_cf_list->m_active_head = &cf;
+    init_cachefile(&cf, 0, true);
+    m_cp.m_cf_list->add_cf_unlocked(&cf);
     create_dummy_functions(&cf);
 
     CACHEKEY k;
@@ -258,6 +259,7 @@ void checkpointer_test::test_pending_bits() {
 
     ctbl.list.destroy();
     m_cp.destroy();
+    cfl.remove_cf(&cf);
     cfl.destroy();
 }
 
@@ -337,14 +339,11 @@ void checkpointer_test::test_end_checkpoint() {
     cfl.init();
 
     struct cachefile cf;
-    memset(&cf, 0, sizeof(cf));
-    cf.next = NULL;
-    cf.for_checkpoint = true;
-    create_dummy_functions(&cf);
+    init_cachefile(&cf, 0, true);
 
     ZERO_STRUCT(m_cp);
     m_cp.init(&ctbl.list, NULL, &ctbl.ev, &cfl);
-    m_cp.m_cf_list->m_active_head = &cf;
+    m_cp.m_cf_list->add_cf_unlocked(&cf);
 
     // 2. Add data before running checkpoint.
     const uint32_t count = 6;
@@ -394,6 +393,7 @@ void checkpointer_test::test_end_checkpoint() {
         assert(pp);
         m_cp.m_list->evict_completely(pp);
     }
+    cfl.remove_cf(&cf);
     m_cp.destroy();
     ctbl.list.destroy();
     cfl.destroy();