Add autocommit option, and connection is released to pool after client closed with 'autocommit=true'

plugins/proxy/proxy-sql-log.c

@@ -546,7 +546,8 @@ log_sql_client(sql_log_t *sql_log, network_mysqld_con *con)
 void
 log_sql_backend(sql_log_t *sql_log, network_mysqld_con *con, injection *inj)
 {
-    gfloat latency_ms = 0.0;
+    gfloat latency_ms = 0.0, proc_query_ms = 0.0;
+    gfloat db_process_ms = 0.0, proc_result_ms = 0.0;
     GString *message = NULL;
     GString *begin_time = NULL;
 
@@ -559,14 +560,18 @@ log_sql_backend(sql_log_t *sql_log, network_mysqld_con *con, injection *inj)
     latency_ms = (inj->ts_read_query_result_last - inj->ts_read_query)/1000.0;
     if ((gint)latency_ms < sql_log->sql_log_slow_ms) return;
 
+    proc_query_ms = (inj->ts_after_send_query - inj->ts_read_query)/1000.0;
+    db_process_ms = (inj->ts_read_query_result_first - inj->ts_after_send_query)/1000.0;
+    proc_result_ms = (inj->ts_read_query_result_last - inj->ts_read_query_result_first)/1000.0;
+
     message = g_string_sized_new(sizeof("2004-01-01T00:00:00.000Z"));
 
     begin_time = chassis_log_microsecond_tostring(con->conn_status_var.cur_query_start_time,
                                             CHASSIS_RESOLUTION_US);
     /* get current time */
     chassis_log_update_timestamp(message, CHASSIS_RESOLUTION_US);
     g_string_append_printf(message, ": C_begin:%s C:%s C_db:%s C_usr:%s S:%s(thread_id:%u) S_db:%s "
-                                    "S_usr:%s inj(type:%d bytes:%lu rows:%lu) %.3f(ms) %s %s:%s\n",
+                                    "S_usr:%s inj(type:%d bytes:%lu rows:%lu) event_thd:%d query_count:%d %.3f(ms) %.3f(ms) %.3f(ms) %.3f(ms) %s %s:%s\n",
                                 begin_time->str,
                                 NETWORK_SOCKET_SRC_NAME(con->client),
                                 NETWORK_SOCKET_DB_NAME(con->client),
@@ -575,7 +580,8 @@ log_sql_backend(sql_log_t *sql_log, network_mysqld_con *con, injection *inj)
                                 NETWORK_SOCKET_THREADID(con->server),
                                 NETWORK_SOCKET_DB_NAME(con->server),
                                 NETWORK_SOCKET_USR_NAME(con->server),
-                                inj->id, inj->bytes, inj->rows,
+                                inj->id, inj->bytes, inj->rows, cur_thid,
+                                con->conn_status_var.query_count, proc_query_ms, db_process_ms, proc_result_ms,
                                 latency_ms, inj->qstat.query_status == MYSQLD_PACKET_OK ? "OK" : "ERR",
                                 GET_COM_STRING(inj->query));
     g_string_free(begin_time, TRUE);

src/chassis-mainloop.h

@@ -105,6 +105,8 @@ struct chassis {
 
     MYSQL_VERSION my_version;
 
+    network_socket_autocommit_t autocommit;
+
     GPtrArray *threads;
 
     chassis_shutdown_hooks_t *shutdown_hooks;

src/chassis-options-utils.c

@@ -458,6 +458,41 @@ show_instance(void *ex_param)
     return g_strdup(srv->instance_name);
 }
 
+gchar *
+show_autocommit(void *ex_param)
+{
+    gchar *commit = NULL;
+    external_param *opt_param = (external_param *)ex_param;
+    chassis *srv = opt_param->chas;
+
+    if (srv->autocommit == AUTOCOMMIT_UNKNOWN)
+        commit = "default";
+    else if (srv->autocommit == AUTOCOMMIT_FALSE)
+        commit = "false";
+    else if (srv->autocommit == AUTOCOMMIT_TRUE)
+        commit = "true";
+
+    return g_strdup(commit);
+}
+
+gint
+assign_autocommit(const char *newval, void *ex_param) {
+    gint ret = 0;
+    external_param *opt_param = (external_param *)ex_param;
+    chassis *srv = opt_param->chas;
+
+    if (strcasecmp(newval, "true") == 0) {
+        srv->autocommit = AUTOCOMMIT_TRUE;
+    } else if (strcasecmp(newval, "false") == 0) {
+        srv->autocommit = AUTOCOMMIT_FALSE;
+    } else if (strcasecmp(newval, "default") == 0) {
+        srv->autocommit = AUTOCOMMIT_UNKNOWN;
+    } else {
+        ret = 1;
+    }
+    return ret;
+}
+
 gint
 assign_wait_timeout(const char *newval, void *ex_param)
 {

src/chassis-options-utils.h

@@ -65,6 +65,7 @@ CHASSIS_API gint assign_remove_backend_timeout(const char *newval, void *ex_para
 CHASSIS_API gint assign_log_trace_modules(const char *newval, void *ex_param);
 CHASSIS_API gint assign_backend_monitor_pwds(const char *newval, void *ex_param);
 CHASSIS_API gint assign_db_connect_timeout(const char *newval, void *ex_param);
+CHASSIS_API gint assign_autocommit(const char *newval, void *ex_param);
 
 /* show utils */
 CHASSIS_API gchar* show_verbose_shutdown(void *external_param);
@@ -85,6 +86,7 @@ CHASSIS_API gchar* show_event_threads(void *external_param);
 CHASSIS_API gchar* show_lua_path(void *external_param);
 CHASSIS_API gchar* show_user(void *external_param);
 CHASSIS_API gchar* show_instance(void *external_param);
+CHASSIS_API gchar* show_autocommit(void *external_param);
 CHASSIS_API gchar* show_wait_timeout(void *external_param);
 CHASSIS_API gchar* show_shutdown_timeout(void *external_param);
 CHASSIS_API gchar* show_db_connection_idle_timeout(void *external_param);

src/mysql-proxy-cli.c

@@ -183,6 +183,8 @@ typedef struct {
 
     gchar *instance_name;
 
+    gchar *autocommit;
+
     gint wait_timeout;
 
     gint shutdown_timeout;
@@ -287,6 +289,7 @@ void chassis_frontend_free(chassis_frontend_t *frontend) {
     if (frontend->sql_log) g_free(frontend->sql_log);
     if (frontend->sql_log_mode_str) g_free(frontend->sql_log_mode_str);
 
+    if (frontend->autocommit) g_free(frontend->autocommit);
     if (frontend->percentile_switch) g_free(frontend->percentile_switch);
 
     g_slice_free(chassis_frontend_t, frontend);
@@ -320,6 +323,7 @@ int chassis_frontend_set_chassis_options(chassis_frontend_t *frontend, chassis_o
     chassis_options_add(opts, "event-threads", 0, 0, G_OPTION_ARG_INT, &(frontend->event_thread_count), "number of event-handling threads (default: 1)", NULL, NULL, show_event_threads, SHOW_OPTS_PROPERTY);
     chassis_options_add(opts, "lua-path", 0, 0, G_OPTION_ARG_STRING, &(frontend->lua_path), "set the LUA_PATH", "<...>", NULL, NULL, 0);
     chassis_options_add(opts, "lua-cpath", 0, 0, G_OPTION_ARG_STRING, &(frontend->lua_cpath), "set the LUA_CPATH", "<...>", NULL, NULL, 0);
+    chassis_options_add(opts, "autocommit", 0, 0, G_OPTION_ARG_STRING, &(frontend->autocommit), "autocommit", NULL, assign_autocommit, show_autocommit, ALL_OPTS_PROPERTY);
     chassis_options_add(opts, "instance", 0, 0, G_OPTION_ARG_STRING, &(frontend->instance_name), "instance name", "<name>", NULL, show_instance, SHOW_OPTS_PROPERTY);
     chassis_options_add(opts, "wait-timeout", 0, 0, G_OPTION_ARG_INT, &(frontend->wait_timeout), "the number of seconds which dbproxy waits for activity on a connection before closing it (default:0)", NULL,
                                                     assign_wait_timeout, show_wait_timeout, ALL_OPTS_PROPERTY);
@@ -563,6 +567,18 @@ int main_cmdline(int argc, char **argv) {
         GOTO_EXIT(EXIT_FAILURE);
     }
 
+    srv->autocommit = AUTOCOMMIT_TRUE;
+    if (frontend->autocommit) {
+        if (!strcasecmp(frontend->autocommit, "default"))
+            srv->autocommit = AUTOCOMMIT_UNKNOWN;
+        else if (!strcasecmp(frontend->autocommit, "false"))
+            srv->autocommit = AUTOCOMMIT_FALSE;
+        else if (0 < strlen(frontend->autocommit) && strcasecmp(frontend->autocommit, "true")) {
+            g_log_dbproxy(g_critical, "--autocommit is assigned an invalid value: %s", frontend->autocommit);
+            GOTO_EXIT(EXIT_FAILURE);
+        }
+    }
+
     if (frontend->max_connections < 0) {
         g_log_dbproxy(g_critical, "--max-connections has to be >= 0, is %d", frontend->max_connections);
         GOTO_EXIT(EXIT_FAILURE);

src/network-injection.h

@@ -88,6 +88,7 @@ typedef struct {
     query_status qstat;                     /**< summary information about the query status */
 
     guint64 ts_read_query;                  /**< microsec timestamp when we added this query to the queues */
+    guint64 ts_after_send_query;
     guint64 ts_read_query_result_first;     /**< microsec timestamp when we received the first packet */
     guint64 ts_read_query_result_last;      /**< microsec timestamp when we received the last packet */
 

src/network-mysqld-stats.h

@@ -129,6 +129,7 @@ typedef struct connection_status_var_t
 {
     guint64 cur_query_type;
     guint64 cur_query_start_time;
+    gint32  query_count;
     gboolean query_running;
     gchar cur_query[STMT_LENTH];
     guint64 cur_query_com_type;

src/network-mysqld.c

@@ -1381,6 +1381,16 @@ void network_mysqld_con_handle(int event_fd, short events, void *user_data) {
              * the server connection is still fine, 
              * let's keep it open for reuse */
 
+           if (con->server && con->server->conn_attr.autocommit_status == AUTOCOMMIT_TRUE) {
+                gboolean release = ((con->state == CON_STATE_CLOSE_CLIENT) &&
+                          //!con->conn_status.is_in_select_calc_found_rows &&
+                          con->client->conn_attr.savepoint_flag == FALSE &&
+                          !con->conn_status.is_in_transaction &&
+                          (g_hash_table_size(con->locks) == 0));
+                if (release)
+                    network_connection_pool_lua_add_connection(con);
+            }
+
             plugin_call_cleanup(srv, con);
 
             chassis_event_remove_connection(srv, con);
@@ -1946,7 +1956,9 @@ void network_mysqld_con_handle(int event_fd, short events, void *user_data) {
                         g_atomic_int_add(&srv->proxy_aborted_clients, 1);
                         break;
                     }
-                }
+            }
+
+            con->conn_status_var.query_count = 0;
 
             /**
              * there should be 3 possible next states from here:
@@ -1985,7 +1997,7 @@ void network_mysqld_con_handle(int event_fd, short events, void *user_data) {
              *
              * this state will loop until all the packets from the send-queue are flushed 
              */
-
+            con->conn_status_var.query_count++;
             if (events != EV_TIMEOUT && con->server->send_queue->offset == 0) {
                 /* only parse the packets once */
                 network_packet packet;
@@ -2062,8 +2074,15 @@ void network_mysqld_con_handle(int event_fd, short events, void *user_data) {
                 if (con->server) network_mysqld_queue_reset(con->server);
                 break;
             default:
-                con->state = CON_STATE_READ_QUERY_RESULT;
-                break;
+                {
+                    network_mysqld_con_lua_t *st = con->plugin_con_state;
+                    if (st && 0 < st->injected.queries->length) {
+                        injection *inj = g_queue_peek_head(st->injected.queries);
+                        if (inj) inj->ts_after_send_query = chassis_get_rel_microseconds();
+                    }
+                    con->state = CON_STATE_READ_QUERY_RESULT;
+                    break;
+                }
             }
 
             if (TRACE_SQL(con->srv->log->log_trace_modules)) {
@@ -2450,6 +2469,8 @@ void network_mysqld_con_accept(int G_GNUC_UNUSED event_fd, short events, void *u
     client = network_socket_accept(listen_con->server);
     if (!client) return;
 
+    client->conn_attr.autocommit_status = listen_con->srv->autocommit;
+
     /* looks like we open a client connection */
     client_con = network_mysqld_con_new();
     client_con->client = client;