Add infrastructure for SQL drivers to use trunks

diff --git a/src/modules/rlm_sql/rlm_sql.c b/src/modules/rlm_sql/rlm_sql.c
index 6b2c19e471ce8d62fd66155ed632f4d694f7913b..2c68ff5e804fa7cfbbfa9dcd0bc23585c395200b 100644 (file)
--- a/src/modules/rlm_sql/rlm_sql.c
+++ b/src/modules/rlm_sql/rlm_sql.c
@@ -2055,8 +2055,13 @@ static int mod_instantiate(module_inst_ctx_t const *mctx)
        /*
         *      Export these methods, too.  This avoids RTDL_GLOBAL.
         */
-       inst->query                     = rlm_sql_query;
-       inst->select                    = rlm_sql_select_query;
+       if (inst->driver->uses_trunks) {
+               inst->query             = rlm_sql_trunk_query;
+               inst->select            = rlm_sql_trunk_query;
+       } else {
+               inst->query             = rlm_sql_query;
+               inst->select            = rlm_sql_select_query;
+       }
        inst->fetch_row                 = rlm_sql_fetch_row;
        inst->query_alloc               = fr_sql_query_alloc;
 

diff --git a/src/modules/rlm_sql/rlm_sql.h b/src/modules/rlm_sql/rlm_sql.h
index 207916c3d5b16546793f49fbf1a85344d28722cf..5fe28cc46121aabafa996b857fe5537fbfbad127 100644 (file)
--- a/src/modules/rlm_sql/rlm_sql.h
+++ b/src/modules/rlm_sql/rlm_sql.h
@@ -204,8 +204,11 @@ typedef struct {
        sql_rcode_t     (*sql_socket_init)(rlm_sql_handle_t *handle, rlm_sql_config_t const *config,
                                           fr_time_delta_t timeout);
 
-       unlang_function_t       sql_query;
-       unlang_function_t       sql_select_query;
+       unlang_function_t       sql_query;                      //!< Run an SQL query on a pool connection.
+       unlang_function_t       sql_select_query;               //!< Run an SQL select query on a pool connection.
+
+       unlang_function_t       sql_query_resume;               //!< Callback run after an SQL trunk query is run.
+       unlang_function_t       sql_select_query_resume;        //!< Callback run after an SQL select trunk query is run.
 
        int             (*sql_num_rows)(fr_sql_query_t *query_ctx, rlm_sql_config_t const *config);
        int             (*sql_affected_rows)(fr_sql_query_t *query_ctx, rlm_sql_config_t const *config);
@@ -256,6 +259,7 @@ unlang_action_t     sql_get_map_list(request_t *request, fr_sql_map_ctx_t *map_ctx,
 void           rlm_sql_query_log(rlm_sql_t const *inst, char const *filename, char const *query) CC_HINT(nonnull);
 unlang_action_t rlm_sql_select_query(rlm_rcode_t *p_result, UNUSED int *priority, request_t *request, void *uctx);
 unlang_action_t        rlm_sql_query(rlm_rcode_t *p_result, int *priority, request_t *request, void *uctx);
+unlang_action_t rlm_sql_trunk_query(rlm_rcode_t *p_result, UNUSED int *priority, request_t *request, void *uctx);
 unlang_action_t rlm_sql_fetch_row(rlm_rcode_t *p_result, UNUSED int *priority, request_t *request, void *uctx);
 void           rlm_sql_print_error(rlm_sql_t const *inst, request_t *request, fr_sql_query_t *query_ctx, bool force_debug);
 fr_sql_query_t *fr_sql_query_alloc(TALLOC_CTX *ctx, rlm_sql_t const *inst, request_t *request, rlm_sql_handle_t *handle, fr_trunk_t *trunk, char const *query_str, fr_sql_query_type_t type);

diff --git a/src/modules/rlm_sql/sql.c b/src/modules/rlm_sql/sql.c
index 3b5648c16f1853fd942ef205d6fa99eeee660c2f..a912c5138ffb3f6ef6c42d8b7275aa116cbacdcb 100644 (file)
--- a/src/modules/rlm_sql/sql.c
+++ b/src/modules/rlm_sql/sql.c
@@ -392,6 +392,7 @@ void rlm_sql_print_error(rlm_sql_t const *inst, request_t *request, fr_sql_query
 
 /** Automatically run the correct `finish` function when freeing an SQL query
  *
+ * And mark any associated trunk request as complete.
  */
 static int fr_sql_query_free(fr_sql_query_t *to_free)
 {
@@ -401,6 +402,7 @@ static int fr_sql_query_free(fr_sql_query_t *to_free)
        } else {
                (to_free->inst->driver->sql_finish_query)(to_free, &to_free->inst->config);
        }
+       if (to_free->treq) fr_trunk_request_signal_complete(to_free->treq);
        return 0;
 }
 
@@ -531,6 +533,84 @@ unlang_action_t rlm_sql_query(rlm_rcode_t *p_result, UNUSED int *priority, reque
        RETURN_MODULE_FAIL;
 }
 
+/** Yield processing after submitting a trunk request.
+ */
+static unlang_action_t sql_trunk_query_start(UNUSED rlm_rcode_t *p_result, UNUSED int *priority,
+                                      UNUSED request_t *request, UNUSED void *uctx)
+{
+       return UNLANG_ACTION_YIELD;
+}
+
+/** Cancel an SQL query submitted on a trunk
+ */
+static void sql_trunk_query_cancel(UNUSED request_t *request, UNUSED fr_signal_t action, void *uctx)
+{
+       fr_sql_query_t  *query_ctx = talloc_get_type_abort(uctx, fr_sql_query_t);
+
+       if (!query_ctx->treq) return;
+
+       /*
+        *      The query_ctx needs to be parented by the treq so that it still exists
+        *      when the cancel_mux callback is run.
+        */
+       talloc_steal(query_ctx->treq, query_ctx);
+
+       fr_trunk_request_signal_cancel(query_ctx->treq);
+
+       query_ctx->treq = NULL;
+}
+
+/** Submit an SQL query using a trunk connection.
+ *
+ * @param p_result     Result of current module call.
+ * @param priority     Unused.
+ * @param request      Current request.
+ * @param uctx         query context containing query to execute.
+ * @return an unlang_action_t.
+ */
+unlang_action_t rlm_sql_trunk_query(rlm_rcode_t *p_result, UNUSED int *priority, request_t *request, void *uctx)
+{
+       fr_sql_query_t  *query_ctx = talloc_get_type_abort(uctx, fr_sql_query_t);
+       fr_trunk_enqueue_t      status;
+
+       fr_assert(query_ctx->trunk);
+
+       /* There's no query to run, return an error */
+       if (query_ctx->query_str[0] == '\0') {
+               if (request) REDEBUG("Zero length query");
+               RETURN_MODULE_INVALID;
+       }
+
+       /*
+        *      If the query already has a treq, and that is not in the "init" state
+        *      then this is part of an ongoing transaction and needs requeueing
+        *      to submit on the same connection.
+        */
+       if (query_ctx->treq && query_ctx->treq->state != FR_TRUNK_REQUEST_STATE_INIT) {
+               status = fr_trunk_request_requeue(query_ctx->treq);
+       } else {
+               status = fr_trunk_request_enqueue(&query_ctx->treq, query_ctx->trunk, request, query_ctx, NULL);
+       }
+       switch (status) {
+       case FR_TRUNK_ENQUEUE_OK:
+       case FR_TRUNK_ENQUEUE_IN_BACKLOG:
+               if (unlang_function_push(request, sql_trunk_query_start,
+                                        query_ctx->type == SQL_QUERY_SELECT ?
+                                               query_ctx->inst->driver->sql_select_query_resume :
+                                               query_ctx->inst->driver->sql_query_resume,
+                                        sql_trunk_query_cancel, ~FR_SIGNAL_CANCEL,
+                                        UNLANG_SUB_FRAME, query_ctx) < 0) RETURN_MODULE_FAIL;
+               *p_result = RLM_MODULE_OK;
+               return UNLANG_ACTION_PUSHED_CHILD;
+
+       default:
+               REDEBUG("Unable to enqueue SQL query");
+               query_ctx->status = SQL_QUERY_FAILED;
+               query_ctx->rcode = RLM_SQL_ERROR;
+               RETURN_MODULE_FAIL;
+       }
+}
+
 /** Call the driver's sql_select_query method, reconnecting if necessary.
  *
  * @note Caller must call ``(inst->driver->sql_finish_select_query)(handle, &inst->config);``