Fix lock-inversion on physrep-sc

Signed-off-by: Mark Hannum <[email protected]>

bdb/phys_rep_lsn.c

@@ -250,9 +250,13 @@ uint32_t get_next_offset(DB_ENV *dbenv, LOG_INFO log_info)
     return log_info.offset + log_info.size + dbenv->get_log_header_size(dbenv);
 }
 
-int apply_log(DB_ENV *dbenv, unsigned int file, unsigned int offset, int64_t rectype, void *blob, int blob_len)
+//int apply_log(DB_ENV *dbenv, unsigned int file, unsigned int offset, int64_t rectype, void *blob, int blob_len)
+int apply_log(bdb_state_type *bdb_state, unsigned int file, unsigned int offset, int64_t rectype, void *blob, int blob_len)
 {
-    return dbenv->apply_log(dbenv, file, offset, rectype, blob, blob_len);
+    BDB_READLOCK("apply_log");
+    int rc = bdb_state->dbenv->apply_log(bdb_state->dbenv, file, offset, rectype, blob, blob_len);
+    BDB_RELLOCK();
+    return rc;
 }
 
 int truncate_log_lock(bdb_state_type *bdb_state, unsigned int file, unsigned int offset, uint32_t flags)

bdb/phys_rep_lsn.h

@@ -35,7 +35,7 @@ int physrep_list_ignored_tables(void);
 LOG_INFO get_last_lsn(struct bdb_state_tag *);
 LOG_INFO get_first_lsn(struct bdb_state_tag *);
 uint32_t get_next_offset(struct __db_env *, LOG_INFO log_info);
-int apply_log(struct __db_env *, unsigned int file, unsigned int offset,
+int apply_log(struct bdb_state_tag *, unsigned int file, unsigned int offset,
               int64_t rectype, void *blob, int blob_len);
 int truncate_log_lock(struct bdb_state_tag *, unsigned int file,
                       unsigned int offset, uint32_t flags);

db/phys_rep.c

@@ -638,7 +638,7 @@ static LOG_INFO handle_record(cdb2_hndl_tp *repl_db, LOG_INFO prev_info)
     if (stop_physrep_worker == 0) {
         /* check if we need to call new file flag */
         if (prev_info.file < file) {
-            rc = apply_log(thedb->bdb_env->dbenv, prev_info.file,
+            rc = apply_log(thedb->bdb_env, prev_info.file,
                            get_next_offset(thedb->bdb_env->dbenv, prev_info),
                            REP_NEWFILE, NULL, 0);
             if (rc != 0) {
@@ -648,7 +648,7 @@ static LOG_INFO handle_record(cdb2_hndl_tp *repl_db, LOG_INFO prev_info)
             }
         }
 
-        rc = apply_log(thedb->bdb_env->dbenv, file, offset, REP_LOG, blob,
+        rc = apply_log(thedb->bdb_env, file, offset, REP_LOG, blob,
                        blob_len);
 
         if (is_commit((u_int32_t)*rectype)) {

db/request_stats.c

@@ -21,6 +21,7 @@
 #include <string.h>
 #include <strings.h>
 #include <stddef.h>
+#include <comdb2_atomic.h>
 
 #include "request_stats.h"
 
@@ -34,6 +35,8 @@ extern void __berkdb_register_fsync_callback(void (*callback)(int fd));
 
 static void user_request_done(void *st) { free(st); }
 
+static struct global_stats global = {0};
+
 void user_request_begin(enum request_type type, int flags)
 {
     struct per_request_stats *st;
@@ -67,6 +70,8 @@ void user_request_fsync_callback(int fd)
     if (st) {
         st->nfsyncs++;
     }
+
+    ATOMIC_ADD64(global.total_nfsyncs, 1);
 }
 
 void user_request_read_callback(int bytes)
@@ -81,6 +86,9 @@ void user_request_read_callback(int bytes)
         st->nreads++;
         st->readbytes += bytes;
     }
+
+    ATOMIC_ADD64(global.total_nreads, 1);
+    ATOMIC_ADD64(global.total_readbytes, bytes);
 }
 
 void user_request_write_callback(int bytes)
@@ -95,6 +103,9 @@ void user_request_write_callback(int bytes)
         st->nwrites++;
         st->writebytes += bytes;
     }
+
+    ATOMIC_ADD64(global.total_nwrites, 1);
+    ATOMIC_ADD64(global.total_writebytes, bytes);
 }
 
 void user_request_memp_callback(void)
@@ -107,6 +118,17 @@ void user_request_memp_callback(void)
     st = pthread_getspecific(key);
     if (st)
         st->mempgets++;
+
+    ATOMIC_ADD64(global.total_mempgets, 1);
+}
+
+void global_request_stats(struct global_stats *stats) { 
+    stats->total_nreads = ATOMIC_LOAD64(global.total_nreads);
+    stats->total_nwrites = ATOMIC_LOAD64(global.total_nwrites);
+    stats->total_nfsyncs = ATOMIC_LOAD64(global.total_nfsyncs);
+    stats->total_mempgets = ATOMIC_LOAD64(global.total_mempgets);
+    stats->total_readbytes = ATOMIC_LOAD64(global.total_readbytes);
+    stats->total_writebytes = ATOMIC_LOAD64(global.total_writebytes);
 }
 
 void user_request_init(void)

db/request_stats.h

@@ -55,13 +55,23 @@ struct per_request_stats {
     long long writebytes;
 };
 
+struct global_stats {
+    int64_t total_nreads;
+    int64_t total_nwrites;
+    int64_t total_nfsyncs;
+    int64_t total_mempgets;
+    int64_t total_readbytes;
+    int64_t total_writebytes;
+};
+
 void user_request_begin(enum request_type type, int flags);
 struct per_request_stats *user_request_get_stats(void);
 void user_request_end(void);
 void user_request_read_callback(int);
 void user_request_write_callback(int);
 void user_request_memp_callback(void);
 void user_request_init(void);
+void global_request_stats(struct global_stats *stats);
 
 void user_request_on(void);
 void user_request_off(void);