Added new var mysql-query_processor_iterations

When query rules are processed, even if flagOUT is set rules are processed
in numerical order.
This new variable allows to jump back (reset rule_id to 0) N amount of times.
Using this variable is possible to implement loops.
Futhermore, a loop is possible with only one rule setting flagIN=flagOUT .
The default is 0. The maximum value is 1000000 .

include/MySQL_Thread.h

@@ -299,6 +299,7 @@ class MySQL_Threads_Handler
 		int default_max_latency_ms;
 		int default_query_delay;
 		int default_query_timeout;
+		int query_processor_iterations;
 		int long_query_time;
 		char *init_connect;
 #ifdef DEBUG

include/proxysql_structs.h

@@ -723,6 +723,7 @@ __thread int mysql_thread___connect_retries_delay;
 __thread int mysql_thread___connection_max_age_ms;
 __thread int mysql_thread___connect_timeout_server;
 __thread int mysql_thread___connect_timeout_server_max;
+__thread int mysql_thread___query_processor_iterations;
 __thread uint16_t mysql_thread___server_capabilities;
 __thread uint8_t mysql_thread___default_charset;
 __thread int mysql_thread___poll_timeout;
@@ -801,6 +802,7 @@ extern __thread int mysql_thread___connect_retries_delay;
 extern __thread int mysql_thread___connection_max_age_ms;
 extern __thread int mysql_thread___connect_timeout_server;
 extern __thread int mysql_thread___connect_timeout_server_max;
+extern __thread int mysql_thread___query_processor_iterations;
 extern __thread uint16_t mysql_thread___server_capabilities;
 extern __thread uint8_t mysql_thread___default_charset;
 extern __thread int mysql_thread___poll_timeout;

lib/MySQL_Thread.cpp

@@ -190,6 +190,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"default_max_latency_ms",
 	(char *)"default_query_delay",
 	(char *)"default_query_timeout",
+	(char *)"query_processor_iterations",
 	(char *)"long_query_time",
 	(char *)"query_cache_size_MB",
 	(char *)"ping_interval_server_msec",
@@ -266,6 +267,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.default_max_latency_ms=1*1000; // by default, the maximum allowed latency for a host is 1000ms
 	variables.default_query_delay=0;
 	variables.default_query_timeout=24*3600*1000;
+	variables.query_processor_iterations=0;
 	variables.long_query_time=1000;
 	variables.query_cache_size_MB=256;
 	variables.init_connect=NULL;
@@ -471,6 +473,7 @@ int MySQL_Threads_Handler::get_variable_int(char *name) {
 	if (!strcasecmp(name,"max_connections")) return (int)variables.max_connections;
 	if (!strcasecmp(name,"default_query_delay")) return (int)variables.default_query_delay;
 	if (!strcasecmp(name,"default_query_timeout")) return (int)variables.default_query_timeout;
+	if (!strcasecmp(name,"query_processor_iterations")) return (int)variables.query_processor_iterations;
 	if (!strcasecmp(name,"default_max_latency_ms")) return (int)variables.default_max_latency_ms;
 	if (!strcasecmp(name,"long_query_time")) return (int)variables.long_query_time;
 	if (!strcasecmp(name,"query_cache_size_MB")) return (int)variables.query_cache_size_MB;
@@ -682,6 +685,10 @@ char * MySQL_Threads_Handler::get_variable(char *name) {	// this is the public f
 		sprintf(intbuf,"%d",variables.default_query_timeout);
 		return strdup(intbuf);
 	}
+	if (!strcasecmp(name,"query_processor_iterations")) {
+		sprintf(intbuf,"%d",variables.query_processor_iterations);
+		return strdup(intbuf);
+	}
 	if (!strcasecmp(name,"default_max_latency_ms")) {
 		sprintf(intbuf,"%d",variables.default_max_latency_ms);
 		return strdup(intbuf);
@@ -999,6 +1006,15 @@ bool MySQL_Threads_Handler::set_variable(char *name, char *value) {	// this is t
 			return false;
 		}
 	}
+	if (!strcasecmp(name,"query_processor_iterations")) {
+		int intv=atoi(value);
+		if (intv >= 0 && intv <= 1000*1000) {
+			variables.query_processor_iterations=intv;
+			return true;
+		} else {
+			return false;
+		}
+	}
 	if (!strcasecmp(name,"default_max_latency_ms")) {
 		int intv=atoi(value);
 		if (intv >= 0 && intv <= 20*24*3600*1000) {
@@ -2009,6 +2025,7 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___max_connections=GloMTH->get_variable_int((char *)"max_connections");
 	mysql_thread___default_query_delay=GloMTH->get_variable_int((char *)"default_query_delay");
 	mysql_thread___default_query_timeout=GloMTH->get_variable_int((char *)"default_query_timeout");
+	mysql_thread___query_processor_iterations=GloMTH->get_variable_int((char *)"query_processor_iterations");
 	mysql_thread___default_max_latency_ms=GloMTH->get_variable_int((char *)"default_max_latency_ms");
 	mysql_thread___long_query_time=GloMTH->get_variable_int((char *)"long_query_time");
 	mysql_thread___query_cache_size_MB=GloMTH->get_variable_int((char *)"query_cache_size_MB");

lib/Query_Processor.cpp

@@ -659,6 +659,7 @@ Query_Processor_Output * Query_Processor::process_mysql_query(MySQL_Session *ses
 	QP_rule_t *qr;
 	re2_t *re2p;
 	int flagIN=0;
+	int reiterate=mysql_thread___query_processor_iterations;
 	if (sess->mirror==true) {
 		// we are into a mirror session
 		// we immediately set a destination_hostgroup
@@ -673,6 +674,7 @@ Query_Processor_Output * Query_Processor::process_mysql_query(MySQL_Session *ses
 			goto __exit_process_mysql_query;
 		}
 	}
+__internal_loop:
 	for (std::vector<QP_rule_t *>::iterator it=_thr_SQP_rules->begin(); it!=_thr_SQP_rules->end(); ++it) {
 		qr=*it;
 		if (qr->flagIN != flagIN) {
@@ -753,10 +755,11 @@ Query_Processor_Output * Query_Processor::process_mysql_query(MySQL_Session *ses
 
 		// if we arrived here, we have a match
 		qr->hits++; // this is done without atomic function because it updates only the local variables
-
+		bool set_flagOUT=false;
 		if (qr->flagOUT >= 0) {
 			proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "query rule %d has changed flagOUT\n", qr->rule_id);
 			flagIN=qr->flagOUT;
+			set_flagOUT=true;
 			//sess->query_info.flagOUT=flagIN;
     }
     if (qr->reconnect >= 0) {
@@ -815,6 +818,12 @@ Query_Processor_Output * Query_Processor::process_mysql_query(MySQL_Session *ses
 			proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 5, "query rule %d is the last one to apply: exit!\n", qr->rule_id);
 			goto __exit_process_mysql_query;
 		}
+		if (set_flagOUT==true) {
+			if (reiterate) {
+				reiterate--;
+				goto __internal_loop;
+			}
+		}
 	}
 	
 __exit_process_mysql_query: