recover_ddlk: Acquire schema-lk before reopening curtran

Signed-off-by: Akshat Sikarwar <[email protected]>

db/reqlog.c

@@ -2389,7 +2389,7 @@ static void update_clientstats_cache(nodestats_t *entry) {
     Pthread_mutex_lock(&clientstats_cache_mtx);
     listc_maybe_rfl(&clientstats_cache, entry);
     if (!entry->ref) {
-        logmsg(LOGMSG_INFO, "%s: no active references to nodestats %u, marking as evictable.\n", __func__, entry->checksum);
+        //logmsg(LOGMSG_INFO, "%s: no active references to nodestats %u, marking as evictable.\n", __func__, entry->checksum);
         listc_abl(&clientstats_cache, entry);
     }
     Pthread_mutex_unlock(&clientstats_cache_mtx);

db/sqlglue.c

@@ -10047,8 +10047,11 @@ static int recover_deadlock_flags_int(bdb_state_type *bdb_state,
         } else {
             rc = SQLITE_SCHEMA;
         }
-    } else
+    } else {
+        rdlock_schema_lk();
         rc = get_curtran_flags(thedb->bdb_env, clnt, curtran_flags);
+        unlock_schema_lk();
+    }
 
     if (rc) {
         char *err;

db/sqlinterfaces.c

@@ -5523,10 +5523,7 @@ int recover_deadlock_evbuffer(struct sqlclntstate *clnt)
     if (gbl_fail_client_write_lock && !(rand() % gbl_fail_client_write_lock)) {
         flags = RECOVER_DEADLOCK_FORCE_FAIL;
     }
-    if (!recover_deadlock_flags(env, clnt, NULL, 0, __func__, __LINE__, flags)) {
-        return -1;
-    }
-    return 0;
+    return recover_deadlock_flags(env, clnt, NULL, 0, __func__, __LINE__, flags);
 }
 
 static int recover_deadlock_sbuf(struct sqlclntstate *clnt)

net/sqlwriter.c

@@ -52,12 +52,11 @@ struct sqlwriter {
     unsigned timed_out : 1;
     unsigned wr_continue : 1;
     unsigned packing : 1; /* 1 if writer is in sql_pack_response and wr_lock is held. */
+    unsigned recover_ddlk_failed : 1;
     struct ssl_data *ssl_data;
     int (*wr_evbuffer_fn)(struct sqlwriter *, int);
 };
 
-static void sql_trickle_cb(int fd, short what, void *arg);
-
 static void sql_enable_flush(struct sqlwriter *writer)
 {
     struct timeval recover_ddlk_timeout = {.tv_sec = 1};
@@ -171,8 +170,8 @@ static int wr_evbuffer(struct sqlwriter *writer, int fd)
 static void sql_flush_cb(int fd, short what, void *arg)
 {
     struct sqlwriter *writer = arg;
-    if (what & EV_TIMEOUT) {
-        recover_deadlock_evbuffer(writer->clnt);
+    if ((what & EV_TIMEOUT) && !writer->recover_ddlk_failed) {
+        writer->recover_ddlk_failed = !!recover_deadlock_evbuffer(writer->clnt);
     }
     if (!(what & EV_WRITE)) {
         return;
@@ -195,8 +194,14 @@ static void sql_flush_cb(int fd, short what, void *arg)
 static int sql_flush_int(struct sqlwriter *writer)
 {
     sql_enable_flush(writer);
-    event_base_dispatch(writer->wr_base);
-    return (writer->wr_continue && !writer->bad) ? 0 : -1;
+    event_base_dispatch(writer->wr_base); /* -> sql_flush_cb */
+    if (writer->bad ||
+        writer->recover_ddlk_failed ||
+        writer->wr_continue == 0
+    ){
+        return -1;
+    }
+    return 0;
 }
 
 int sql_flush(struct sqlwriter *writer)
@@ -405,6 +410,7 @@ void sql_reset(struct sqlwriter *writer)
     writer->sent_at = time(NULL);
     writer->timed_out = 0;
     writer->wr_continue = 1;
+    writer->recover_ddlk_failed = 0;
 }
 
 int sql_peer_check(struct sqlwriter *writer)
@@ -483,7 +489,6 @@ struct sqlwriter *sqlwriter_new(struct sqlwriter_arg *arg)
     writer->pack_hb = arg->pack_hb;
     writer->timer_base = arg->timer_base;
     writer->timer_thd = pthread_self();
-    writer->wr_continue = 1;
     writer->wr_buf = evbuffer_new();
 
     struct event_config *cfg = event_config_new();
@@ -496,6 +501,8 @@ struct sqlwriter *sqlwriter_new(struct sqlwriter_arg *arg)
     writer->heartbeat_ev = event_new(writer->timer_base, arg->fd, EV_PERSIST, sql_heartbeat_cb, writer);
     writer->heartbeat_trickle_ev = event_new(writer->timer_base, arg->fd, EV_WRITE, sql_trickle_cb, writer);
 
+    sql_reset(writer);
+
     return writer;
 }