New implementation of fast forward , #528

Author: renecannao

include/MySQL_Data_Stream.h

@@ -192,7 +192,7 @@ class MySQL_Data_Stream
 		MyHGM->destroy_MyConn_from_pool(mc);
 	}
 	void free_mysql_real_query();	
-
+	void reinit_queues();
 //	void destroy_MySQL_Connection() {
 //		MySQL_Connection *mc=myconn;
 //		detach_connection();

lib/MySQL_Session.cpp

@@ -542,6 +542,7 @@ bool MySQL_Session::handler_special_queries(PtrSize_t *pkt) {
 	return false;
 }
 
+#define NEXT_IMMEDIATE(new_st) do { set_status(new_st); goto handler_again; } while (0)
 
 int MySQL_Session::handler() {
 	bool wrong_pass=false;
@@ -654,6 +655,13 @@ int MySQL_Session::handler() {
 							current_hostgroup=default_hostgroup;
 						}
 						proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION, 5, "Statuses: WAITING_CLIENT_DATA - STATE_SLEEP\n");
+						if (session_fast_forward==true) { // if it is fast forward
+							mybe=find_or_create_backend(current_hostgroup); // set a backend
+							mybe->server_myds->reinit_queues();             // reinitialize the queues in the myds . By default, they are not active
+							mybe->server_myds->PSarrayOUT->add(pkt.ptr, pkt.size); // move the first packet
+							previous_status.push(FAST_FORWARD); // next status will be FAST_FORWARD . Now we need a connection
+							NEXT_IMMEDIATE(CONNECTING_SERVER);  // we create a connection . next status will be FAST_FORWARD
+						}
 						//unsigned char c;
 						c=*((unsigned char *)pkt.ptr+sizeof(mysql_hdr));
 						switch ((enum_mysql_command)c) {
@@ -824,14 +832,21 @@ int MySQL_Session::handler() {
 	}
 
 
-#define NEXT_IMMEDIATE(new_st) do { set_status(new_st); goto handler_again; } while (0)
 
 handler_again:
 
 	switch (status) {
 		case FAST_FORWARD:
-			fprintf(stderr,"FAST_FORWARD\n");
-			// FIXME: to implement
+			if (mybe->server_myds->mypolls==NULL) {
+				// register the mysql_data_stream
+				thread->mypolls.add(POLLIN|POLLOUT, mybe->server_myds->fd, mybe->server_myds, thread->curtime);
+			}
+			// copy all packets from backend to frontend
+			for (unsigned int k=0; k < mybe->server_myds->PSarrayIN->len; k++) {
+				PtrSize_t pkt;
+				mybe->server_myds->PSarrayIN->remove_index(0,&pkt);
+				client_myds->PSarrayOUT->add(pkt.ptr, pkt.size);
+			}
 			break;
 		case CONNECTING_CLIENT:
 			//fprintf(stderr,"CONNECTING_CLIENT\n");
@@ -1421,6 +1436,11 @@ int MySQL_Session::handler() {
 						st=previous_status.top();
 						previous_status.pop();
 						myds->wait_until=0;
+						if (session_fast_forward==true) {
+							// we have a successful connection and session_fast_forward enabled
+							// set DSS=STATE_SLEEP or it will believe it have to use MARIADB client library
+							myds->DSS=STATE_SLEEP;
+						}
 						NEXT_IMMEDIATE(st);
 						break;
 					case -1:
@@ -1923,6 +1943,7 @@ void MySQL_Session::handler___client_DSS_QUERY_SENT___server_DSS_NOT_INITIALIZED
 		mybe->server_myds->DSS=STATE_READY;
 		if (session_fast_forward==true) {
 			status=FAST_FORWARD;
+			mybe->server_myds->myconn->reusable=false; // the connection cannot be usable anymore
 		}
 	}
 }

lib/MySQL_Thread.cpp

@@ -1673,6 +1673,14 @@ void MySQL_Thread::process_data_on_data_stream(MySQL_Data_Stream *myds, unsigned
 					if (myds->sess->client_myds==myds) {
 						proxy_debug(PROXY_DEBUG_NET,1, "Session=%p, DataStream=%p -- Deleting FD %d\n", myds->sess, myds, myds->fd);
 						myds->sess->set_unhealthy();
+					} else {
+						// if this is a backend with fast_forward, set unhealthy
+						// if this is a backend without fast_forward, do not set unhealthy: it will be handled by client library
+						if (myds->sess->session_fast_forward) { // if fast forward
+							if (myds->myds_type==MYDS_BACKEND) { // and backend
+								myds->sess->set_unhealthy(); // set unhealthy
+							}
+						}
 					}
 				}
 }

lib/mysql_data_stream.cpp

@@ -237,6 +237,13 @@ void MySQL_Data_Stream::init() {
 	}
 }
 
+void MySQL_Data_Stream::reinit_queues() {
+	if (queueIN.buffer==NULL)
+		queue_init(queueIN,QUEUE_T_DEFAULT_SIZE);
+	if (queueOUT.buffer==NULL)
+		queue_init(queueOUT,QUEUE_T_DEFAULT_SIZE);
+}
+
 // this function initializes a MySQL_Data_Stream with arguments
 void MySQL_Data_Stream::init(enum MySQL_DS_type _type, MySQL_Session *_sess, int _fd) {
 	myds_type=_type;