Merge PR#5776 PR#5784

include/Base_Thread.h

@@ -65,7 +65,7 @@ class Base_Thread {
 	template<typename T>
 	void check_for_invalid_fd(unsigned int n);
 	template<typename S>
-	void ProcessAllSessions_SortingSessions();
+	void ProcessAllSessions_Partition();
 	template<typename T>
 	void ProcessAllMyDS_AfterPoll();
 	template<typename T>

lib/Base_Thread.cpp

@@ -11,8 +11,8 @@
 // Explicitly instantiate the required template class and member functions
 template MySQL_Session* Base_Thread::create_new_session_and_client_data_stream<MySQL_Thread, MySQL_Session*>(int);
 template PgSQL_Session* Base_Thread::create_new_session_and_client_data_stream<PgSQL_Thread, PgSQL_Session*>(int);
-template void Base_Thread::ProcessAllSessions_SortingSessions<MySQL_Session>();
-template void Base_Thread::ProcessAllSessions_SortingSessions<PgSQL_Session>();
+template void Base_Thread::ProcessAllSessions_Partition<MySQL_Session>();
+template void Base_Thread::ProcessAllSessions_Partition<PgSQL_Session>();
 template void Base_Thread::ProcessAllMyDS_AfterPoll<MySQL_Thread>();
 template void Base_Thread::ProcessAllMyDS_AfterPoll<PgSQL_Thread>();
 template void Base_Thread::ProcessAllMyDS_BeforePoll<MySQL_Thread>();
@@ -230,33 +230,97 @@
 	}
 }
 
-// this function was inline in  MySQL_Thread::process_all_sessions()
+
 /**
- * @brief Sort all sessions based on maximum connection time.
- * 
- * This function iterates through all MySQL sessions and sorts them based on their maximum connection time.
- * Sessions with a valid maximum connection time are compared, and if one session has a greater maximum connection
- * time than another, their positions in the session list are swapped. The sorting is performed in-place.
- * 
- * @note This function assumes that MySQL sessions and their associated data structures have been initialized
- * and are accessible within the MySQL Thread.
+ * @brief Partition all sessions into three blocks, then FIFO-bump
+ * the long-waiters within block B.
+ *
+ * Block layout produced in mysql_sessions->pdata:
+ *   [0, running_end)         block A - running a query against the backend
+ *                            (myconn != NULL, mct == 0, status != WAITING_CLIENT_DATA)
+ *   [running_end, idle_begin) block B - acquiring/awaiting a backend
+ *                            (mct != 0)
+ *   [idle_begin, len)        block C - idle, or holds-conn-but-WAITING_CLIENT_DATA
+ *
+ * Block A drives the backend and may release its conn at end-of-query, giving
+ * block B sessions a fairness chance to acquire it. Sessions parked in
+ * WAITING_CLIENT_DATA (idle in a transaction after BEGIN) hold the conn but
+ * cannot release it until the client sends the next packet, so they live in C.
+ *
+ * Classification tests max_connect_time first: it must win over A even when
+ * myconn != NULL, to catch CHANGING_USER_SERVER on pooled connections and the
+ * post-error retry path where the old conn hasn't been destroyed yet.
+ *
+ * Loop 1 (partition + stats): single O(n) pass, in place. idx walks up,
+ * idle_begin walks down, they meet and terminate. While partitioning, we
+ * also accumulate sum_wait and b_count over block B, where
+ *   wt = curtime - CurrentQuery.start_time  (microseconds)
+ *
+ * Loop 2 (O(b_count)): promote sessions in B with
+ *   wt > K_WAIT * avg(wt)
+ * to the front of the B band, preserving their relative pdata order. This
+ * ensures the longer-waiting sessions in B get a chance at the next released
+ * backend conn before the newer arrivals do.
+ *
  */
 template<typename S>
