Unified cancel for sql statements, for both running and queued sql.

Syntax:

exec procedure sys.cmd.cancel('all') // cancel all statements in libevent sql_evbuffers

exec procedure sys.cmd.cancel('running') // cancel all running statements

exec procedure sys.cmd.cancel('queued') // cancel all queued statements

exec procedure sys.cmd.cancel('cnonce', 'uuid') // cancel 'uuid' statement, 'uuid' provided by comdb2_connections

exec procedure sys.cmd.cancel('fp', 'uuid') // cancel all fp 'uuid' statements, fp provided by comdb2_connections.

NOTE: canceled requests will show as 'queued_canceled', and so on, in comdb2_connections

Author: dorinhogea

db/process_message.c

@@ -237,6 +237,11 @@ static const char *HELP_SQL[] = {
     "hist               - show recently run statements",
     "cancel N           - cancel running statement with id N",
     "cancelcnonce N     - cancel running statement with cnonce N",
+    "ucancel cnonce N   - cancel running statement with unified uuid N (per comdb2_connections)",
+    "ucancel fp N       - cancel running statement with fingerprint N (per comdb2_connections)",
+    "ucancel running    - cancel all running statement (leaves queued statements intact)",
+    "ucancel queued     - cancel all queued statement (leaves running statements intact)",
+    "ucancel all        - cancel all queued and running statements",
     "wrtimeout N        - set write timeout in ms",
     "help               - this information",
     NULL,
@@ -3118,6 +3123,34 @@ int process_command(struct dbenv *dbenv, char *line, int lline, int st)
                 cancel_sql_statement_with_cnonce(cnonce);
                 free(cnonce);
             }
+        } else if (tokcmp(tok, ltok, "ucancel") == 0) {
+            char *uuid = NULL;
+            enum ucancel_type t = UCANCEL_INV;
+            tok = segtok(line, lline, &st, &ltok);
+            if (ltok) {
+                if (tokcmp(tok, ltok, "all") == 0)
+                    t = UCANCEL_ALL;
+                else if (tokcmp(tok, ltok, "running") == 0)
+                    t = UCANCEL_RUN;
+                else if (tokcmp(tok, ltok, "queued") == 0)
+                    t = UCANCEL_QUE;
+                else if (tokcmp(tok, ltok, "cnonce") == 0)
+                    t = UCANCEL_CNO;
+                else if (tokcmp(tok, ltok, "fp") == 0)
+                    t = UCANCEL_FPT;
+
+                if (t == UCANCEL_CNO || t == UCANCEL_FPT) {
+                    tok =  segtok(line, lline, &st, &ltok);
+                    if (ltok)
+                        uuid = tokdup(tok, ltok);
+                    else
+                        t = UCANCEL_INV;
+                }
+                if (t != UCANCEL_INV)
+                    ucancel_sql_statements(t, uuid);
+            }
+            if (t == UCANCEL_INV)
+                logmsg(LOGMSG_ERROR, "Usage sql ucancel [all|queued|running|fp N|cnonce N]");
         } else if (tokcmp(tok, ltok, "help") == 0) {
             print_help_page(HELP_SQL);
         } else if (tokcmp(tok, ltok, "debug") == 0) {

db/sql.h

@@ -322,6 +322,17 @@ typedef struct sqlclntstate_fdb {
     int failed_heartbeats; /* used to signal failed communication with remotes */
 } sqlclntstate_fdb_t;
 
+enum ucancel_type {
+    UCANCEL_INV = 0,
+    UCANCEL_ALL = 1,    /* both queued and running */
+    UCANCEL_RUN = 2,    /* running only */
+    UCANCEL_QUE = 4,    /* queued only */
+    UCANCEL_CNO = 8,    /* filter by cnonce */
+    UCANCEL_FPT = 16    /* filter by fp */
+};
+
+int ucancel_sql_statements(enum ucancel_type type, char *uuid);
+
 CurRange *currange_new();
 #define CURRANGEARR_INIT_CAP 2
 void currangearr_init(CurRangeArr *arr);
@@ -745,6 +756,7 @@ struct sqlclntstate {
 
     struct rawnodestats *rawnodestats;
 
+    uuid_t      unifieduuid;         /* assigned to any statement running, used for canceling live sql */
     osqlstate_t osql;                /* offload sql state is kept here */
     enum ctrl_sqleng ctrl_sqlengine; /* use to mark a begin/end out of state,
                                         see enum ctrl_sqleng
@@ -1068,6 +1080,7 @@ struct sqlclntstate {
     int tail_offset;
 
     struct features features;
+    int discard_this; /* set by a cancel() trap, complement thd->stop_this_statement for queued request */
 };
 typedef struct sqlclntstate sqlclntstate;
 
@@ -1338,6 +1351,8 @@ struct connection_info {
     enum connection_state state_int;
     int64_t in_transaction;
     int64_t in_local_cache;
+    char *uuid;
+    int64_t is_canceled;
 };
 
 /* makes master swing verbose */
@@ -1721,6 +1736,8 @@ int osql_test_create_genshard(struct schema_change_type *sc, char **errmsg, int
                               char **dbnames, uint32_t numcols, char **columns, char **shardnames);
 int osql_test_remove_genshard(struct schema_change_type *sc, char **errmsg);
 
+void cancel_connections(int only_queued, uuid_t filter_fp);
+
 struct sp_tmptbl {
     pthread_mutex_t lk;
     uint64_t rowid;

db/sqlglue.c

@@ -629,6 +629,10 @@ int check_sql_client_disconnect(struct sqlclntstate *clnt, char *file, int line)
             return 1;
         }
     }
