Handling INIT_DB on backends

include/mysql_connection.h

@@ -104,6 +104,7 @@ class MySQL_Connection {
 
 	int async_connect(short event);
 	int async_change_user(short event);
+	int async_select_db(short event);
 	int async_query(short event, char *stmt, unsigned long length);
 	int async_ping(short event);
 	void async_free_result();

lib/MySQL_Session.cpp

@@ -571,6 +571,10 @@ handler_again:
 						previous_status.push(PROCESSING_QUERY);
 						NEXT_IMMEDIATE(CHANGING_USER_SERVER);
 					}
+					if (strcmp(client_myds->myconn->userinfo->schemaname,myds->myconn->userinfo->schemaname)) {
+						previous_status.push(PROCESSING_QUERY);
+						NEXT_IMMEDIATE(CHANGING_SCHEMA);
+					}
 				}
 				status=PROCESSING_QUERY;
 				mybe->server_myds->max_connect_time=0;
@@ -717,6 +721,62 @@ handler_again:
 			}
 			break;
 
+		case CHANGING_SCHEMA:
+			//fprintf(stderr,"CHANGING_SCHEMA\n");
+			assert(mybe->server_myds->myconn);
+			{
+				MySQL_Data_Stream *myds=mybe->server_myds;
+				MySQL_Connection *myconn=myds->myconn;
+				myds->DSS=STATE_MARIADB_QUERY;
+				enum session_status st=status;
+				if (myds->mypolls==NULL) {
+					thread->mypolls.add(POLLIN|POLLOUT, mybe->server_myds->fd, mybe->server_myds, thread->curtime);
+				}
+				int rc=myconn->async_select_db(myds->revents);
+				if (rc==0) {
+					myds->myconn->userinfo->set(client_myds->myconn->userinfo);
+					st=previous_status.top();
+					previous_status.pop();
+					NEXT_IMMEDIATE(st);
+				} else {
+					if (rc==-1) {
+						// the command failed
+						int myerr=mysql_errno(myconn->mysql);
+						if (myerr > 2000) {
+							bool retry_conn=false;
+							// client error, serious
+							proxy_error("Detected a broken connection during INIT_DB: %d, %s\n", myerr, mysql_error(myconn->mysql));
+							if ((myds->myconn->reusable==true) && ((myds->myprot.prot_status & SERVER_STATUS_IN_TRANS)==0)) {
+								retry_conn=true;
+							}
+							myds->destroy_MySQL_Connection_From_Pool();
+							myds->fd=0;
+							if (retry_conn) {
+								myds->DSS=STATE_NOT_INITIALIZED;
+								//previous_status.push(PROCESSING_QUERY);
+								NEXT_IMMEDIATE(CONNECTING_SERVER);
+							}
+							return -1;
+						} else {
+							proxy_warning("Error during INIT_DB: %d, %s\n", myerr, mysql_error(myconn->mysql));
+								// we won't go back to PROCESSING_QUERY
+							st=previous_status.top();
+							previous_status.pop();
+							char sqlstate[10];
+							sprintf(sqlstate,"#%s",mysql_sqlstate(myconn->mysql));
+							client_myds->myprot.generate_pkt_ERR(true,NULL,NULL,1,mysql_errno(myconn->mysql),sqlstate,mysql_error(myconn->mysql));
+								myds->destroy_MySQL_Connection_From_Pool();
+								myds->fd=0;
+							status=WAITING_CLIENT_DATA;
+							client_myds->DSS=STATE_SLEEP;
+						}
+					} else {
+						// rc==1 , nothing to do for now
+					}
+				}
+			}
+			break;
+
 		case CONNECTING_SERVER:
 			//fprintf(stderr,"CONNECTING_SERVER\n");
 			if (mybe->server_myds->max_connect_time) {

lib/MySQL_Thread.cpp

@@ -1662,6 +1662,12 @@ SQLite3_result * MySQL_Threads_Handler::SQL3_Processlist() {
 					case WAITING_CLIENT_DATA:
 						pta[9]=strdup("Sleep");
 						break;
+					case CHANGING_USER_SERVER:
+						pta[9]=strdup("Change user");
+						break;
+					case CHANGING_SCHEMA:
+						pta[9]=strdup("InitDB");
+						break;
 					default:
 						pta[9]=strdup("");
 						break;

lib/mysql_connection.cpp

@@ -280,7 +280,8 @@ void MySQL_Connection::ping_cont(short event) {
 
 void MySQL_Connection::initdb_start() {
 	PROXY_TRACE();
-	async_exit_status = mysql_select_db_start(&interr,mysql,userinfo->schemaname);
+	MySQL_Connection_userinfo *client_ui=myds->sess->client_myds->myconn->userinfo;
+	async_exit_status = mysql_select_db_start(&interr,mysql,client_ui->schemaname);
 }
 
 void MySQL_Connection::initdb_cont(short event) {
@@ -683,7 +684,7 @@ int MySQL_Connection::async_change_user(short event) {
 			handler(event);
 			break;
 	}
-	
+
 	// check again
 	switch (async_state_machine) {
 		case ASYNC_CHANGE_USER_SUCCESSFUL:
@@ -700,6 +701,41 @@ int MySQL_Connection::async_change_user(short event) {
 	return 1;
 }
 
+int MySQL_Connection::async_select_db(short event) {
+	PROXY_TRACE();
+	assert(mysql);
+	assert(ret_mysql);
+	switch (async_state_machine) {
+		case ASYNC_INITDB_SUCCESSFUL:
+			async_state_machine=ASYNC_IDLE;
+			return 0;
+			break;
+		case ASYNC_INITDB_FAILED:
+			return -1;
+			break;
+		case ASYNC_IDLE:
+			async_state_machine=ASYNC_INITDB_START;
+		default:
+			handler(event);
+			break;
+	}
+
+	// check again
+	switch (async_state_machine) {
+		case ASYNC_INITDB_SUCCESSFUL:
+			async_state_machine=ASYNC_IDLE;
+			return 0;
+			break;
+		case ASYNC_INITDB_FAILED:
+			return -1;
+			break;
+		default:
+			return 1;
+			break;
+	}
+	return 1;
+}
+
 void MySQL_Connection::async_free_result() {
 	PROXY_TRACE();
 	assert(mysql);