-void Base_Thread::ProcessAllSessions_SortingSessions() {
-	unsigned int a=0;
-	for (unsigned int n=0; n<mysql_sessions->len; n++) {
-		S *sess=(S *)mysql_sessions->index(n);
-		if (sess->mybe && sess->mybe->server_myds) {
-			if (sess->mybe->server_myds->max_connect_time) {
-				S *sess2=(S *)mysql_sessions->index(a);
-				if (sess2->mybe && sess2->mybe->server_myds && sess2->mybe->server_myds->max_connect_time && sess2->mybe->server_myds->max_connect_time <= sess->mybe->server_myds->max_connect_time) {
-					// do nothing
-				} else {
-					void *p=mysql_sessions->pdata[a];
-					mysql_sessions->pdata[a]=mysql_sessions->pdata[n];
-					mysql_sessions->pdata[n]=p;
-					a++;
+void Base_Thread::ProcessAllSessions_Partition() {
+	unsigned long long sum_wait = 0;
+	size_t b_count = 0;
+	size_t running_end = 0;
+	size_t idle_begin = mysql_sessions->len;
+	size_t idx = 0;
+
+	while (idx < idle_begin) {
+		S* s = static_cast<S*>(mysql_sessions->index(idx));
+
+		const bool has_be = (s->mybe && s->mybe->server_myds);
+		const bool is_B = has_be && (s->mybe->server_myds->max_connect_time != 0);
+		const bool is_A = !is_B && has_be && (s->mybe->server_myds->myconn != nullptr) && (s->status != WAITING_CLIENT_DATA);
+
+		if (is_A) {
+			if (idx != running_end) {
+				void* p = mysql_sessions->pdata[idx];
+				mysql_sessions->pdata[idx] = mysql_sessions->pdata[running_end];
+				mysql_sessions->pdata[running_end] = p;
+			}
+			++running_end;
+			++idx;
+		} else if (is_B) {
+			const unsigned long long st = s->CurrentQuery.start_time;
+			const unsigned long long wt = (st && curtime > st) ? (curtime - st) : 0;
+			sum_wait += wt;
+			++b_count;
+			++idx;
+		} else {
+			--idle_begin;
+			if (idx != idle_begin) {
+				void* p = mysql_sessions->pdata[idx];
+				mysql_sessions->pdata[idx] = mysql_sessions->pdata[idle_begin];
+				mysql_sessions->pdata[idle_begin] = p;
+			}
+			// do NOT advance idx - re-examine the swapped-in element test
+		}
+	}
+
+	// promote sessions with wait_time > K * avg(wait_time)
+	// to the front of B
+	constexpr double K_WAIT = 1.1;
+	if (b_count > 1) {
+		const unsigned long long avg_wait = sum_wait / b_count;
+		const unsigned long long threshold = static_cast<unsigned long long>(static_cast<double>(avg_wait) * K_WAIT);
+		size_t a = running_end;
+		for (size_t n = running_end; n < idle_begin; ++n) {
+			S * sn = static_cast<S*>(mysql_sessions->pdata[n]);
+			const unsigned long long st = sn->CurrentQuery.start_time;
+			const unsigned long long wt = (st && curtime > st) ? (curtime - st) : 0;
+			if (wt > threshold) {
+				if (a != n) {
+					void* p = mysql_sessions->pdata[a];
+					mysql_sessions->pdata[a] = mysql_sessions->pdata[n];
+					mysql_sessions->pdata[n] = p;
 				}
+				++a;
 			}
 		}
 	}
@@ -347,10 +411,23 @@
 template<typename T, typename DS>
 void Base_Thread::tune_timeout_for_session_needs_pause(DS * myds) {
 	T* thr = static_cast<T*>(this);
-	if (thr->mypolls.poll_timeout==0 || (myds->sess->pause_until - curtime < thr->mypolls.poll_timeout) ) {
-		thr->mypolls.poll_timeout= myds->sess->pause_until - curtime;
-		proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION, 7, "Session=%p , poll_timeout=%u , pause_until=%llu , curtime=%llu\n", myds->sess, thr->mypolls.poll_timeout, myds->sess->pause_until, curtime);
-	}
+
+	if (myds->sess->pause_until > curtime) {
+		// Future pause: align poll_timeout to the pause expiration.
+		if (thr->mypolls.poll_timeout == 0 || (myds->sess->pause_until - curtime < thr->mypolls.poll_timeout)) {
+			thr->mypolls.poll_timeout = myds->sess->pause_until - curtime;
+			proxy_debug(PROXY_DEBUG_MYSQL_CONNECTION, 7, "Session=%p , poll_timeout=%u , pause_until=%llu , curtime=%llu\n", myds->sess, thr->mypolls.poll_timeout, 
+				myds->sess->pause_until, curtime);
+		}
+	} 
+	/* Do we need to immediately wake up poll() because of an already expired pause?
+	else {
+		// pause_until > 0 (caller checked) but <= curtime: pause has already expired.
+		// Wake poll() immediately rather than computing (pause_until - curtime)
+		if (thr->mypolls.poll_timeout == 0 || thr->mypolls.poll_timeout > 1) {
+			thr->mypolls.poll_timeout = 1;
+		}
+	}*/
 }
 
 template<typename T, typename DS>

lib/MySQL_Session.cpp

@@ -2997,6 +2997,7 @@ bool MySQL_Session::handler_again___status_CONNECTING_SERVER(int *_rc) {
 
 			st=previous_status.top();
 			previous_status.pop();
+			pause_until = 0;
 			NEXT_IMMEDIATE_NEW(st);
 		}
 		assert(st==status);
@@ -3023,7 +3024,7 @@ bool MySQL_Session::handler_again___status_CONNECTING_SERVER(int *_rc) {
 				st=previous_status.top();
 				previous_status.pop();
 				myds->wait_until=0;
-
+				pause_until = 0;
 				if (handle_session_track_capabilities() == false) {
 					previous_status.push(st);
 					pause_until = thread->curtime + mysql_thread___connect_retries_delay * 1000;

lib/MySQL_Thread.cpp

@@ -4453,7 +4453,7 @@ void MySQL_Thread::process_all_sessions() {
 	}
 #endif // IDLE_THREADS
 	if (sess_sort && mysql_sessions->len > 3) {
-		ProcessAllSessions_SortingSessions<MySQL_Session>();
+		ProcessAllSessions_Partition<MySQL_Session>();
 	}
 	for (n=0; n<mysql_sessions->len; n++) {
 		MySQL_Session *sess=(MySQL_Session *)mysql_sessions->index(n);

lib/PgSQL_Session.cpp

@@ -1634,6 +1634,7 @@ bool PgSQL_Session::handler_again___status_CONNECTING_SERVER(int* _rc) {
 		if (mybe->server_myds->myconn->async_state_machine == ASYNC_IDLE) {
 			st = previous_status.top();
 			previous_status.pop();
+			pause_until = 0;
 			NEXT_IMMEDIATE_NEW(st);
 		}
 		assert(st == status);
@@ -1671,6 +1672,7 @@ bool PgSQL_Session::handler_again___status_CONNECTING_SERVER(int* _rc) {
 			st = previous_status.top();
 			previous_status.pop();
 			myds->wait_until = 0;
+			pause_until = 0;
 			if (session_fast_forward) {
 				// 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

lib/PgSQL_Thread.cpp

@@ -3833,7 +3833,7 @@ void PgSQL_Thread::process_all_sessions() {
 	}
 #endif // IDLE_THREADS
 	if (sess_sort && mysql_sessions->len > 3) {
-		ProcessAllSessions_SortingSessions<PgSQL_Session>();
+		ProcessAllSessions_Partition<PgSQL_Session>();
 	}
 	for (n = 0; n < mysql_sessions->len; n++) {
 		PgSQL_Session* sess = (PgSQL_Session*)mysql_sessions->index(n);