From ca202ceba1ea08711c5b1803ddffdbfbb41fb9dd Mon Sep 17 00:00:00 2001 From: Rahim Kanji Date: Tue, 12 May 2026 13:26:01 +0500 Subject: [PATCH 1/3] Refactor: partition sessions by backend state instead of sorting Replace ProcessAllSessions_SortingSessions with ProcessAllSessions_Partition. The old function did an implicit one-pass swap-sort over mysql_sessions by max_connect_time; the new one explicitly partitions sessions into three contiguous blocks in pdata according to their backend state: [0, running_end) A: running a query against the backend [running_end, idle_begin) B: acquiring a backend (max_connect_time != 0) [idle_begin, len) C: idle (no backend, or holding a conn while parked in WAITING_CLIENT_DATA) --- include/Base_Thread.h | 2 +- lib/Base_Thread.cpp | 76 ++++++++++++++++++++++++++++--------------- lib/MySQL_Thread.cpp | 2 +- lib/PgSQL_Thread.cpp | 2 +- 4 files changed, 53 insertions(+), 29 deletions(-) diff --git a/include/Base_Thread.h b/include/Base_Thread.h index 485fec913b..b4fb25c1a7 100644 --- a/include/Base_Thread.h +++ b/include/Base_Thread.h @@ -65,7 +65,7 @@ class Base_Thread { template void check_for_invalid_fd(unsigned int n); template - void ProcessAllSessions_SortingSessions(); + void ProcessAllSessions_Partition(); template void ProcessAllMyDS_AfterPoll(); template diff --git a/lib/Base_Thread.cpp b/lib/Base_Thread.cpp index 263ac1004f..25155dbf6b 100644 --- a/lib/Base_Thread.cpp +++ b/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(int); template PgSQL_Session* Base_Thread::create_new_session_and_client_data_stream(int); -template void Base_Thread::ProcessAllSessions_SortingSessions(); -template void Base_Thread::ProcessAllSessions_SortingSessions(); +template void Base_Thread::ProcessAllSessions_Partition(); +template void Base_Thread::ProcessAllSessions_Partition(); template void Base_Thread::ProcessAllMyDS_AfterPoll(); template void Base_Thread::ProcessAllMyDS_AfterPoll(); template void Base_Thread::ProcessAllMyDS_BeforePoll(); @@ -230,34 +230,58 @@ void Base_Thread::check_for_invalid_fd(unsigned int n) { } } -// 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 contiguous blocks + * + * Layout produced in mysql_sessions->pdata: + * [0, running_end) block A: running a query against the backend + * [running_end, idle_begin) block B: acquiring a backend (max_connect_time != 0) + * [idle_begin, len) block C: idle (no backend, or holds-conn-but-WAITING_CLIENT_DATA) + * + * Block A is "session is currently driving the backend" — these sessions have a chance + * of releasing the connection at end-of-query (depending on multiplexing eligibility, + * transaction state, etc.), giving block B sessions a fairness chance to acquire it. + * Sessions parked in WAITING_CLIENT_DATA (e.g., 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. + * + * Single O(n) pass, in place. idx walks up, idle_begin walks down, they meet and terminate. */ template -void Base_Thread::ProcessAllSessions_SortingSessions() { - unsigned int a=0; - for (unsigned int n=0; nlen; 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() { + size_t running_end = 0; + size_t idle_begin = mysql_sessions->len; + size_t idx = 0; + + while (idx < idle_begin) { + S* s = static_cast(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) { + ++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 } } } diff --git a/lib/MySQL_Thread.cpp b/lib/MySQL_Thread.cpp index 1e48476639..26abae76e1 100644 --- a/lib/MySQL_Thread.cpp +++ b/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(); + ProcessAllSessions_Partition(); } for (n=0; nlen; n++) { MySQL_Session *sess=(MySQL_Session *)mysql_sessions->index(n); diff --git a/lib/PgSQL_Thread.cpp b/lib/PgSQL_Thread.cpp index e99e3f8c5e..94874e39e7 100644 --- a/lib/PgSQL_Thread.cpp +++ b/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(); + ProcessAllSessions_Partition(); } for (n = 0; n < mysql_sessions->len; n++) { PgSQL_Session* sess = (PgSQL_Session*)mysql_sessions->index(n); From 587950de65e84cb200fcdb85f0da96894f1fc5b1 Mon Sep 17 00:00:00 2001 From: Rahim Kanji Date: Wed, 13 May 2026 11:54:05 +0500 Subject: [PATCH 2/3] promote long-waiters in block B by wait-time threshold After the 3-way A/B/C partition, sweep block B once and promote any session whose wait_time exceeds K_WAIT * avg(wait_time) toward the front of the B band. This gives sessions that have been waiting longer than the block average a chance at the next released backend conn, reducing head-of-line starvation when many clients share few conns. - Loop 1 (partition pass) now also accumulates sum_wait and b_count over block B, where wait_time = curtime - CurrentQuery.start_time in microseconds. CurrentQuery.start_time is guaranteed non-zero for any session in B; the (st && curtime > st) guard is defensive. - Loop 2 (O(b_count)) is a one-sided in-place partition: any B session with wt > K_WAIT * avg(wt) is swapped toward running_end. The promoted group is moved together but not sorted internally; same on the non-promoted side. The next pass re-evaluates with fresh wt. --- lib/Base_Thread.cpp | 68 +++++++++++++++++++++++++++++++++++---------- 1 file changed, 54 insertions(+), 14 deletions(-) diff --git a/lib/Base_Thread.cpp b/lib/Base_Thread.cpp index 25155dbf6b..f84dfa4d4e 100644 --- a/lib/Base_Thread.cpp +++ b/lib/Base_Thread.cpp @@ -232,27 +232,41 @@ void Base_Thread::check_for_invalid_fd(unsigned int n) { /** - * @brief Partition all sessions into three contiguous blocks + * @brief Partition all sessions into three blocks, then FIFO-bump + * the long-waiters within block B. * - * Layout produced in mysql_sessions->pdata: - * [0, running_end) block A: running a query against the backend - * [running_end, idle_begin) block B: acquiring a backend (max_connect_time != 0) - * [idle_begin, len) block C: idle (no backend, or holds-conn-but-WAITING_CLIENT_DATA) + * 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 is "session is currently driving the backend" — these sessions have a chance - * of releasing the connection at end-of-query (depending on multiplexing eligibility, - * transaction state, etc.), giving block B sessions a fairness chance to acquire it. - * Sessions parked in WAITING_CLIENT_DATA (e.g., 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. + * 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. + * 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. * - * Single O(n) pass, in place. idx walks up, idle_begin walks down, they meet and terminate. */ template 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; @@ -273,6 +287,10 @@ void Base_Thread::ProcessAllSessions_Partition() { ++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; @@ -284,6 +302,28 @@ void Base_Thread::ProcessAllSessions_Partition() { // 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(static_cast(avg_wait) * K_WAIT); + size_t a = running_end; + for (size_t n = running_end; n < idle_begin; ++n) { + S * sn = static_cast(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; + } + } + } } // this function was inline in MySQL_Thread::run() From 1430f50caaf2a71ec0f490a488aec5a5329eb936 Mon Sep 17 00:00:00 2001 From: Rahim Kanji Date: Wed, 13 May 2026 14:37:48 +0500 Subject: [PATCH 3/3] prevent poll_timeout corruption from stale pause_until tune_timeout_for_session_needs_pause computed (pause_until - curtime) as unsigned subtraction without checking if pause_until was still in the future. When pause_until <= curtime (stale), the result underflowed to ~1.8e19, which was then assigned to a signed int poll_timeout, becoming a large negative integer. The downstream ttw calculation at PgSQL_Thread.cpp:3210 compares this against (unsigned int)pgsql_thread___poll_timeout. Signed-to-unsigned promotion turned the negative value into a huge unsigned, the comparison took the wrong branch, and ttw fell back to the default poll_timeout (2000 ms). The worker thread then blocked in poll() for up to 2 seconds, freezing all sessions on that thread. --- lib/Base_Thread.cpp | 21 +++++++++++++++++---- lib/MySQL_Session.cpp | 3 ++- lib/PgSQL_Session.cpp | 2 ++ 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/lib/Base_Thread.cpp b/lib/Base_Thread.cpp index 263ac1004f..ee7102b984 100644 --- a/lib/Base_Thread.cpp +++ b/lib/Base_Thread.cpp @@ -347,10 +347,23 @@ void Base_Thread::tune_timeout_for_myds_needs_pause(DS * myds) { template void Base_Thread::tune_timeout_for_session_needs_pause(DS * myds) { T* thr = static_cast(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 diff --git a/lib/MySQL_Session.cpp b/lib/MySQL_Session.cpp index 957c1c25a0..5f49f65747 100644 --- a/lib/MySQL_Session.cpp +++ b/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; diff --git a/lib/PgSQL_Session.cpp b/lib/PgSQL_Session.cpp index c63ac7c432..04577360e6 100644 --- a/lib/PgSQL_Session.cpp +++ b/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