Merge pull request #4229 from sysown/v2.x-poll_optimization

Minor optimizations , fixed memory leaks and invalid reads

include/MySQL_Thread.h

@@ -57,17 +57,18 @@ class ProxySQL_Poll {
 	void expand(unsigned int more);
 
 	public:
-	unsigned int poll_timeout;
-	unsigned long loops;
-	StatCounters *loop_counters;
 	unsigned int len;
 	unsigned int size;
 	struct pollfd *fds;
 	MySQL_Data_Stream **myds;
 	unsigned long long *last_recv;
 	unsigned long long *last_sent;
+	std::atomic<bool> bootstrapping_listeners;
 	volatile int pending_listener_add;
 	volatile int pending_listener_del;
+	unsigned int poll_timeout;
+	unsigned long loops;
+	StatCounters *loop_counters;
 
 	ProxySQL_Poll();
 	~ProxySQL_Poll();

lib/MySQL_PreparedStatement.cpp

@@ -581,6 +581,10 @@ MySQL_STMT_Manager_v14::MySQL_STMT_Manager_v14() {
 }
 
 MySQL_STMT_Manager_v14::~MySQL_STMT_Manager_v14() {
+	for (auto it = map_stmt_id_to_info.begin(); it != map_stmt_id_to_info.end(); ++it) {
+		MySQL_STMT_Global_info * a = it->second;
+		delete a;
+	}
 }
 
 void MySQL_STMT_Manager_v14::ref_count_client(uint64_t _stmt_id ,int _v, bool lock) {

lib/MySQL_Protocol.cpp

@@ -2177,7 +2177,7 @@ __exit_do_auth:
 			}
 		}
 		proxy_debug(PROXY_DEBUG_MYSQL_PROTOCOL,1,"Handshake (%s auth) <user:\"%s\" pass:\"%s\" db:\"%s\" max_pkt:%u>, capabilities:%u char:%u, use_ssl:%s\n",
-			(capabilities & CLIENT_SECURE_CONNECTION ? "new" : "old"), user, tmp_pass, db, max_pkt, capabilities, charset, ((*myds)->encrypted ? "yes" : "no"));
+			(capabilities & CLIENT_SECURE_CONNECTION ? "new" : "old"), user, tmp_pass, db, (*myds)->myconn->options.max_allowed_pkt, capabilities, charset, ((*myds)->encrypted ? "yes" : "no"));
 		free(tmp_pass);
 	}
 #endif

lib/MySQL_Session.cpp

@@ -7488,7 +7488,9 @@ void MySQL_Session::RequestEnd(MySQL_Data_Stream *myds) {
 			// if a prepared statement is executed, LogQuery was already called
 			break;
 		default:
-			LogQuery(myds);
+			if (session_fast_forward==false) {
+				LogQuery(myds);
+			}
 			break;
 	}
 
@@ -7502,13 +7504,15 @@ void MySQL_Session::RequestEnd(MySQL_Data_Stream *myds) {
 		}
 		myds->free_mysql_real_query();
 	}
-	// reset status of the session
-	status=WAITING_CLIENT_DATA;
-	if (client_myds) {
-		// reset status of client data stream
-		client_myds->DSS=STATE_SLEEP;
-		// finalize the query
-		CurrentQuery.end();
+	if (session_fast_forward==false) {
+		// reset status of the session
+		status=WAITING_CLIENT_DATA;
+		if (client_myds) {
+			// reset status of client data stream
+			client_myds->DSS=STATE_SLEEP;
+			// finalize the query
+			CurrentQuery.end();
+		}
 	}
 	started_sending_data_to_client=false;
 	previous_hostgroup = current_hostgroup;

lib/MySQL_Thread.cpp

@@ -211,6 +211,7 @@ ProxySQL_Poll::ProxySQL_Poll() {
 	len=0;
 	pending_listener_add=0;
 	pending_listener_del=0;
+	bootstrapping_listeners = true;
 	size=MIN_POLL_LEN;
 	fds=(struct pollfd *)malloc(size*sizeof(struct pollfd));
 	myds=(MySQL_Data_Stream **)malloc(size*sizeof(MySQL_Data_Stream *));
@@ -3198,18 +3199,25 @@ __run_skip_1a:
 #endif // IDLE_THREADS
 
 		pthread_mutex_unlock(&thread_mutex);
-		while ( // spin here if ...
-			(n=__sync_add_and_fetch(&mypolls.pending_listener_add,0)) // there is a new listener to add
-			||
-			(GloMTH->bootstrapping_listeners == true) // MySQL_Thread_Handlers has more listeners to configure
-		) {
-			if (n) {
-				poll_listener_add(n);
-				assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_add,n,0));
-			}
+		if (unlikely(mypolls.bootstrapping_listeners == true)) {
+			while ( // spin here if ...
+				(n=__sync_add_and_fetch(&mypolls.pending_listener_add,0)) // there is a new listener to add
+				||
+				(GloMTH->bootstrapping_listeners == true) // MySQL_Thread_Handlers has more listeners to configure
+			) {
+				if (n) {
+					poll_listener_add(n);
+					assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_add,n,0));
+				} else {
+					if (GloMTH->bootstrapping_listeners == false) {
+						// we stop looping
+						mypolls.bootstrapping_listeners = false;
+					}
+				}
 #ifdef DEBUG
-			usleep(5+rand()%10);
+				usleep(5+rand()%10);
 #endif
+			}
 		}
 
 		proxy_debug(PROXY_DEBUG_NET, 7, "poll_timeout=%u\n", mypolls.poll_timeout);
