[TDB-20] With checkpoint disabled, closing FT file still rotates header.

This is an attempt based on @BohuTANG's patch
percona#331

in particular, the ctest is all his credits. I included his commit
history in the git log,

1) The issue is real.
2) @bohu's patch works for mediating the issue but somewhat introduces
a ft leak that the ft with ref=0 may be left until next open/close or
checkpoint.
3) If ft_close should be held off upon a backup, then it simply means
a backup should hold a ref to the ft and responsible for closing ft up
if it is the last one to hold it, like any other possible referencer.
including the checkpoint, the txn and ft handlers.
4) mtr test contributed by @BohuTANG still passes

Author: Jun

ft/cachetable/cachetable-internal.h

@@ -161,6 +161,9 @@ struct cachefile {
     void (*end_checkpoint_userdata)(CACHEFILE cf, int fd, void *userdata); // after checkpointing cachefiles call this function.
     void (*note_pin_by_checkpoint)(CACHEFILE cf, void *userdata); // add a reference to the userdata to prevent it from being removed from memory
     void (*note_unpin_by_checkpoint)(CACHEFILE cf, void *userdata); // add a reference to the userdata to prevent it from being removed from memory
+
+    void (*note_pin_by_backup)(CACHEFILE cf, void *userdata); // add a reference to the userdata to prevent it from being removed from memory
+    void (*note_unpin_by_backup)(CACHEFILE cf, void *userdata); // add a reference to the userdata to prevent it from being removed from memory
     BACKGROUND_JOB_MANAGER bjm;
 };
 
@@ -604,7 +607,7 @@ struct cachetable {
     KIBBUTZ client_kibbutz; // pool of worker threads and jobs to do asynchronously for the client.
     KIBBUTZ ct_kibbutz; // pool of worker threads and jobs to do asynchronously for the cachetable
     KIBBUTZ checkpointing_kibbutz; // small pool for checkpointing cloned pairs
-    bool in_backup; // we are in back up or NOT, default is false
+    //bool in_backup; // we are in back up or NOT, default is false
 
     char *env_dir;
 };

ft/cachetable/cachetable.cc

@@ -269,7 +269,7 @@ int toku_cachetable_create_ex(CACHETABLE *ct_result, long size_limit,
     CACHETABLE XCALLOC(ct);
     ct->list.init();
     ct->cf_list.init();
-    ct->in_backup = false;
+    //ct->in_backup = false;
 
     int num_processors = toku_os_get_number_active_processors();
     int checkpointing_nworkers = (num_processors/4) ? num_processors/4 : 1;
@@ -2784,24 +2784,40 @@ void toku_cachetable_end_checkpoint(CHECKPOINTER cp, TOKULOGGER UU(logger),
 }
 
 // in_backup begin
+
+struct iterate_note_pin_backup {
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), void **UU(extra)) {
+        assert(cf->note_pin_by_backup);
+        cf->note_pin_by_backup(cf, cf->userdata);
+        return 0;
+    }
+};
+
+struct iterate_note_unpin_backup {
+    static int fn(const CACHEFILE &cf, uint32_t UU(idx), void **UU(extra)) {
+        assert(cf->note_unpin_by_backup);
+        cf->note_unpin_by_backup(cf, cf->userdata);
+        return 0;
+    }
+};
 void toku_cachetable_begin_backup(CACHETABLE ct)
 {
     ct->cf_list.read_lock();
-    ct->in_backup = true;
+    ct->cf_list.m_active_fileid.iterate<void *, iterate_note_pin_backup::fn>(nullptr); 
     ct->cf_list.read_unlock();
 }
 
 void toku_cachetable_end_backup(CACHETABLE ct)
 {
-    ct->cf_list.read_lock();
-    ct->in_backup = false;
-    ct->cf_list.read_unlock();
+    ct->cf_list.m_active_fileid.iterate<void *, iterate_note_unpin_backup::fn>(nullptr); 
 }
 
+#if 0
 bool toku_cachefile_in_backup(CACHEFILE cf)
 {
     return cf->cachetable->in_backup;
 }