+    if (clnt->discard_this) {
+        clnt->thd->sqlthd->stop_this_statement = 1;
+        clnt->discard_this = 0;
+    }
     return 0;
 }
 
@@ -665,6 +669,10 @@ static int sql_tick(struct sql_thread *thd, int no_recover_deadlock)
     if (thd->stop_this_statement) {
         rc = SQLITE_ABORT;
         goto done;
+    } else if (clnt->discard_this) {
+        rc = SQLITE_ABORT;
+        clnt->discard_this = 0;
+        goto done;
     }
 
     if (clnt->statement_timedout) {
@@ -9700,6 +9708,105 @@ void cancel_sql_statement_with_cnonce(const char *cnonce)
         logmsg(LOGMSG_USER, "Query with cnonce %s not found (finished?)\n", cnonce);
 }
 
+static int _ucancel_sql_statements_cno(char *uuidstr)
+{
+    struct sql_thread *thd;
+    int rc = -1;
+
+    uuid_t uuid;
+
+    if (uuid_parse(uuidstr, uuid)) {
+        logmsg(LOGMSG_ERROR, "Failed to parse uuid \"%s\"\n", uuidstr);
+        return rc;
+    }
+
+    Pthread_mutex_lock(&gbl_sql_lock);
+    LISTC_FOR_EACH(&thedb->sql_threads, thd, lnk) {
+        if (thd->clnt && memcmp(thd->clnt->unifieduuid, uuid, sizeof(uuid_t)) == 0) {
+            logmsg(LOGMSG_ERROR, "Cancelling sql %s\n", thd->clnt->sql);
+            thd->stop_this_statement = 1;
+            rc = 0;
+            break;
+        }
+    }
+    Pthread_mutex_unlock(&gbl_sql_lock);
+    if (rc)
+        logmsg(LOGMSG_ERROR, "Sql cnonce \"%s\" not found\n", uuidstr);
+    return rc;
+}
+
+static int _ucancel_sql_statements_run(void)
+{
+    struct sql_thread *thd;
+
+    Pthread_mutex_lock(&gbl_sql_lock);
+    LISTC_FOR_EACH(&thedb->sql_threads, thd, lnk) {   
+        if (thd->clnt)
+            thd->stop_this_statement = 1;
+    }
+    Pthread_mutex_unlock(&gbl_sql_lock);
+
+    return 0;
+}
+
+static int _ucancel_sql_statements_que(void)
+{
+    uuid_t zerouuid;
+    comdb2uuid_clear(zerouuid);
+    cancel_connections(1, zerouuid);
+    return 0;
+}
+
+static int _ucancel_sql_statements_all(void)
+{
+    uuid_t zerouuid;
+    comdb2uuid_clear(zerouuid);
+    cancel_connections(0, zerouuid);
+    return 0;
+}
+
+static int _ucancel_sql_statements_fp(char * uuidstr)
+{
+    uuid_t uuid;
+
+    if (uuid_parse(uuidstr, uuid)) {
+        logmsg(LOGMSG_ERROR, "Failed to parse uuid \"%s\"\n", uuidstr);
+        return -1;
+    }
+
+    cancel_connections(0, uuid);
+    return 0;
+}
+
+/*
+ * "Unified" sql statements cancel routine
+ * MODES:
+ *  type == UCANCEL_ALL #cancel all running and queued statements
+ *  type == UCANCEL_RUN #cancel all running statements
+ *  type == UCANCEL_QUE #cancel all queued statements
+ *  type == UCANCEL_CNO #cancel one running statement with cnonce = "uuid"
+ *  type == UCANCEL_FPT #cancel all running statement with fingerprint = "uuid"
+ *
+ */
+int ucancel_sql_statements(enum ucancel_type type, char *uuid) {
+    switch(type) {
+    case UCANCEL_CNO:
+        return _ucancel_sql_statements_cno(uuid);
+    case UCANCEL_RUN:
+        return _ucancel_sql_statements_run();
+    case UCANCEL_QUE:
+        return _ucancel_sql_statements_que();
+    case UCANCEL_ALL:
+        return _ucancel_sql_statements_all();
+    case UCANCEL_FPT:
+        return _ucancel_sql_statements_fp(uuid);
+    default:
+       logmsg(LOGMSG_ERROR, "%d unimplemented\n", type);
+       return -1;
+    }
+    return 0;
+}
+
 void sql_dump_running_statements(void)
 {
     struct sql_thread *thd;

db/sqlinterfaces.c

@@ -4437,7 +4437,13 @@ static void sqlengine_work_lua_thread(void *thddata, void *work)
     clnt->deque_timeus = comdb2_time_epochus();
     clnt->thd = thd;
     /* Reset the cancel-statement flag */
-    thd->sqlthd->stop_this_statement = 0;
+    if (clnt->discard_this) {
+        thd->sqlthd->stop_this_statement = 1;
+        clnt->discard_this = 0;
+    } else
+        thd->sqlthd->stop_this_statement = 0;
+
+    clnt->discard_this = 0;
     sql_update_usertran_state(clnt);
 
     rdlock_schema_lk();
@@ -4695,7 +4701,11 @@ void sqlengine_work_appsock(struct sqlthdstate *thd, struct sqlclntstate *clnt)
     sqlthd->clnt = clnt;
     clnt->thd = thd;
     /* Reset the cancel-statement flag */
-    sqlthd->stop_this_statement = 0;
+    if (clnt->discard_this) {
+        thd->sqlthd->stop_this_statement = 1;
+        clnt->discard_this = 0;
+    } else 
+        thd->sqlthd->stop_this_statement = 0;
 
     thr_set_user("appsock", (intptr_t)clnt->appsock_id);
 
@@ -6475,6 +6485,11 @@ static void gather_connection_int(struct connection_info *c, struct sqlclntstate
     c->in_transaction = clnt->in_client_trans;
     c->in_local_cache = clnt->in_local_cache;
     Pthread_mutex_unlock(&clnt->state_lk);
+    uuidstr_t us;
+    comdb2uuidstr(clnt->unifieduuid, us);
+    c->uuid = malloc(strlen(us) + 1);
+    snprintf(c->uuid, strlen(us) + 1, "%s", us);
+    c->is_canceled = clnt->discard_this;
 }
 
 static void gather_connections_evbuffer(struct connection_info **info, int *num_connections)
@@ -6504,13 +6519,40 @@ void free_connection_info(struct connection_info *info, int num_connections)
 {
     if (info == NULL) return;
     for (int i = 0; i < num_connections; i++) {
-        if (info[i].sql) free(info[i].sql);
-        if (info[i].fingerprint) free(info[i].fingerprint);
+        free(info[i].sql);
+        free(info[i].fingerprint);
+        free(info[i].uuid);
         /* state is static, don't free */
     }
     free(info);
 }
 
+void cancel_connections(int only_queued, uuid_t filter_fp)
+{
+    struct sqlclntstate *clnt;
+    Pthread_mutex_lock(&lru_evbuffers_mtx);
+    TAILQ_FOREACH(clnt, &sql_evbuffers, sql_entry) {
+        if (!comdb2uuid_is_zero(filter_fp)) {
+            if (!comdb2uuidcmp(filter_fp, clnt->unifieduuid)) {
+                clnt->discard_this = 1;
+                fprintf(stderr, "%s:%d Cancelling %s\n", __func__,  __LINE__, clnt->sql);
+            }
+        } else {
+            if (only_queued) {
+                if (clnt->state == CONNECTION_QUEUED) {
+                    clnt->discard_this = 1;
+                    fprintf(stderr, "%s:%d Cancelling %s\n", __func__,  __LINE__, clnt->sql);
+                }
+            } else if (clnt->state == CONNECTION_RUNNING ||
+                    clnt->state == CONNECTION_QUEUED) {
+                clnt->discard_this = 1;
+                fprintf(stderr, "%s:%d Cancelling %s\n", __func__,  __LINE__, clnt->sql);
+            }
+        }
+    }
+    Pthread_mutex_unlock(&lru_evbuffers_mtx);
+}
+
 void reqlog_long_running_sql_statements(void)
 {
     struct sqlclntstate *clnt;