{180685083} feat(reqlog): post schema change activity

Statreqs tell us what the database is busy doing. If any schema changes are
running, that is significant and worth reporting in statreqs.

Signed-off-by: Shoumyo Chakravorti <[email protected]>

Author: SChakravorti21

db/comdb2.c

@@ -141,6 +141,7 @@ void berk_memp_sync_alarm_ms(int);
 #include <net_appsock.h>
 #include "sc_csc2.h"
 #include "sc_util.h"
+#include "sc_global.h"
 #include "reverse_conn.h"
 #include "alias.h"
 #include "str_util.h" /* QUOTE */
@@ -4626,8 +4627,6 @@ int throttle_lim = 10000;
 int cpu_throttle_threshold = 100000;
 
 double gbl_cpupercent;
-#include <sc_util.h>
-
 
 static inline void log_tbl_item(int64_t curr, int64_t *prev, const char *(*type_to_str)(int), int type, char *string,
                                 int *hdr_p, struct reqlogger *statlogger, dbtable *tbl, int first) 
@@ -4729,6 +4728,8 @@ void *statthd(void *p)
     int have_scon_stats = 0;
     int64_t rw_evicts;
 
+    int last_num_sc = 0;
+    struct running_sc_info *last_schema_changes = NULL;
 
     thrman_register(THRTYPE_GENERIC);
     thread_started("statthd");
@@ -4902,7 +4903,9 @@ void *statthd(void *p)
             }
 
             int aa_include_updates = bdb_attr_get(thedb->bdb_attr, BDB_ATTR_AA_COUNT_UPD);
+
             rdlock_schema_lk();
+
             for (ii = 0; ii < dbenv->num_dbs; ++ii) {
                 dbtable *tbl = dbenv->dbs[ii];
                 int hdr = 0;
@@ -4949,6 +4952,50 @@ void *statthd(void *p)
                 log_tbl_item(tbl->deadlock_count, &tbl->saved_deadlock_count, NULL, 0, "deadlock count", &hdr,
                              statlogger, tbl, 0);
             }
+
+            int num_sc = 0;
+            struct running_sc_info *schema_changes = NULL;
+
+            if (list_running_schema_changes(&schema_changes, &num_sc) == 0) {
+                if (num_sc > 0) {
+                    /* log header */
+                    reqlog_logf(statlogger, REQL_INFO, "SCHEMA CHANGE STATS\n");
+                }
+
+                for (ii = 0; ii < num_sc; ii++) {
+                    struct running_sc_info sc = schema_changes[ii];
+                    struct running_sc_info *last_sc = NULL;
+
+                    for (jj = 0; jj < last_num_sc; jj++) {
+                        if (strcmp(last_schema_changes[jj].table_name, sc.table_name) == 0) {
+                            last_sc = &last_schema_changes[jj];
+                            break;
+                        }
+                    }
+
+                    uint64_t last_nrecs = last_sc ? last_sc->nrecs : 0;
+                    uint32_t last_adds = last_sc ? last_sc->adds : 0;
+                    uint32_t last_updates = last_sc ? last_sc->updates : 0;
+                    uint32_t last_deletes = last_sc ? last_sc->deletes : 0;
+
+                    uint64_t diff_nrecs = sc.nrecs - last_nrecs;
+
+                    reqlog_logf(
+                        statlogger, REQL_INFO,
+                        "    table '%s' records converted %ld diff %ld rate %ld r/s adds %d updates %d deletes %d\n",
+                        sc.table_name,                             //
+                        sc.nrecs, diff_nrecs, diff_nrecs / thresh, //
+                        sc.adds - last_adds,                       //
+                        sc.updates - last_updates,                 //
+                        sc.deletes - last_deletes);
+                }
+
+                free(last_schema_changes);
+
+                last_num_sc = num_sc;
+                last_schema_changes = schema_changes;
+            }
+
             unlock_schema_lk();
 
             pstats = bdb_get_process_stats();

schemachange/sc_global.c

@@ -697,3 +697,49 @@ void sc_alter_latency(int counter)
         }
     }
 }