+#endif
 // in_backup end
 
 TOKULOGGER toku_cachefile_logger (CACHEFILE cf) {
@@ -2920,7 +2936,9 @@ toku_cachefile_set_userdata (CACHEFILE cf,
                              void (*begin_checkpoint_userdata)(LSN, void*),
                              void (*end_checkpoint_userdata)(CACHEFILE, int, void*),
                              void (*note_pin_by_checkpoint)(CACHEFILE, void*),
-                             void (*note_unpin_by_checkpoint)(CACHEFILE, void*)) {
+                             void (*note_unpin_by_checkpoint)(CACHEFILE, void*),
+                             void (*note_pin_by_backup)(CACHEFILE, void*),
+                             void (*note_unpin_by_backup)(CACHEFILE, void*)) {
     cf->userdata = userdata;
     cf->log_fassociate_during_checkpoint = log_fassociate_during_checkpoint;
     cf->close_userdata = close_userdata;
@@ -2930,6 +2948,8 @@ toku_cachefile_set_userdata (CACHEFILE cf,
     cf->end_checkpoint_userdata = end_checkpoint_userdata;
     cf->note_pin_by_checkpoint = note_pin_by_checkpoint;
     cf->note_unpin_by_checkpoint = note_unpin_by_checkpoint;
+    cf->note_pin_by_backup = note_pin_by_backup;
+    cf->note_unpin_by_backup = note_unpin_by_backup;
 }
 
 void *toku_cachefile_get_userdata(CACHEFILE cf) {

ft/cachetable/cachetable.h

@@ -289,7 +289,9 @@ void toku_cachefile_set_userdata(CACHEFILE cf, void *userdata,
     void (*begin_checkpoint_userdata)(LSN, void*),
     void (*end_checkpoint_userdata)(CACHEFILE, int, void*),
     void (*note_pin_by_checkpoint)(CACHEFILE, void*),
-    void (*note_unpin_by_checkpoint)(CACHEFILE, void*));
+    void (*note_unpin_by_checkpoint)(CACHEFILE, void*),
+    void (*note_pin_by_backup)(CACHEFILE, void*),
+    void (*note_unpin_by_backup)(CACHEFILE, void*));
 // Effect: Store some cachefile-specific user data.  When the last reference to a cachefile is closed, we call close_userdata().
 // Before starting a checkpoint, we call checkpoint_prepare_userdata().
 // When the cachefile needs to be checkpointed, we call checkpoint_userdata().

ft/ft-internal.h

@@ -197,7 +197,9 @@ struct ft {
     uint32_t num_txns;
     // A checkpoint is running.  If true, then keep this header around for checkpoint, like a transaction
     bool pinned_by_checkpoint;
-
+    // Number of backups that are using this FT. If it is nonzero, keep this header around until backup
+    // is completd.
+    uint32_t num_backups;
     // is this ft a blackhole? if so, all messages are dropped.
     bool blackhole;
 

ft/ft.cc

@@ -293,13 +293,7 @@ static void ft_note_pin_by_checkpoint (CACHEFILE UU(cachefile), void *header_v)
     FT ft = (FT) header_v;
     toku_ft_grab_reflock(ft);
     assert(!ft->pinned_by_checkpoint);
-
-    // because we add in_backup for cachetable, with this case:
-    // a cachefile in_backup is true and then do a checkpoint
-    // this assertion fails since nobody references to the ft
-#if 0
     assert(toku_ft_needed_unlocked(ft));
-#endif
     ft->pinned_by_checkpoint = true;
     toku_ft_release_reflock(ft);
 }
@@ -319,6 +313,35 @@ static void ft_note_unpin_by_checkpoint (CACHEFILE UU(cachefile), void *header_v
     toku_ft_remove_reference(ft, false, ZERO_LSN, unpin_by_checkpoint_callback, NULL);
 }
 
+
+// maps to cf->note_pin_by_backup
+//Must be protected by ydb lock.
+//Is only called by backup begin, which holds it
+static void ft_note_pin_by_backup (CACHEFILE UU(cachefile), void *header_v) {
+    // Note: open_close lock is held by checkpoint begin
+    FT ft = (FT) header_v;
+    toku_ft_grab_reflock(ft);
+    assert(toku_ft_needed_unlocked(ft));
+    ft->num_backups ++;
+    toku_ft_release_reflock(ft);
+}
+
+// Requires: the reflock is held.
+static void unpin_by_backup_callback(FT ft, void *extra) {
+    invariant(extra == NULL);
+    invariant(ft->num_backups>0);
+    ft->num_backups --;
+}
+
+// maps to cf->note_unpin_by_backup
+//Must be protected by ydb lock.
+//Called by end_backup, which grabs ydb lock around note_unpin
+static void ft_note_unpin_by_backup (CACHEFILE UU(cachefile), void *header_v) {
+    FT ft = (FT) header_v;
+    toku_ft_remove_reference(ft, false, ZERO_LSN, unpin_by_backup_callback, NULL);
+}
+
+
 //
 // End of Functions that are callbacks to the cachefile
 /////////////////////////////////////////////////////////////////////////
@@ -364,7 +387,9 @@ static void ft_init(FT ft, FT_OPTIONS options, CACHEFILE cf) {
                                 ft_begin_checkpoint,
                                 ft_end_checkpoint,
                                 ft_note_pin_by_checkpoint,
-                                ft_note_unpin_by_checkpoint);
+                                ft_note_unpin_by_checkpoint,
+				ft_note_pin_by_backup,
+				ft_note_unpin_by_backup);
 
     ft->blocktable.verify_no_free_blocknums();
 }
@@ -461,7 +486,9 @@ int toku_read_ft_and_store_in_cachefile (FT_HANDLE ft_handle, CACHEFILE cf, LSN
                                 ft_begin_checkpoint,
                                 ft_end_checkpoint,
                                 ft_note_pin_by_checkpoint,
-                                ft_note_unpin_by_checkpoint);
+                                ft_note_unpin_by_checkpoint,
+				ft_note_pin_by_backup,
+				ft_note_unpin_by_backup);
     *header = ft;
     return 0;
 }
