Remove physrep min-log logic

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

@@ -520,7 +520,6 @@ extern int gbl_tranlog_maxpoll;
 
 /* Physical replication */
 extern int gbl_blocking_physrep;
-extern int gbl_physrep_check_minlog_freq_sec;
 extern int gbl_physrep_debug;
 extern int gbl_physrep_exit_on_invalid_logstream;
 extern int gbl_physrep_fanout;

db/db_tunables.h

@@ -1860,8 +1860,6 @@ 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: 600)",
-                 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)",

db/phys_rep.c

@@ -70,7 +70,6 @@ int gbl_deferred_phys_flag = 0;
 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_check_minlog_freq_sec = 600;
 int gbl_physrep_hung_replicant_threshold = 60;
 int gbl_physrep_revconn_check_interval = 60;
 int gbl_physrep_update_registry_interval = 60;
@@ -79,7 +78,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;
@@ -1205,10 +1203,6 @@ 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) {
@@ -1248,38 +1242,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];
@@ -1340,94 +1302,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);
-    }
-
-    ATOMIC_ADD64(gbl_physrep_metadb_sql_count, 1);
-    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.
@@ -1994,7 +1868,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);
@@ -2030,13 +1903,6 @@ static void *physrep_watcher(void *args) {
             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/phys_rep.h

@@ -49,7 +49,6 @@ int stop_physrep_threads();
 int physrep_exited();
 int physrep_get_metadb_or_local_hndl(cdb2_hndl_tp**);
 void physrep_cleanup(void);
-void physrep_update_low_file_num(int*, int*);
 void physrep_fanout_override(const char *dbname, int fanout);
 int physrep_fanout_get(const char *dbname);
 void physrep_fanout_dump(void);

docs/pages/operating/physical_replication.md

@@ -171,7 +171,6 @@ CREATE TABLE comdb2_physrep_sources(dbname CSTRING(60),
 ## Tunables
 
 * blocking_physrep: The `SELECT .. FROM comdb2_transaction_logs` query executed by physical replicants blocks for the next log record. (Default: `false`)
-* physrep_check_minlog_freq_sec: Check the minimum log number to keep this often. (Default: `600`)
 * physrep_debug: Print extended physrep trace. (Default: `off`)
 * physrep_exit_on_invalid_logstream: Exit physreps on invalid logstream. (Default: off)
 * physrep_fanout: Maximum number of physical replicants that a node can service (Default: `8`)

tests/tunables.test/t00_all_tunables.expected

@@ -744,7 +744,6 @@
 (name='pgcompactpool.stacksz', description='Thread stack size.', type='INTEGER', value='1048576', read_only='N')
 (name='physical_ack_interval', description='For logical transactions, have the slave send an 'ack' after this many physical operations.', type='INTEGER', value='0', read_only='N')
 (name='physical_commit_interval', description='Force a physical commit after this many physical operations.', type='INTEGER', value='512', read_only='N')
-(name='physrep_check_minlog_freq_sec', description='Check the minimum log number to keep this often. (Default: 600)', type='INTEGER', value='600', read_only='N')
 (name='physrep_debug', description='Print extended physrep trace. (Default: off)', type='BOOLEAN', value='OFF', read_only='N')
 (name='physrep_exit_on_invalid_logstream', description='Exit physreps on invalid logstream.  (Default: off)', type='BOOLEAN', value='OFF', read_only='N')
 (name='physrep_fanout', description='Maximum number of physical replicants that a node can service (Default: 8)', type='INTEGER', value='8', read_only='N')