+
+int list_running_schema_changes(struct running_sc_info **info, int *num_running_sc)
+{
+    unsigned int bucket;
+    void *entry;
+    sc_table_t *sc_table = NULL;
+    *num_running_sc = 0;
+
+    Pthread_mutex_lock(&schema_change_in_progress_mutex);
+
+    if (!sc_tables) {
+        Pthread_mutex_unlock(&schema_change_in_progress_mutex);
+        return 1;
+    }
+
+    /* preallocate outparam */
+    int unused;
+    int num_entries = 0;
+    hash_info(sc_tables, &num_entries, &unused, &unused, &unused, &unused, &unused, &unused);
+    *info = realloc(*info, sizeof(struct running_sc_info) * num_entries);
+
+    for (sc_table = hash_first(sc_tables, &entry, &bucket); sc_table;
+         sc_table = hash_next(sc_tables, &entry, &bucket)) {
+
+        /* get_dbtable_by_name looks at thedb->db_hash, so caller must
+         * hold the schema lock */
+        struct dbtable *db = get_dbtable_by_name(sc_table->tablename);
+
+        Pthread_rwlock_rdlock(&db->sc_live_lk);
+
+        if (db->doing_conversion || db->doing_upgrade) {
+            struct running_sc_info *sc_info = &((*info)[*num_running_sc]);
+            sc_info->table_name = strdup(sc_table->tablename);
+            sc_info->nrecs = db->sc_nrecs;
+            sc_info->adds = db->sc_adds;
+            sc_info->updates = db->sc_updates;
+            sc_info->deletes = db->sc_deletes;
+            ++(*num_running_sc);
+        }
+
+        Pthread_rwlock_unlock(&db->sc_live_lk);
+    }
+
+    Pthread_mutex_unlock(&schema_change_in_progress_mutex);
+    return 0;
+}

schemachange/sc_global.h

@@ -17,6 +17,10 @@
 #ifndef INCLUDE_SC_GLOBAL_H
 #define INCLUDE_SC_GLOBAL_H
 
+#include <stdint.h>
+
+#include "schemachange.h"
+
 extern pthread_mutex_t schema_change_in_progress_mutex;
 extern pthread_mutex_t fastinit_in_progress_mutex;
 extern pthread_mutex_t schema_change_sbuf2_lock;
@@ -60,6 +64,14 @@ extern int rep_sync_save;
 extern int log_sync_save;
 extern int log_sync_time_save;
 
+struct running_sc_info {
+    const char *table_name;
+    uint64_t nrecs;   /* num records converted (does not include `adds`) */
+    uint32_t adds;    /* num added in front of sc cursor */
+    uint32_t updates; /* num updated behind sc cursor */
+    uint32_t deletes; /* num deleted behind sc cursor */
+};
+
 int is_dta_being_rebuilt(struct scplan *plan);
 const char *get_sc_to_name(const char *);
 void wait_for_sc_to_stop(const char *operation, const char *func, int line);
@@ -87,5 +99,8 @@ struct schema_change_type *preempt_ongoing_alter(char *table, int action);
 void clear_ongoing_alter();
 int get_stopsc(const char *func, int line);
 void sc_alter_latency(int counter);
+/* List all running schema changes and their progress. Caller must acquire schema
+ * lock in read mode. */
+int list_running_schema_changes(struct running_sc_info **info, int *num_running_sc);
 
 #endif

schemachange/sc_records.c

@@ -140,16 +140,20 @@ static inline int print_aggregate_sc_stat(struct convert_record_data *data,
     /* totals across all threads */
     if (data->scanmode != SCAN_PARALLEL) return 1;
 
-    long long total_nrecs_diff =
-        data->from->sc_nrecs - data->from->sc_prev_nrecs;
+    uint64_t total_nrecs_diff = data->from->sc_nrecs - data->from->sc_prev_nrecs;
     data->from->sc_prev_nrecs = data->from->sc_nrecs;
+
+    /* actual progress made after accounting for updates and deletes that
+     * happened behind the schema change cursor (by definition, all adds happen
+     * in front of the schema change cursor) */
+    uint64_t actual_nrecs = data->from->sc_nrecs - (data->from->sc_updates + data->from->sc_deletes);
+
     sc_printf(data->s,
-              "[%s] progress TOTAL %lld +%lld actual "
-              "progress total %lld rate %lld r/s\n",
-              data->from->tablename, data->from->sc_nrecs, total_nrecs_diff,
-              data->from->sc_nrecs -
-                  (data->from->sc_adds + data->from->sc_updates),
-              total_nrecs_diff / sc_report_freq);
+              "[%s] progress TOTAL %lld +%lld rate %lld r/s "
+              "actual progress total %lld\n",
+              data->from->tablename, data->from->sc_nrecs, total_nrecs_diff, total_nrecs_diff / sc_report_freq,
+              actual_nrecs);
+
     return 1;
 }