@@ -3243,17 +3251,19 @@ __run_skip_1a:
 		}
 #endif // IDLE_THREADS
 
-		while ((n=__sync_add_and_fetch(&mypolls.pending_listener_del,0))) {	// spin here
-			if (static_cast<int>(n) == -1) {
-				for (unsigned int i = 0; i < mypolls.len; i++) {
-					if (mypolls.myds[i] && mypolls.myds[i]->myds_type == MYDS_LISTENER) {
-						poll_listener_del(mypolls.myds[i]->fd);
+		if (unlikely(maintenance_loop == true)) {
+			while ((n=__sync_add_and_fetch(&mypolls.pending_listener_del,0))) {	// spin here
+				if (static_cast<int>(n) == -1) {
+					for (unsigned int i = 0; i < mypolls.len; i++) {
+						if (mypolls.myds[i] && mypolls.myds[i]->myds_type == MYDS_LISTENER) {
+							poll_listener_del(mypolls.myds[i]->fd);
+						}
 					}
+				} else {
+					poll_listener_del(n);
 				}
-			} else {
-				poll_listener_del(n);
+				assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_del,n,0));
 			}
-			assert(__sync_bool_compare_and_swap(&mypolls.pending_listener_del,n,0));
 		}
 
 		pthread_mutex_lock(&thread_mutex);

lib/debug.cpp

@@ -226,7 +226,8 @@ void proxy_debug_func(enum debug_module module, int verbosity, int thr, const ch
 			rc=(*proxy_sqlite3_bind_text)(statement1, 11, longdebugbuff2, -1, SQLITE_TRANSIENT); ASSERT_SQLITE_OK(rc, db);
 			SAFE_SQLITE3_STEP2(statement1);
 			rc=(*proxy_sqlite3_clear_bindings)(statement1); ASSERT_SQLITE_OK(rc, db);
-			rc=(*proxy_sqlite3_reset)(statement1); ASSERT_SQLITE_OK(rc, db);
+			// Note: no assert() in proxy_debug_func() after sqlite3_reset() because it is possible that we are in shutdown
+			rc=(*proxy_sqlite3_reset)(statement1); // ASSERT_SQLITE_OK(rc, db);
 		}
 	}
 	pthread_mutex_unlock(&debug_mutex);

src/SQLite3_Server.cpp

@@ -1015,21 +1015,22 @@ static void *child_mysql(void *arg) {
 			}
 		}
 		sess->to_process=1;
-		// Get and set the client address before the sesion is processed.
-		union {
-			struct sockaddr_in in;
-			struct sockaddr_in6 in6;
-		} custom_sockaddr;
-		struct sockaddr *addr=(struct sockaddr *)malloc(sizeof(custom_sockaddr));
-		socklen_t addrlen=sizeof(custom_sockaddr);
-		memset(addr, 0, sizeof(custom_sockaddr));
-		sess->client_myds->client_addrlen=addrlen;
-		sess->client_myds->client_addr=addr;
-		int g_rc = getpeername(sess->client_myds->fd, addr, &addrlen);
-		if (g_rc == -1) {
-			proxy_error("'getpeername' failed with error: %d\n", g_rc);
+		if (sess->client_myds->client_addr == NULL) {
+			// Get and set the client address before the sesion is processed.
+			union {
+				struct sockaddr_in in;
+				struct sockaddr_in6 in6;
+			} custom_sockaddr;
+			struct sockaddr *addr=(struct sockaddr *)malloc(sizeof(custom_sockaddr));
+			socklen_t addrlen=sizeof(custom_sockaddr);
+			memset(addr, 0, sizeof(custom_sockaddr));
+			sess->client_myds->client_addrlen=addrlen;
+			sess->client_myds->client_addr=addr;
+			int g_rc = getpeername(sess->client_myds->fd, addr, &addrlen);
+			if (g_rc == -1) {
+				proxy_error("'getpeername' failed with error: %d\n", g_rc);
+			}
 		}
-
 		int rc=sess->handler();
 		if (rc==-1) goto __exit_child_mysql;
 	}