Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions db/osqlsqlthr.c
Original file line number Diff line number Diff line change
Expand Up @@ -246,9 +246,12 @@ static int osql_sock_start_int(struct sqlclntstate *clnt, int type,
rc = clnt_check_bdb_lock_desired(clnt);
if (rc) {
logmsg(LOGMSG_ERROR, "recover_deadlock returned %d\n", rc);
rc = osql_end(clnt);
if (rc) {
logmsg(LOGMSG_ERROR, "%s failed to end osql %d\n", __func__, rc);
/* keep_rqid: session is being restarted, don't zero rqid */
if (!keep_rqid) {
rc = osql_end(clnt);
if (rc) {
logmsg(LOGMSG_ERROR, "%s failed to end osql %d\n", __func__, rc);
}
}
return SQLITE_BUSY;
}
Expand Down Expand Up @@ -1211,7 +1214,8 @@ int osql_sock_commit(struct sqlclntstate *clnt, int type, enum trans_clntcomm si
int keep_session = !is_final || (get_cnonce(clnt, &snap) == 0);

rc = osql_sock_restart(clnt, 1, keep_session, is_final);
if (sock_restart_retryable_rcode(rc) && !clnt->is_coordinator) {
/* don't resend on an ended session (rqid 0) */
if (sock_restart_retryable_rcode(rc) && osql->rqid && !clnt->is_coordinator) {
if (gbl_master_swing_sock_restart_sleep) {
sleep(gbl_master_swing_sock_restart_sleep);
}
Expand Down
5 changes: 4 additions & 1 deletion db/reqlog.c
Original file line number Diff line number Diff line change
Expand Up @@ -1992,7 +1992,10 @@ void reqlog_long_running_clnt(struct sqlclntstate *clnt)
{
int have_fingerprint = 0;
char fp[FINGERPRINTSZ] = {0};
if (clnt->done || !clnt->thd || !clnt->sql || !clnt->thd->logger) return;
/* single read: worker clears clnt->thd concurrently */
struct sqlthdstate *thd = clnt->thd;
if (clnt->done || !thd || !clnt->sql || !thd->logger)
return;

if (can_consume(clnt) == 1) {
return; /* Do not log consumers */
Expand Down
6 changes: 5 additions & 1 deletion db/sqlglue.c
Original file line number Diff line number Diff line change
Expand Up @@ -10268,7 +10268,11 @@ int recover_deadlock_flags(bdb_state_type *bdb_state, struct sqlclntstate *clnt,
clnt->recover_deadlock_thd = pthread_self();
comdb2_cheapstack_char_array(clnt->recover_deadlock_stack, RECOVER_DEADLOCK_MAX_STACK);
#endif
recover_deadlock_sc_cleanup(clnt->thd->sqlthd);
/* use TLS thd (as recover_deadlock_flags_int does): clnt->thd may
* already be cleared when called from the post-done flush path */
struct sql_thread *sqlthd = pthread_getspecific(query_info_key);
if (sqlthd)
recover_deadlock_sc_cleanup(sqlthd);
assert(bdb_lockref() == 0);
} else {
assert(bdb_lockref() > 0);
Expand Down
29 changes: 26 additions & 3 deletions db/sqlinterfaces.c
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,7 @@ void rcache_destroy(void);
void sql_reset_sqlthread(struct sql_thread *thd);
int blockproc2sql_error(int rc, const char *func, int line);
static int test_no_btcursors(struct sqlthdstate *thd);
static void clnt_detach_thd(struct sqlclntstate *clnt, struct sql_thread *sqlthd);
static void sql_thread_describe(void *obj, FILE *out);
static char *get_query_cost_as_string(struct sql_thread *, struct sqlclntstate *);
void handle_sql_intrans_unrecoverable_error(struct sqlclntstate *clnt);
Expand Down Expand Up @@ -4081,8 +4082,8 @@ static int run_stmt(struct sqlthdstate *thd, struct sqlclntstate *clnt,
if (rc)
return rc;
} else {
postponed_write = 1;
send_row(clnt, stmt, row_id, 1, NULL);
/* only claim a postponed row if the save succeeded */
postponed_write = (send_row(clnt, stmt, row_id, 1, NULL) == 0);
}

rowcount++;
Expand Down Expand Up @@ -4626,6 +4627,7 @@ static void sqlengine_work_lua_thread(void *thddata, void *work)
osql_log_time_done(clnt);

debug_close_clnt(clnt);
clnt_detach_thd(clnt, thd->sqlthd);
signal_clnt_as_done(clnt);

thrman_setid(thrman_self(), "[done]");
Expand Down Expand Up @@ -4858,6 +4860,22 @@ static int can_execute_sql_query_now(
return 1;
}

/* clear before signal: signal hands clnt back to the event thread.
* nested replay call: outer frame still owns thd.
* only call this while we still own clnt (ie not after a redispatch) */
static void clnt_detach_thd(struct sqlclntstate *clnt, struct sql_thread *sqlthd)
{
if (clnt->osql.in_replay_nested)
return;
Pthread_mutex_lock(&gbl_sql_lock);
sqlthd->clnt = NULL;
Pthread_mutex_unlock(&gbl_sql_lock);
/* sql_lk: watchdog reads clnt->thd under it */
Pthread_mutex_lock(&clnt->sql_lk);
clnt->thd = NULL; /* thd is about to go away */
Pthread_mutex_unlock(&clnt->sql_lk);
}

void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
{
struct sql_thread *sqlthd = thd->sqlthd;
Expand Down Expand Up @@ -4891,7 +4909,10 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
if (srs_tran_replay(clnt) == RC_INTERNAL_RETRY) {
/* Another iteration was scheduled on a new worker.
* That worker now owns the clnt; do NOT signal_clnt_as_done
* here or it would race with the new worker's enqueue. */
* here or it would race with the new worker's enqueue.
* Do NOT clnt_detach_thd() either: enqueue_sql_query() already
* cleared clnt->thd before dispatching, and the new owner may
* have finished and freed clnt by now. */
thrman_setid(thrman_self(), "[done]");
return;
}
Expand All @@ -4908,6 +4929,7 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
clnt->osql.timings.query_finished = osql_log_time();
osql_log_time_done(clnt);
clnt_change_state(clnt, CONNECTION_IDLE);
clnt_detach_thd(clnt, sqlthd);
signal_clnt_as_done(clnt);
return;
}
Expand Down Expand Up @@ -4960,6 +4982,7 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
osql_log_time_done(clnt);
clnt_change_state(clnt, CONNECTION_IDLE);
debug_close_clnt(clnt);
clnt_detach_thd(clnt, sqlthd);
signal_clnt_as_done(clnt);

thrman_setid(thrman_self(), "[done]");
Expand Down
2 changes: 2 additions & 0 deletions db/sqloffload.c
Original file line number Diff line number Diff line change
Expand Up @@ -641,7 +641,9 @@ int osql_clean_sqlclntstate(struct sqlclntstate *clnt)
abort();
}

int in_replay_nested = osql->in_replay_nested; /* call-stack state, not txn state */
bzero(osql, sizeof(*osql));
osql->in_replay_nested = in_replay_nested;
listc_init(&osql->shadtbls, offsetof(struct shad_tbl, linkv));

sql_set_sqlengine_state(clnt, __FILE__, __LINE__, SQLENG_NORMAL_PROCESS);
Expand Down
2 changes: 2 additions & 0 deletions plugins/newsql/newsql.c
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,8 @@ static int newsql_save_postponed_row(struct sqlclntstate *clnt,
static int newsql_send_postponed_row(struct sqlclntstate *clnt)
{
struct newsql_appdata *appdata = clnt->appdata;
if (appdata->postponed == NULL)
return -1;
return appdata->write_postponed(clnt);
}

Expand Down
Loading