[DONOTMERGE] PgSQL pool starvation / worker imbalance investigation

include/Base_Thread.h

@@ -53,8 +53,15 @@ class Base_Thread {
 	// once per outer process_all_sessions iteration. Single-threaded per worker.
 	unsigned int partition_pool_attempts = 0;
 	unsigned int partition_pool_nulls = 0;
+	// The value update_partition_gate() last consumed. Kept because the live
+	// counter is zeroed at the top of every process_all_sessions() pass, so
+	// without it the first connection releases of a pass always look
+	// contention-free even in the middle of sustained starvation.
+	unsigned int partition_pool_nulls_prev = 0;
 	unsigned int partition_streak = 0;
 	bool partition_active = false;
+	// curtime of the last B-band sort, for PARTITION_SORT_MIN_INTERVAL_US.
+	unsigned long long last_partition_sort_time = 0;
 
 public:
 	// Gate thresholds: NULL-ratio (NUM/DEN) classifies a tick as "stressed";
@@ -67,6 +74,20 @@ class Base_Thread {
 	// streak are left untouched. Avoids "2/2 NULL = 100% stressed" noise.
 	static constexpr unsigned int PARTITION_GATE_MIN_ATTEMPTS   = 4;
 	static constexpr unsigned int PARTITION_FAIRNESS_MIN_B      = 4;
+	// Sorting the whole B band by wait time used to run on every iteration and
+	// was removed: it cost ~12% throughput at 500 clients / 50-conn pool under
+	// SSL. It is rate limited by TIME rather than by iteration count, because
+	// iteration count is not a stable unit -- loop frequency depends on session
+	// count and load, so a per-N-iterations gate makes a fast worker sort far
+	// more often than a slow one, exactly when it can least afford to. A wall
+	// clock bound caps the cost at a known number of sorts per second whatever
+	// the loop is doing.
+	//
+	// Ordering decays between sorts but degrades gracefully: max_connect_time
+	// is a fixed offset from when a session began waiting, so it is monotonic
+	// in arrival order, and the sessions that matter for the tail are the
+	// long-waiting ones whose relative order is stable.
+	static constexpr unsigned long long PARTITION_SORT_MIN_INTERVAL_US = 50000;	// 50ms
 
 	// Called by sessions inside this worker at the get_MyConn_from_pool()
 	// call site to feed the gate.
@@ -75,6 +96,14 @@ class Base_Thread {
 		if (was_null) ++partition_pool_nulls;
 	}
 
+	/// True when a session failed to get a connection from the pool in this
+	/// pass or the one before it, i.e. somebody is waiting for a connection
+	/// right now. Cheap proxy for a real waiter count, using counters the
+	/// partition gate already maintains.
+	inline bool pool_has_waiters() const {
+		return partition_pool_nulls > 0 || partition_pool_nulls_prev > 0;
+	}
+
 	// Runs the hysteresis state machine from per-tick counters, resets them,
 	// and returns whether the partition pass should run. MUST be called every
 	// outer iteration of process_all_sessions (even when the partition path

include/MySQL_Thread.h

@@ -157,7 +157,9 @@ class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread
 	void worker_thread_assigns_sessions_to_idle_thread(MySQL_Thread *thr);
 	void worker_thread_gets_sessions_from_idle_thread();
 	void idle_thread_gets_sessions_from_worker_thread();
-	void idle_thread_assigns_sessions_to_worker_thread(MySQL_Thread *thr);
+	/// Hand resumed sessions from this idle thread to a worker.
+	/// max_sessions == 0 means "all of them".
+	void idle_thread_assigns_sessions_to_worker_thread(MySQL_Thread *thr, unsigned int max_sessions = 0);
 	void idle_thread_check_if_worker_thread_has_unprocess_resumed_sessions_and_signal_it(MySQL_Thread *thr);
 	void idle_thread_prepares_session_to_send_to_worker_thread(int i);
 	void idle_thread_to_kill_idle_sessions();

include/PgSQL_Thread.h

@@ -175,7 +175,10 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	void worker_thread_assigns_sessions_to_idle_thread(PgSQL_Thread * thr);
 	void worker_thread_gets_sessions_from_idle_thread();
 	void idle_thread_gets_sessions_from_worker_thread();
-	void idle_thread_assigns_sessions_to_worker_thread(PgSQL_Thread * thr);
+	/// Hand resumed sessions from this idle thread to a worker.
+	/// max_sessions == 0 means "all of them"; a non-zero value moves at most
+	/// that many, so a batch can be split across more than one worker.
+	void idle_thread_assigns_sessions_to_worker_thread(PgSQL_Thread * thr, unsigned int max_sessions = 0);
 	void idle_thread_check_if_worker_thread_has_unprocess_resumed_sessions_and_signal_it(PgSQL_Thread * thr);
 	void idle_thread_prepares_session_to_send_to_worker_thread(int i);
 	void idle_thread_to_kill_idle_sessions();
@@ -235,6 +238,52 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	int pipefd[2];
 	PgSQL_Session_Interrupt_Queue_t sess_intrpt_queue;
 
+	/*
+	 * Adaptive poll timeout.
+	 *
+	 * A worker that has drained its own sessions blocks in poll() on its own
+	 * file descriptors. When it is waiting for a backend connection, the
+	 * connection that would unblock it is freed by a *different* worker into
+	 * the shared pool, and nothing in this worker's fd set ever fires. It
+	 * therefore sleeps for the whole poll timeout before retrying. Measured:
+	 * max_ms tracked pgsql-poll_timeout (~8s at 2000ms, ~1.3s at 500ms) and
+	 * collapsed to 74ms with threads=1, where no cross-worker handoff exists.
+	 *
+	 * Waking peers is not a fix: under starvation every worker wants a
+	 * connection, a returning connection carries no indication of which
+	 * worker can use it, and a wakeup may just be client traffic.
+	 *
+	 * So only the timeout is shortened, and only when both hold:
+	 *   - a session in this worker just failed to get a connection from the
+	 *     pool (partition_pool_nulls, already counted for the partition gate)
+	 *   - this worker has demonstrated recent throughput
+	 *
+	 * The second condition is what bounds the CPU cost. Throughput proves
+	 * connections are being returned at a high rate, so a short sleep is
+	 * near-certain to find one. With no recent throughput there is no
+	 * evidence any connection is coming, so the full timeout is kept and the
+	 * worker does not spin. A rolling window is used rather than an
+	 * instantaneous rate so that a single quiet iteration does not disengage
+	 * it mid-episode.
+	 */
+	static constexpr unsigned int      APT_BUCKETS      = 10;
+	static constexpr unsigned long long APT_BUCKET_US   = 200000;	// 200ms => 2s window
+	// Queries across the window before shortening is considered justified.
+	// At this rate a connection returns far more often than once per
+	// APT_SHORT_TTW_MS, so the shortened sleep is expected to be productive.
+	static constexpr unsigned long long APT_MIN_QUERIES = 2000;
+	static constexpr int               APT_SHORT_TTW_MS = 1;
+
+	unsigned long long apt_buckets[APT_BUCKETS] = {};
+	unsigned long long apt_window_total = 0;
+	unsigned long long apt_last_queries = 0;
+	unsigned long long apt_bucket_start = 0;
+	unsigned int       apt_idx = 0;
+
+	/// Roll the query-rate window forward to curtime. Called once per loop
+	/// iteration, not per query.
+	void apt_update_window();
+
 	//bool epoll_thread;
 	bool poll_timeout_bool;
 

lib/Base_Thread.cpp

@@ -1,5 +1,7 @@
 #include "Base_Thread.h"
 
+#include <algorithm>
+
 #include "cpp.h"
 
 #include <unistd.h>
@@ -40,6 +42,7 @@
 	const uint64_t nulls    = partition_pool_nulls;
 	partition_pool_attempts = 0;
 	partition_pool_nulls    = 0;
+	partition_pool_nulls_prev = (unsigned int)nulls;
 
 	// Low-volume ticks carry no signal; leave gate and streak unchanged.
 	if (attempts < PARTITION_GATE_MIN_ATTEMPTS) {
@@ -281,7 +284,7 @@
  * than run unconditionally on every iteration.
  */
 template<typename S>
 void Base_Thread::ProcessAllSessions_Partition() {
 	size_t running_end = 0;
 	size_t idle_begin = mysql_sessions->len;
 	size_t idx = 0;
@@ -328,10 +331,59 @@
 		}
 	}
 
-	// Promote the longest-waiting B session (smallest max_connect_time) to
-	// running_end so the CONNECTING_SERVER pass serves it first. Gated by a
-	// minimum B-band size to avoid churn on tiny bands.
-	if (idle_begin > running_end + PARTITION_FAIRNESS_MIN_B
+	// Order the B band oldest-first, occasionally.
+	//
+	// max_connect_time is stamped as curtime + connect_timeout_server_max when
+	// a session enters PROCESSING_QUERY, before it tries the pool -- so it is a
+	// fixed offset from when the session started waiting, and ascending order
+	// is FIFO by arrival. Band B is therefore not just "sessions connecting":
+	// with connect_timeout_server_max non-zero (default 10000) it holds every
+	// session with a pending query, including all the ones that failed a pool
+	// checkout.
+	//
+	// Promoting only the single oldest session, as the fallback below does,
+	// rescues one waiter per pass and leaves the rest in arbitrary order, which
+	// lets a subset lose the checkout race repeatedly. A full sort gives
+	// approximate FIFO across the whole band. It is rate-limited because doing
+	// it every iteration was measured at ~12% throughput loss.
+	const size_t b_len = (idle_begin > running_end) ? (idle_begin - running_end) : 0;
+	const bool sort_due = (curtime >= last_partition_sort_time + PARTITION_SORT_MIN_INTERVAL_US);
+	if (b_len > 1 && sort_due) {
+		last_partition_sort_time = curtime;
+		// Every element in [running_end, idle_begin) satisfied is_B, so
+		// mybe->server_myds is non-null and max_connect_time is non-zero.
+		auto cmp = [](void* a, void* b) {
+			return static_cast<S*>(a)->mybe->server_myds->max_connect_time
+			     < static_cast<S*>(b)->mybe->server_myds->max_connect_time;
+		};
+
+		// Only the front of the band can be served this pass -- the pool is
+		// far smaller than the client count, so most of the band is scrap we
+		// were never going to reach anyway. Fully sorting all of it (the
+		// std::sort this replaces) spends O(n log n) to order elements whose
+		// relative order will never be observed before the next sort.
+		//
+		// nth_element partitions the N smallest to the front in O(n) average,
+		// unordered among themselves; sorting just that front slice is
+		// O(N log N). N = 10% of the band: small enough to be cheap even at
+		// full band size, large enough that a session just past the cutoff
+		// this pass is very likely inside it on the next one.
+		void** begin = mysql_sessions->pdata + running_end;
+		void** end   = mysql_sessions->pdata + idle_begin;
+		size_t top_n = b_len / 5;
+		if (top_n < 1) top_n = 1;
+		if (top_n >= b_len) {
+			std::sort(begin, end, cmp);
+		} else {
+			void** nth = begin + top_n;
+			std::nth_element(begin, nth, end, cmp);
+			std::sort(begin, nth, cmp);
+		}
+	}
+	// Fallback when the band was not sorted this pass: promote the
+	// longest-waiting B session so the CONNECTING_SERVER pass serves it first.
+	// Gated by a minimum B-band size to avoid churn on tiny bands.
+	else if (idle_begin > running_end + PARTITION_FAIRNESS_MIN_B
 	    && oldest_idx != SIZE_MAX && oldest_idx != running_end) {
 		void* p = mysql_sessions->pdata[running_end];
 		mysql_sessions->pdata[running_end] = mysql_sessions->pdata[oldest_idx];

lib/MySQL_Thread.cpp

@@ -4223,7 +4223,45 @@
 			unsigned int w=rand_fast()%(GloMTH->num_threads);
 			MySQL_Thread *thr=GloMTH->mysql_threads[w].worker;
 			if (resume_mysql_sessions->len) {
-				idle_thread_assigns_sessions_to_worker_thread(thr);
+				// Power of two choices: sample two workers and give the whole
+				// batch to the less loaded one.
+				//
+				// Picking a single worker at random and moving the entire
+				// resume queue there is unstable. Measured on the PgSQL path
+				// under an equivalent workload: 1598 sessions on one worker
+				// and 2 on the other, at 1600 clients. The split is
+				// established while connections ramp up and is never
+				// corrected, because once sessions stop going idle the
+				// migration that would rebalance them stops too.
+				//
+				// A fair coin cannot fix it, and splitting the batch evenly
+				// between two workers was tried and did not either: a worker
+				// exports all of its idle sessions every loop iteration, and a
+				// light worker's loop is fast because poll() and session
+				// processing are both O(sessions), so it re-exports new
+				// arrivals almost immediately. Directing the whole batch at
+				// the lighter worker gives a restoring force rather than
+				// merely matching that bleed. This produced an exact 800/800
+				// split where every other approach produced 1598/2.
+				//
+				// Dirty reads are deliberate: a load hint, not an invariant.
+				// A stale value costs at most one misdirected batch, and
+				// locking to read two counters would cost more than it saves.
+				// resume_mysql_sessions is included because those sessions are
+				// already promised to that worker but not yet absorbed.
+				unsigned int nthr = GloMTH->num_threads;
+				if (nthr > 1) {
+					unsigned int w2 = (w + 1) % nthr;
+					MySQL_Thread *thr2 = GloMTH->mysql_threads[w2].worker;
+					unsigned int load1 = thr->mysql_sessions->len
+						+ thr->myexchange.resume_mysql_sessions->len;
+					unsigned int load2 = thr2->mysql_sessions->len
+						+ thr2->myexchange.resume_mysql_sessions->len;
+					idle_thread_assigns_sessions_to_worker_thread(
+						(load2 < load1) ? thr2 : thr, 0);
+				} else {
+					idle_thread_assigns_sessions_to_worker_thread(thr, 0);
+				}
 			} else {
 				idle_thread_check_if_worker_thread_has_unprocess_resumed_sessions_and_signal_it(thr);
 			}
@@ -4361,14 +4399,20 @@
  * 
  * @param thr The worker thread to which idle sessions will be assigned.
  */
-void MySQL_Thread::idle_thread_assigns_sessions_to_worker_thread(MySQL_Thread *thr) {
+void MySQL_Thread::idle_thread_assigns_sessions_to_worker_thread(MySQL_Thread *thr, unsigned int max_sessions) {
 	bool send_signal = false;
 	// send_signal variable will control if we need to signal or not
 	// the worker thread
 	pthread_mutex_lock(&thr->myexchange.mutex_resumes);
 	if (shutdown==0 && thr->shutdown==0)
 	if (resume_mysql_sessions->len) {
-		while (resume_mysql_sessions->len) {
+		// max_sessions == 0 means "move everything", preserving the original
+		// behaviour for any caller that does not care.
+		unsigned int to_move = resume_mysql_sessions->len;
+		if (max_sessions && max_sessions < to_move) {
+			to_move = max_sessions;
+		}
+		while (to_move--) {
 			MySQL_Session *mysess=(MySQL_Session *)resume_mysql_sessions->remove_index_fast(0);
 			thr->myexchange.resume_mysql_sessions->add(mysess);
 		}