[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.

        Summary:
        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;
 };
 
@@ -413,6 +416,8 @@ class checkpointer {
     void add_background_job();
     void remove_background_job();
     void end_checkpoint(void (*testcallback_f)(void*),  void* testextra);
+    void begin_backup();
+    void end_backup();
     TOKULOGGER get_logger();
     // used during begin_checkpoint
     void increment_num_txns();
@@ -602,6 +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
 
     char *env_dir;
 };

ft/cachetable/cachetable.cc

@@ -2782,6 +2782,37 @@ void toku_cachetable_end_checkpoint(CHECKPOINTER cp, TOKULOGGER UU(logger),
     cp->end_checkpoint(testcallback_f, testextra);
 }
 
+// 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->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.m_active_fileid.iterate<void *, iterate_note_unpin_backup::fn>(nullptr); 
+}
+
+// in_backup end
+
 TOKULOGGER toku_cachefile_logger (CACHEFILE cf) {
     return cf->cachetable->cp.get_logger();
 }
@@ -2898,7 +2929,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;
@@ -2908,6 +2941,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

@@ -147,6 +147,8 @@ void toku_cachetable_begin_checkpoint (CHECKPOINTER cp, struct tokulogger *logge
 void toku_cachetable_end_checkpoint(CHECKPOINTER cp, struct tokulogger *logger, 
                                    void (*testcallback_f)(void*),  void * testextra);
 
+void toku_cachetable_begin_backup(CACHETABLE ct);
+void toku_cachetable_end_backup(CACHETABLE ct);
 
 // Shuts down checkpoint thread
 // Requires no locks be held that are taken by the checkpoint function
@@ -285,7 +287,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

@@ -313,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
 /////////////////////////////////////////////////////////////////////////
@@ -358,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();
 }
@@ -455,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;
 }
@@ -475,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

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

@@ -0,0 +1,148 @@
+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+/*======
+This file is part of PerconaFT.
+
+
+Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved.
+
+    PerconaFT is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License, version 2,
+    as published by the Free Software Foundation.
+
+    PerconaFT is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+
+----------------------------------------
+
+    PerconaFT is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License, version 3,
+    as published by the Free Software Foundation.
+
+    PerconaFT is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with PerconaFT.  If not, see <http://www.gnu.org/licenses/>.
+======= */
+
+#ident "Copyright (c) 2006, 2015, Percona and/or its affiliates. All rights reserved."
+
+#include "test.h"
+#include "cachetable/checkpoint.h"
+
+static TOKUTXN const null_txn = 0;
+static const char *fname = TOKU_TEST_FILENAME;
+
+/* test for_backup in ft_close */
+static void test_in_backup() {
+    int r;
+    CACHETABLE ct;
+    FT_HANDLE ft;
+    unlink(fname);
+
+    toku_cachetable_create(&ct, 0, ZERO_LSN, nullptr);
+    // 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);
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+
+    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);
+    {
+        DBT k,v;
+        toku_ft_insert(ft, toku_fill_dbt(&k, "hello", 6), toku_fill_dbt(&v, "there", 6), null_txn);
+    }
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+
+    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);
+    ft_lookup_and_check_nodup(ft, "hello", "there");
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+    toku_cachetable_close(&ct);
+
+    // TEST2: in fly without checkpoint test
+    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);
+
+    toku_cachetable_begin_backup(ct);
+    // this key/value just in fly since we are in backing up
+    {
+        DBT k,v;
+        toku_ft_insert(ft, toku_fill_dbt(&k, "halou", 6), toku_fill_dbt(&v, "not there", 10), null_txn);
+    }
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+
+    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);
+    ft_lookup_and_check_nodup(ft, "halou", "not there");
+
+    // because we are in backup, so the FT header is stale after cachefile&cachetable closed
+    // 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_close(&ct);
+
+    // check the in fly key/value, it shouldn't exist
+    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);
+    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);  
+    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);
+
+    toku_cachetable_begin_backup(ct);
+    // this key/value just in fly since we are in backup
+    {
+        DBT k,v;
+        toku_ft_insert(ft, toku_fill_dbt(&k, "halou1", 7), toku_fill_dbt(&v, "not there", 10), null_txn);
+    }
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+
+    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);
+    ft_lookup_and_check_nodup(ft, "halou1", "not there");
+
+    // because we are in backup, so the FT header is stale after cachefile&cachetable closed
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+    toku_cachetable_end_backup(ct);
+    toku_cachetable_close(&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);
+    ft_lookup_and_check_nodup(ft, "halou1", "not there");
+    r = toku_close_ft_handle_nolsn(ft, 0);
+    assert_zero(r);
+    toku_cachetable_close(&ct);
+}
+
+int
+test_main (int argc , const char *argv[]) {
+    default_parse_args(argc, argv);
+    test_in_backup();
+    if (verbose) printf("test ok\n");
+    return 0;
+}