Implemented variable threshold_query_length , #366

include/MySQL_Thread.h

@@ -273,6 +273,7 @@ class MySQL_Threads_Handler
 		bool default_reconnect;
 		bool have_compress;
 		int max_transaction_time;
+		int threshold_query_length;
 		int wait_timeout;
 		int max_connections;
 		int default_query_delay;

include/mysql_connection.h

@@ -46,6 +46,7 @@ class MySQL_Connection {
 	MYSQL_ROW mysql_row;
 	bool async_fetch_row_start;
 	MySQL_ResultSet *MyRS;
+	unsigned long largest_query_length;
 	struct {
 		char *ptr;
 		unsigned long length;

include/proxysql_structs.h

@@ -671,6 +671,7 @@ MySQL_HostGroups_Manager *MyHGM;
 __thread char *mysql_thread___default_schema;
 __thread char *mysql_thread___server_version;
 __thread int mysql_thread___max_transaction_time;
+__thread int mysql_thread___threshold_query_length;
 __thread int mysql_thread___wait_timeout;
 __thread int mysql_thread___max_connections;
 __thread int mysql_thread___default_query_delay;
@@ -724,6 +725,7 @@ extern MySQL_HostGroups_Manager *MyHGM;
 extern __thread char *mysql_thread___default_schema;
 extern __thread char *mysql_thread___server_version;
 extern __thread int mysql_thread___max_transaction_time;
+extern __thread int mysql_thread___threshold_query_length;
 extern __thread int mysql_thread___wait_timeout;
 extern __thread int mysql_thread___max_connections;
 extern __thread int mysql_thread___default_query_delay;

lib/MySQL_HostGroups_Manager.cpp

@@ -468,6 +468,11 @@ void MySQL_HostGroups_Manager::push_MyConn_to_pool(MySQL_Connection *c) {
 	mysrvc=(MySrvC *)c->parent;
 	proxy_debug(PROXY_DEBUG_MYSQL_CONNPOOL, 7, "Returning MySQL_Connection %p, server %s:%d with status %d\n", c, mysrvc->address, mysrvc->port, mysrvc->status);
   mysrvc->ConnectionsUsed->remove(c);
+	if (c->largest_query_length > (unsigned int)mysql_thread___threshold_query_length) {
+		proxy_debug(PROXY_DEBUG_MYSQL_CONNPOOL, 7, "Destroying MySQL_Connection %p, server %s:%d with status %d . largest_query_length = %lu\n", c, mysrvc->address, mysrvc->port, mysrvc->status, c->largest_query_length);
+		delete c;
+		goto __exit_push_MyConn_to_pool;
+	}
 	if (mysrvc->status==MYSQL_SERVER_STATUS_ONLINE) {
 		if (c->async_state_machine==ASYNC_IDLE) {
 			mysrvc->ConnectionsFree->add(c);
@@ -479,6 +484,7 @@ void MySQL_HostGroups_Manager::push_MyConn_to_pool(MySQL_Connection *c) {
 		proxy_debug(PROXY_DEBUG_MYSQL_CONNPOOL, 7, "Destroying MySQL_Connection %p, server %s:%d with status %d\n", c, mysrvc->address, mysrvc->port, mysrvc->status);
 		delete c;
 	}
+__exit_push_MyConn_to_pool:
 	wrunlock();
 }
 

lib/MySQL_Thread.cpp

@@ -159,6 +159,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"monitor_query_timeout",
 	(char *)"monitor_timer_cached",
 	(char *)"max_transaction_time",
+	(char *)"threshold_query_length",
 	(char *)"wait_timeout",
 	(char *)"max_connections",
 	(char *)"default_query_delay",
@@ -221,6 +222,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.monitor_query_status=strdup((char *)"SELECT * FROM INFORMATION_SCHEMA.GLOBAL_STATUS");
 	variables.monitor_timer_cached=true;
 	variables.max_transaction_time=4*3600*1000;
+	variables.threshold_query_length=512*1024;
 	variables.wait_timeout=8*3600*1000;
 	variables.max_connections=10*1000;
 	variables.default_query_delay=0;
@@ -366,6 +368,7 @@ int MySQL_Threads_Handler::get_variable_int(char *name) {
 	if (!strcasecmp(name,"connect_timeout_server_max")) return (int)variables.connect_timeout_server_max;
 	if (!strcasecmp(name,"connect_retries_delay")) return (int)variables.connect_retries_delay;
 	if (!strcasecmp(name,"max_transaction_time")) return (int)variables.max_transaction_time;
+	if (!strcasecmp(name,"threshold_query_length")) return (int)variables.threshold_query_length;
 	if (!strcasecmp(name,"wait_timeout")) return (int)variables.wait_timeout;
 	if (!strcasecmp(name,"max_connections")) return (int)variables.max_connections;
 	if (!strcasecmp(name,"default_query_delay")) return (int)variables.default_query_delay;
@@ -484,6 +487,10 @@ char * MySQL_Threads_Handler::get_variable(char *name) {	// this is the public f
 		sprintf(intbuf,"%d",variables.max_transaction_time);
 		return strdup(intbuf);
 	}
+	if (!strcasecmp(name,"threshold_query_length")) {
+		sprintf(intbuf,"%d",variables.threshold_query_length);
+		return strdup(intbuf);
+	}
 	if (!strcasecmp(name,"wait_timeout")) {
 		sprintf(intbuf,"%d",variables.wait_timeout);
 		return strdup(intbuf);
@@ -708,6 +715,15 @@ bool MySQL_Threads_Handler::set_variable(char *name, char *value) {	// this is t
 			return false;
 		}
 	}
+	if (!strcasecmp(name,"threshold_query_length")) {
+		int intv=atoi(value);
+		if (intv >= 1024 && intv <= 1*1024*1024*1024) {
+			variables.threshold_query_length=intv;
+			return true;
+		} else {
+			return false;
+		}
+	}
 	if (!strcasecmp(name,"wait_timeout")) {
 		int intv=atoi(value);
 		if (intv >= 0 && intv <= 20*24*3600*1000) {
@@ -1589,6 +1605,7 @@ void MySQL_Thread::refresh_variables() {
 	GloMTH->wrlock();
 	__thread_MySQL_Thread_Variables_version=__global_MySQL_Thread_Variables_version;
 	mysql_thread___max_transaction_time=GloMTH->get_variable_int((char *)"max_transaction_time");
+	mysql_thread___threshold_query_length=GloMTH->get_variable_int((char *)"threshold_query_length");
 	mysql_thread___wait_timeout=GloMTH->get_variable_int((char *)"wait_timeout");
 	mysql_thread___max_connections=GloMTH->get_variable_int((char *)"max_connections");
 	mysql_thread___default_query_delay=GloMTH->get_variable_int((char *)"default_query_delay");

lib/mysql_connection.cpp

@@ -155,6 +155,7 @@ MySQL_Connection::MySQL_Connection() {
 	mysql_result=NULL;
 	query.ptr=NULL;
 	query.length=0;
+	largest_query_length=0;
 	MyRS=NULL;
 	proxy_debug(PROXY_DEBUG_MYSQL_CONNPOOL, 4, "Creating new MySQL_Connection %p\n", this);
 };
@@ -326,6 +327,9 @@ void MySQL_Connection::set_names_cont(short event) {
 void MySQL_Connection::set_query(char *stmt, unsigned long length) {
 	query.length=length;
 	query.ptr=stmt;
+	if (length > largest_query_length) {
+		largest_query_length=length;
+	}
 	//query.ptr=(char *)malloc(length);
 	//memcpy(query.ptr,stmt,length);
 }