Reduce physrep-sql against the metadb

Signed-off-by: Mark Hannum <mhannum@bloomberg.net>

Author: markhannum

bdb/file.c

@@ -3663,8 +3663,6 @@ static void delete_log_files_int(bdb_state_type *bdb_state)
         }
     }
 
-    physrep_update_low_file_num(&lowfilenum, &local_lowfilenum);
-
     /* debug: print filenums from other nodes */
 
     /* if we have a maximum filenum defined in bdb attributes which is lower,

db/db_tunables.c

@@ -517,11 +517,13 @@ extern int gbl_physrep_hung_replicant_threshold;
 extern int gbl_physrep_revconn_check_interval;
 extern int gbl_physrep_fake_revconn_populate_error;
 extern int gbl_physrep_fake_revconn_populate_error_once;
+extern int gbl_physrep_fake_check_revconn_error;
 extern int gbl_debug_fake_rte_failure;
 extern int gbl_physrep_update_registry_interval;
 extern int gbl_physrep_i_am_metadb;
 extern int gbl_physrep_keepalive_v2;
 extern int gbl_physrep_keepalive_freq_sec;
+extern int gbl_physrep_slow_replicant_check_freq_sec;
 extern int gbl_physrep_max_candidates;
 extern int gbl_physrep_reconnect_penalty;
 extern int gbl_physrep_reconnect_interval;

db/db_tunables.h

@@ -1821,19 +1821,15 @@ REGISTER_TUNABLE("tranlog_incoherent_timeout", "Timeout in seconds for incoheren
                  TUNABLE_INTEGER, &gbl_tranlog_incoherent_timeout, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("tranlog_maxpoll", "Tranlog timeout in seconds for blocking poll. (Default: 60)", TUNABLE_INTEGER,
                  &gbl_tranlog_maxpoll, 0, NULL, NULL, NULL, NULL);
-REGISTER_TUNABLE("physrep_check_minlog_freq_sec", "Check the minimum log number to keep this often. (Default: 10)",
-                 TUNABLE_INTEGER, &gbl_physrep_check_minlog_freq_sec, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_debug", "Print extended physrep trace. (Default: off)", TUNABLE_BOOLEAN, &gbl_physrep_debug,
                  0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_exit_on_invalid_logstream", "Exit physreps on invalid logstream.  (Default: off)",
                  TUNABLE_BOOLEAN, &gbl_physrep_exit_on_invalid_logstream, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_fanout",
                  "Maximum number of physical replicants that a node can service (Default: 8)",
                  TUNABLE_INTEGER, &gbl_physrep_fanout, 0, NULL, NULL, NULL, NULL);
-REGISTER_TUNABLE("physrep_hung_replicant_check_freq_sec",
-                 "Check for hung physical replicant this often. (Default: 10)",
-                 TUNABLE_INTEGER, &gbl_physrep_hung_replicant_check_freq_sec, 0, NULL,
-                 NULL, NULL, NULL);
+REGISTER_TUNABLE("physrep_hung_replicant_check_freq_sec", "Check for hung physical replicant this often. (Default: 60)",
+                 TUNABLE_INTEGER, &gbl_physrep_hung_replicant_check_freq_sec, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_hung_replicant_threshold",
                  "Report if the physical replicant has been inactive for this duration. (Default: 60)",
                  TUNABLE_INTEGER, &gbl_physrep_hung_replicant_threshold, 0, NULL,
@@ -1849,20 +1845,24 @@ REGISTER_TUNABLE("physrep_fake_revconn_populate_error_once",
                  "Fake physrep error on first attempt only, allowing retry to succeed.  (Default: off)",
                  TUNABLE_BOOLEAN, &gbl_physrep_fake_revconn_populate_error_once, EXPERIMENTAL | INTERNAL, NULL, NULL,
                  NULL, NULL);
+REGISTER_TUNABLE("physrep_fake_check_revconn_error",
+                 "Fake physrep error when asking if we should wait for a revconn.  (Default: off)", TUNABLE_BOOLEAN,
+                 &gbl_physrep_fake_check_revconn_error, EXPERIMENTAL | INTERNAL, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("debug_fake_rte_failure", "Fake rte failures in connect-remote-db.  (Default: off)", TUNABLE_BOOLEAN,
                  &gbl_debug_fake_rte_failure, EXPERIMENTAL | INTERNAL, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_i_am_metadb", "I am physical replication metadb (Default: off)",
                  TUNABLE_BOOLEAN, &gbl_physrep_i_am_metadb, NOARG, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_keepalive_v2", "Use version 2 of keepalive which includes first lsn. (Default: off)",
                  TUNABLE_BOOLEAN, &gbl_physrep_keepalive_v2, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_keepalive_freq_sec",
-                 "Periodically send lsn to source node after this interval. (Default: 10)", TUNABLE_INTEGER,
+                 "Periodically send lsn to source node after this interval. (Default: 60)", TUNABLE_INTEGER,
                  &gbl_physrep_keepalive_freq_sec, 0, NULL, NULL, NULL, NULL);
+REGISTER_TUNABLE("physrep_slow_replicant_check_freq_sec", "Check for slow physical replicant this often. (Default: 60)",
+                 TUNABLE_INTEGER, &gbl_physrep_slow_replicant_check_freq_sec, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_max_candidates",
                  "Maximum number of candidates that should be returned to a "
                  "new physical replicant during registration. (Default: 6)",
-                 TUNABLE_INTEGER, &gbl_physrep_max_candidates, 0, NULL,
-                 NULL, NULL, NULL);
+                 TUNABLE_INTEGER, &gbl_physrep_max_candidates, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_metadb_host", "List of physical replication metadb cluster hosts.", TUNABLE_STRING,
                  &gbl_physrep_metadb_host, READONLY, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("physrep_metadb_name", "Physical replication metadb cluster name.",
@@ -1922,10 +1922,10 @@ REGISTER_TUNABLE("revsql_debug",
                  "Print extended reversql-sql trace. (Default: off)",
                  TUNABLE_BOOLEAN, &gbl_revsql_debug, EXPERIMENTAL | INTERNAL,
                  NULL, NULL, NULL, NULL);
-REGISTER_TUNABLE("revsql_host_refresh_freq_sec", "The frequency at which the "
-                 "reverse connection host list will be refreshed (Default: 5secs)",
-                 TUNABLE_INTEGER, &gbl_revsql_host_refresh_freq_sec, EXPERIMENTAL | INTERNAL,
-                 NULL, NULL, NULL, NULL);
+REGISTER_TUNABLE("revsql_host_refresh_freq_sec",
+                 "The frequency at which the "
+                 "reverse connection host list will be refreshed (Default: 60 secs)",
+                 TUNABLE_INTEGER, &gbl_revsql_host_refresh_freq_sec, EXPERIMENTAL | INTERNAL, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("admin_revsql", "Run revsql sessions as admin.  (Default: Off)", TUNABLE_BOOLEAN, &gbl_admin_revsql, 0,
                  NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("revconn_rdtimeout", "Initial ms timeout for revconn connections.  (Default: 100)", TUNABLE_INTEGER,

db/phys_rep.c

@@ -58,18 +58,17 @@ typedef struct DB_Connection {
     } while (0)
 
 int gbl_physrep_debug = 0;
+int gbl_physrep_fake_check_revconn_error = 0;
 int gbl_physrep_reconnect_interval = 3600; // force re-registration every hour
 int gbl_physrep_reconnect_penalty = 0;
 int gbl_blocking_physrep = 1;
 int gbl_physrep_fanout = 8;
 int gbl_physrep_max_candidates = 6;
 int gbl_physrep_max_pending_replicants = 10;
 int gbl_deferred_phys_flag = 0;
-int gbl_physrep_source_nodes_refresh_freq_sec = 10;
-int gbl_physrep_slow_replicant_check_freq_sec = 10;
-int gbl_physrep_keepalive_freq_sec = 10;
-int gbl_physrep_check_minlog_freq_sec = 10;
-int gbl_physrep_hung_replicant_check_freq_sec = 10;
+int gbl_physrep_slow_replicant_check_freq_sec = 60;
+int gbl_physrep_keepalive_freq_sec = 60;
+int gbl_physrep_hung_replicant_check_freq_sec = 60;
 int gbl_physrep_hung_replicant_threshold = 60;
 int gbl_physrep_revconn_check_interval = 60;
 int gbl_physrep_update_registry_interval = 60;
@@ -78,7 +77,6 @@ int gbl_physrep_i_am_metadb = 0;
 int gbl_physrep_filter_by_class = 1;
 int gbl_started_physrep_threads = 0;
 
-unsigned int physrep_min_logfile;
 unsigned int gbl_deferred_phys_update;
 
 char *gbl_physrep_source_dbname;
@@ -1173,17 +1171,18 @@ static int send_keepalive(void)
     return 0;
 }
 
-unsigned int physrep_min_filenum() {
-    return physrep_min_logfile;
-}
-
 extern int gbl_reverse_hosts_v2;
 
 static int check_for_reverse_conn(cdb2_hndl_tp *hndl) {
     int rc;
     char cmd[400];
     int do_wait = 0;
 
+    if (gbl_physrep_fake_check_revconn_error) {
+        logmsg(LOGMSG_USER, "%s:%d: Fake error checking for reverse connection\n", __func__, __LINE__);
+        return -1;
+    }
+
     if (!gbl_reverse_hosts_v2) {
         rc = snprintf(cmd, sizeof(cmd), "exec procedure sys.physrep.should_wait_for_con('%s', '%s')", gbl_dbname,
                       (gbl_machine_class) ? gbl_machine_class : gbl_myhostname);
@@ -1215,38 +1214,6 @@ static int check_for_reverse_conn(cdb2_hndl_tp *hndl) {
     return (rc == 0) ? do_wait : -1;
 }
 
-void physrep_update_low_file_num(int *lowfilenum, int *local_lowfilenum) {
-    unsigned int physrep_minfilenum;
-    if ((get_dbtable_by_name("comdb2_physreps")) == NULL) {
-        return;
-    }
-
-    physrep_minfilenum = physrep_min_filenum();
-    if (physrep_minfilenum <= 0) {
-        if (gbl_physrep_debug) {
-            physrep_logmsg(LOGMSG_USER, "%s:%d: lowfilenum unchanged (physrep_minfilenum: %d)\n",
-                           __func__, __LINE__, physrep_minfilenum);
-        }
-    } else {
-        if (physrep_minfilenum <= *lowfilenum) {
-            if (gbl_physrep_debug) {
-                physrep_logmsg(LOGMSG_USER, "%s:%d: lowfilenum %d being changed "
-                               "physical replicant(s) (physrep_minfilenum: %d)\n",
-                               __func__, __LINE__, *lowfilenum, physrep_minfilenum);
-            }
-            *lowfilenum = physrep_minfilenum - 1;
-        }
-        if (physrep_minfilenum <= *local_lowfilenum) {
-            *local_lowfilenum = physrep_minfilenum - 1;
-        }
-    }
-
-    if (gbl_physrep_debug) {
-        physrep_logmsg(LOGMSG_USER, "%s:%d: lowfilenum: %d (physrep_minfilenum: %d)\n",
-                       __func__, __LINE__, *lowfilenum, physrep_minfilenum);
-    }
-}
-
 static int slow_replicants_count_int(cdb2_hndl_tp *metadb, unsigned int *count)
 {
     char query[400];
@@ -1306,93 +1273,6 @@ static int slow_replicants_count(unsigned int *count)
     return badrc ? -1 : 0;
 }
 
-static int update_min_logfile_int(cdb2_hndl_tp *metadb)
-{
-    char cmd[120+nodes_list_sz];
-    char *buf;
-    size_t buf_len;
-    int bytes_written;
-    int rc = 0;
-
-    if (gbl_ready == 0)
-        return 0;
-
-    bytes_written = 0;
-    buf = cmd;
-    buf_len = sizeof(cmd);
-
-    bytes_written +=
-        snprintf(buf+bytes_written, buf_len-bytes_written,
-                "WITH RECURSIVE replication_tree(dbname, host, file) AS "
-                "    (SELECT dbname, host, file FROM comdb2_physreps "
-                "         WHERE dbname='%s' AND host IN (",
-                gbl_dbname);
-    if (bytes_written >= buf_len) {
-        physrep_logmsg(LOGMSG_ERROR, "%s:%d Buffer is not long enough!\n", __func__, __LINE__);
-        return 1;
-    }
-
-    bytes_written += append_quoted_local_hosts(buf+bytes_written, buf_len-bytes_written, ",");
-    if (bytes_written >= buf_len) {
-        physrep_logmsg(LOGMSG_ERROR, "%s:%d Buffer is not long enough!\n", __func__, __LINE__);
-        return 1;
-    }
-
-    bytes_written += snprintf(buf + bytes_written, buf_len - bytes_written,
-                              "     ) "
-                              "     UNION "
-                              "     SELECT p.dbname, p.host, p.file FROM comdb2_physreps p, "
-                              "         comdb2_physrep_connections c, replication_tree t "
-                              "         WHERE p.state = 'Active' AND p.file <> 0 AND "
-                              "             t.dbname = c.source_dbname AND c.dbname = p.dbname) "
-                              "    SELECT file FROM replication_tree WHERE file IS NOT NULL ORDER BY file LIMIT 1");
-    if (bytes_written >= buf_len) {
-        physrep_logmsg(LOGMSG_ERROR, "%s:%d Buffer is not long enough!\n", __func__, __LINE__);
-        return 1;
-    }
-
-    if (gbl_physrep_debug) {
-        physrep_logmsg(LOGMSG_USER, "%s:%d Executing: %s\n", __func__, __LINE__, cmd);
-    }
-
-    rc = cdb2_run_statement(metadb, cmd);
-    if (rc == CDB2_OK) {
-        while ((rc = cdb2_next_record(metadb)) == CDB2_OK) {
-            int64_t *minfile = (int64_t *)cdb2_column_value(metadb, 0);
-            physrep_min_logfile = minfile ? (unsigned int)*minfile : 0;
-        }
-        if (rc == CDB2_OK_DONE)
-            rc = 0;
-    } else {
-        physrep_logmsg(LOGMSG_ERROR, "%s:%d Failed to execute (rc: %d)\n", __func__, __LINE__, rc);
-    }
-
-    return rc;
-}
-
-static int update_min_logfile(void)
-{
-    cdb2_hndl_tp *metadb;
-    int rc, altcnt = gbl_altmetadb_count;
-
-    if ((rc = physrep_get_metadb_or_local_hndl(&metadb)) != 0) {
-        logmsg(LOGMSG_ERROR, "%s: failed to get metadb handle rc=%d\n", __func__, rc);
-    } else {
-        update_min_logfile_int(metadb);
-        cdb2_close(metadb);
-    }
-
-    for (int i = 0; i < altcnt; i++) {
-        if ((rc = get_alt_metadb_hndl(&metadb, i)) != 0) {
-            logmsg(LOGMSG_ERROR, "%s: failed to get alt metadb handle %d rc=%d\n", __func__, i, rc);
-            continue;
-        }
-        update_min_logfile_int(metadb);
-        cdb2_close(metadb);
-    }
-    return 0;
-}
-
 /*
   Check whether we need to wait for a connection from one of the nodes
   in the source db.
@@ -1501,6 +1381,7 @@ static void *physrep_worker(void *args)
     comdb2_name_thread(__func__);
 
     volatile int64_t gen, highest_gen = 0;
+    int do_revconn;
     int64_t first_lcgen = -1;
     size_t sql_cmd_len = 150;
     char sql_cmd[sql_cmd_len];
@@ -1557,7 +1438,7 @@ static void *physrep_worker(void *args)
                 physrep_logmsg(LOGMSG_USER, "%s:%d Re-checking for reverse connection\n", __func__, __LINE__);
             }
             if ((rc = physrep_get_metadb_or_local_hndl(&repl_metadb)) == 0) {
-                int do_revconn = do_wait_for_reverse_conn(repl_metadb);
+                do_revconn = do_wait_for_reverse_conn(repl_metadb);
                 cdb2_close(repl_metadb);
                 if (gbl_physrep_debug) {
                     physrep_logmsg(LOGMSG_USER, "%s:%d Reverse connection check: do-revcon=%d, is-revcon=%d\n",
@@ -1586,8 +1467,10 @@ static void *physrep_worker(void *args)
                 goto sleep_and_retry;
             }
 
-            last_revconn_check = comdb2_time_epoch();
-            if (do_wait_for_reverse_conn(repl_metadb) == 1) {
+            do_revconn = do_wait_for_reverse_conn(repl_metadb);
+
+            if (do_revconn == 1) {
+                last_revconn_check = comdb2_time_epoch();
                 is_revconn = 1;
                 int wait_timeout_sec = 60;
 
@@ -1631,9 +1514,13 @@ static void *physrep_worker(void *args)
                 do_truncate = 1;
 
                 set_repl_db_connected(rev_conn_hndl->remote_dbname, rev_conn_hndl->remote_host);
-            } else {
+            } else if (do_revconn == 0) {
+                last_revconn_check = comdb2_time_epoch();
                 is_revconn = 0;
                 int notfound = 0;
+
+                physrep_logmsg(LOGMSG_DEBUG, "%s:%d do_revconn is 0, registering replicant\n", __func__, __LINE__);
+
                 while (stop_physrep_worker == 0) {
                     if ((rc = register_self(repl_metadb)) == 0)
                         break;
@@ -1668,6 +1555,11 @@ static void *physrep_worker(void *args)
 
                 /* Perform truncation to start fresh */
                 do_truncate = 1;