@@ -481,7 +508,7 @@ static int
 ft_get_reference_count(FT ft) {
     uint32_t pinned_by_checkpoint = ft->pinned_by_checkpoint ? 1 : 0;
     int num_handles = toku_list_num_elements_est(&ft->live_ft_handles);
-    return pinned_by_checkpoint + ft->num_txns + num_handles;
+    return pinned_by_checkpoint + ft->num_txns + ft-> num_backups + num_handles;
 }
 
 // a ft is needed in memory iff its reference count is non-zero
@@ -949,12 +976,8 @@ void toku_ft_remove_reference(
             assert(!needed);
         }
         if (!needed) {
-            // close header if we are not in backup
-            // if close header during backup, the FT files would be inconsistency
-            bool in_backup = toku_cachefile_in_backup(ft->cf);
-            if (!in_backup) {
-                toku_ft_evict_from_memory(ft, oplsn_valid, oplsn);
-            }
+            // close header
+            toku_ft_evict_from_memory(ft, oplsn_valid, oplsn);
         }
 
         toku_ft_open_close_unlock();

ft/tests/cachetable-pin-checkpoint.cc

@@ -376,7 +376,9 @@ cachetable_test (void) {
         &test_begin_checkpoint,
         &dummy_end,
         &dummy_note_pin,
-        &dummy_note_unpin
+        &dummy_note_unpin,
+	&dummy_note_pin,
+	&dummy_note_unpin
         );
 
     toku_pthread_t time_tid;

ft/tests/cachetable-put-checkpoint.cc

@@ -508,7 +508,9 @@ cachetable_test (void) {
         test_begin_checkpoint, // called in begin_checkpoint
         &dummy_end,
         &dummy_note_pin,
-        &dummy_note_unpin
+        &dummy_note_unpin,
+	&dummy_note_pin,
+	&dummy_note_unpin
         );
 
     toku_pthread_t time_tid;

ft/tests/cachetable-simple-close.cc

@@ -60,7 +60,9 @@ static void set_cf_userdata(CACHEFILE f1) {
         &dummy_begin,
         &dummy_end,
         &dummy_note_pin,
-        &dummy_note_unpin
+        &dummy_note_unpin,
+	&dummy_note_pin,
+	&dummy_note_unpin
         );
 }
 

ft/tests/cachetable-test.h

@@ -68,5 +68,7 @@ create_dummy_functions(CACHEFILE cf)
                                &dummy_begin,
                                &dummy_end,
                                &dummy_note_pin,
+                               &dummy_note_unpin,
+			       &dummy_note_pin,
                                &dummy_note_unpin);
 };

ft/tests/ft-in-backup-test.cc

@@ -50,8 +50,6 @@ static void test_in_backup() {
     unlink(fname);
 
     toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);
-    CHECKPOINTER cp = toku_cachetable_get_checkpointer(ct);
-
     // TEST1 : for normal
     r = toku_open_ft_handle(fname, 1, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, null_txn, toku_builtin_compare_fun);
     assert_zero(r);
@@ -76,7 +74,6 @@ static void test_in_backup() {
 
     // TEST2: in fly without checkpoint test
     toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);
-    cp = toku_cachetable_get_checkpointer(ct);
     r = toku_open_ft_handle(fname, 0, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, null_txn, toku_builtin_compare_fun);
     assert_zero(r);
 
@@ -97,7 +94,6 @@ static void test_in_backup() {
     // here has a leak for this ft evicts from memroy, but that makes sense
     r = toku_close_ft_handle_nolsn(ft, 0);
     assert_zero(r);
-    toku_cachetable_end_backup(ct);
     toku_cachetable_close(&ct);
 
     // check the in fly key/value, it shouldn't exist
@@ -107,11 +103,11 @@ static void test_in_backup() {
     ft_lookup_and_fail_nodup(ft, (char*)"halou");
     r = toku_close_ft_handle_nolsn(ft, 0);
     assert_zero(r);
+    toku_cachetable_end_backup(ct);
     toku_cachetable_close(&ct);
 
     // TEST3: in fly with checkpoint test
-    toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);
-    cp = toku_cachetable_get_checkpointer(ct);
+    toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);  
     r = toku_open_ft_handle(fname, 0, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, null_txn, toku_builtin_compare_fun);
     assert_zero(r);
 
@@ -132,11 +128,8 @@ static void test_in_backup() {
     r = toku_close_ft_handle_nolsn(ft, 0);
     assert_zero(r);
     toku_cachetable_end_backup(ct);
-    r = toku_checkpoint(cp, NULL, NULL, NULL, NULL, NULL, CLIENT_CHECKPOINT);
-    assert_zero(r);
     toku_cachetable_close(&ct);
 
-    // check after checkpoint
     toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);
     r = toku_open_ft_handle(fname, 0, &ft, 1<<12, 1<<9, TOKU_DEFAULT_COMPRESSION_METHOD, ct, null_txn, toku_builtin_compare_fun);
     assert_zero(r);