Implementation of variable max_allowed_packet

This variable limits the amount of data that the client can send to the backend.
Note that ProxySQL won't really limit the amount of data that the client can send to the proxy, but the amount of data that the prox can send to the backend.

This commits also fixes issue #526

Closes #526

include/MySQL_Thread.h

@@ -286,6 +286,7 @@ class MySQL_Threads_Handler
 		bool client_found_rows;
 		bool multiplexing;
 		bool enforce_autocommit_on_reads;
+		int max_allowed_packet;
 		int max_transaction_time;
 		int threshold_query_length;
 		int threshold_resultset_size;

include/proxysql_structs.h

@@ -3,6 +3,9 @@
 #define PKT_PARSED 0
 #define PKT_ERROR 1
 
+#ifdef max_allowed_packet
+#undef max_allowed_packet
+#endif
 
 
 
@@ -697,6 +700,7 @@ GOptionEntry cmd_option_entries[] =
 MySQL_HostGroups_Manager *MyHGM;
 __thread char *mysql_thread___default_schema;
 __thread char *mysql_thread___server_version;
+__thread int mysql_thread___max_allowed_packet;
 __thread int mysql_thread___max_transaction_time;
 __thread int mysql_thread___threshold_query_length;
 __thread int mysql_thread___threshold_resultset_size;
@@ -772,6 +776,7 @@ extern MySQL_HostGroups_Manager *MyHGM;
 //extern GOptionEntry cmd_option_entries[];
 extern __thread char *mysql_thread___default_schema;
 extern __thread char *mysql_thread___server_version;
+extern __thread int mysql_thread___max_allowed_packet;
 extern __thread int mysql_thread___max_transaction_time;
 extern __thread int mysql_thread___threshold_query_length;
 extern __thread int mysql_thread___threshold_resultset_size;

lib/MySQL_Session.cpp

@@ -1146,6 +1146,9 @@ handler_again:
 									}
 									return -1;
 									break;
+								case 1153: // ER_NET_PACKET_TOO_LARGE
+									proxy_warning("Error ER_NET_PACKET_TOO_LARGE during query on (%d,%s,%d): %d, %s\n", myconn->parent->myhgc->hid, myconn->parent->address, myconn->parent->port, myerr, mysql_error(myconn->mysql));
+									break;
 								default:
 									break; // continue normally
 							}
@@ -1863,9 +1866,19 @@ bool MySQL_Session::handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_C
 		CurrentQuery.begin((unsigned char *)pkt->ptr,pkt->size,true);
 		delete qpo->new_query;
 	}
+
+	if (pkt->size > (unsigned int) mysql_thread___max_allowed_packet) {
+		// ER_NET_PACKET_TOO_LARGE
+		client_myds->DSS=STATE_QUERY_SENT_NET;
+		client_myds->myprot.generate_pkt_ERR(true,NULL,NULL,client_myds->pkt_sid+1,1153,(char *)"08S01",(char *)"Got a packet bigger than 'max_allowed_packet' bytes");
+		l_free(pkt->size,pkt->ptr);
+		RequestEnd(NULL);
+		return true;
+	}
+
 	if (qpo->error_msg) {
 		client_myds->DSS=STATE_QUERY_SENT_NET;
-		client_myds->myprot.generate_pkt_ERR(true,NULL,NULL,1,1148,(char *)"#42000",qpo->error_msg);
+		client_myds->myprot.generate_pkt_ERR(true,NULL,NULL,client_myds->pkt_sid+1,1148,(char *)"42000",qpo->error_msg);
 //		client_myds->DSS=STATE_SLEEP;
 //		status=WAITING_CLIENT_DATA;
 		l_free(pkt->size,pkt->ptr);

lib/MySQL_Thread.cpp

@@ -180,6 +180,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"monitor_query_timeout",
 	(char *)"monitor_timer_cached",
 	(char *)"monitor_writer_is_also_reader",
+	(char *)"max_allowed_packet",
 	(char *)"max_transaction_time",
 	(char *)"multiplexing",
 	(char *)"enforce_autocommit_on_reads",
@@ -257,6 +258,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.monitor_writer_is_also_reader=true;
+	variables.max_allowed_packet=4*1024*1024;
 	variables.max_transaction_time=4*3600*1000;
 	variables.threshold_query_length=512*1024;
 	variables.threshold_resultset_size=4*1024*1024;
@@ -455,6 +457,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,"eventslog_filesize")) return (int)variables.eventslog_filesize;
+	if (!strcasecmp(name,"max_allowed_packet")) return (int)variables.max_allowed_packet;
 	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,"threshold_resultset_size")) return (int)variables.threshold_resultset_size;
@@ -635,6 +638,10 @@ char * MySQL_Threads_Handler::get_variable(char *name) {	// this is the public f
 		sprintf(intbuf,"%d",variables.eventslog_filesize);
 		return strdup(intbuf);
 	}
+	if (!strcasecmp(name,"max_allowed_packet")) {
+		sprintf(intbuf,"%d",variables.max_allowed_packet);
+		return strdup(intbuf);
+	}
 	if (!strcasecmp(name,"max_transaction_time")) {
 		sprintf(intbuf,"%d",variables.max_transaction_time);
 		return strdup(intbuf);
@@ -913,6 +920,15 @@ bool MySQL_Threads_Handler::set_variable(char *name, char *value) {	// this is t
 			return false;
 		}
 	}
+	if (!strcasecmp(name,"max_allowed_packet")) {
+		int intv=atoi(value);
+		if (intv >= 8192 && intv <= 1024*1024*1024) {
+			variables.max_allowed_packet=intv;
+			return true;
+		} else {
+			return false;
+		}
+	}
 	if (!strcasecmp(name,"max_transaction_time")) {
 		int intv=atoi(value);
 		if (intv >= 1000 && intv <= 20*24*3600*1000) {
@@ -1959,6 +1975,7 @@ void MySQL_Thread::process_all_sessions() {
 void MySQL_Thread::refresh_variables() {
 	GloMTH->wrlock();
 	__thread_MySQL_Thread_Variables_version=__global_MySQL_Thread_Variables_version;
+	mysql_thread___max_allowed_packet=GloMTH->get_variable_int((char *)"max_allowed_packet");
 	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___threshold_resultset_size=GloMTH->get_variable_int((char *)"threshold_resultset_size");