+            } else {
+                physrep_logmsg(LOGMSG_ERROR, "%s:%d Failed to determine if reverse connection is needed\n", __func__,
+                               __LINE__);
+                cdb2_close(repl_metadb);
+                goto sleep_and_retry;
             }
 
             // Close the connection to metadb as we now have a connection to the
@@ -1957,7 +1849,6 @@ static void *physrep_watcher(void *args) {
     static int physrep_slow_replicant_last_checked;
     static int physrep_keepalive_last_sent;
     static int physrep_hung_replicant_last_checked;
-    static int physrep_minlog_last_checked;
 
     while (!gbl_exit && stop_physrep_watcher == 0) {
         sleep(1);
@@ -1986,20 +1877,12 @@ static void *physrep_watcher(void *args) {
         }
         // Common:
         //   1) Send keepalives
-        //   2) Update maximum log file number upto which it is safe to delete
 
         // Periodically send keepalive to report its LSN.
         if ((now - physrep_keepalive_last_sent) >= gbl_physrep_keepalive_freq_sec) {
             send_keepalive();
             physrep_keepalive_last_sent = now;
         }
-
-        // Update the 'minimum log file' marker upto which it is safe to
-        // delete log files.
-        if ((now - physrep_minlog_last_checked) >= gbl_physrep_check_minlog_freq_sec) {
-            update_min_logfile();
-            physrep_minlog_last_checked = now;
-        }
     }
     return NULL;
 }

db/reverse_conn.c

@@ -68,7 +68,7 @@ int gbl_revsql_allow_command_exec;
 int gbl_revsql_debug = 0;
 int gbl_revsql_cdb2_debug;
 // 'reverse-connection host' list refresh frequency
-int gbl_revsql_host_refresh_freq_sec = 5;
+int gbl_revsql_host_refresh_freq_sec = 60;
 // 'reverse-connection' worker's new connection attempt frequency
 int gbl_revsql_connect_freq_sec = 5;
 

lua/lib/physrep_get_reverse_hosts.lua

@@ -5,14 +5,6 @@
 local function main(dbname, hostname)
     db:begin()
 
-    -- Check whether 'comdb2_physrep_sources' table exists
-    local rs, rc = db:exec("SELECT count(*)=1 AS cnt FROM comdb2_tables WHERE tablename = 'comdb2_physrep_sources'")
-    local row = rs:fetch()
-    if row.cnt == 0 then
-        db:commit()
-        return
-    end
-
     local rs, rc = db:exec("SELECT dbname, host FROM comdb2_physrep_sources WHERE " ..
                            "source_dbname = '" ..  dbname .. "' AND source_host = '" .. hostname .. "'")
     local row = rs:fetch()

lua/lib/physrep_get_reverse_hosts_v2.lua

@@ -5,14 +5,6 @@
 local function main(dbname, hostname)
     db:begin()
 
-    -- Check whether 'comdb2_physrep_sources' table exists
-    local rs, rc = db:exec("SELECT count(*)=1 AS cnt FROM comdb2_tables WHERE tablename = 'comdb2_physrep_sources'")
-    local row = rs:fetch()
-    if row.cnt == 0 then
-        db:commit()
-        return
-    end
-
     local rs, rc = db:exec("SELECT dbname, host FROM comdb2_physrep_sources WHERE " ..
                            "source_dbname = '" ..  dbname .. "' AND source_host = '" .. hostname .. "'")
     local row = rs:fetch()