diff --git a/include/Base_Thread.h b/include/Base_Thread.h index de8198a313..727177a878 100644 --- a/include/Base_Thread.h +++ b/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 diff --git a/include/MySQL_Thread.h b/include/MySQL_Thread.h index 9a02809935..c2003639ef 100644 --- a/include/MySQL_Thread.h +++ b/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(); diff --git a/include/PgSQL_Thread.h b/include/PgSQL_Thread.h index bb66d86fb1..87b22062dc 100644 --- a/include/PgSQL_Thread.h +++ b/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; diff --git a/lib/Base_Thread.cpp b/lib/Base_Thread.cpp index 4c1806717e..060487dc24 100644 --- a/lib/Base_Thread.cpp +++ b/lib/Base_Thread.cpp @@ -1,5 +1,7 @@ #include "Base_Thread.h" +#include + #include "cpp.h" #include @@ -40,6 +42,7 @@ bool Base_Thread::update_partition_gate() { 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) { @@ -328,10 +331,59 @@ void Base_Thread::ProcessAllSessions_Partition() { } } - // 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(a)->mybe->server_myds->max_connect_time + < static_cast(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]; diff --git a/lib/MySQL_Thread.cpp b/lib/MySQL_Thread.cpp index f7edf2038e..6d2887d3bd 100644 --- a/lib/MySQL_Thread.cpp +++ b/lib/MySQL_Thread.cpp @@ -4223,7 +4223,45 @@ void MySQL_Thread::run() { 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 @@ void MySQL_Thread::idle_thread_check_if_worker_thread_has_unprocess_resumed_sess * * @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); } diff --git a/lib/PgSQL_Thread.cpp b/lib/PgSQL_Thread.cpp index 1fb8dc709c..e1c5c2a492 100644 --- a/lib/PgSQL_Thread.cpp +++ b/lib/PgSQL_Thread.cpp @@ -3241,6 +3241,27 @@ void PgSQL_Thread::run() { pre_poll_time = curtime; int ttw = (mypolls.poll_timeout ? (mypolls.poll_timeout / 1000 < (unsigned int)pgsql_thread___poll_timeout ? mypolls.poll_timeout / 1000 : pgsql_thread___poll_timeout) : pgsql_thread___poll_timeout); + + // Adaptive poll timeout. partition_pool_nulls holds the count from the + // process_all_sessions() pass that just ran: update_partition_gate() + // consumes the previous tick's value at the top of that function, so + // what is left here is this iteration's. A non-zero value means a + // session in this worker wanted a backend connection and did not get + // one, and the connection that unblocks it may be freed by a peer + // worker without ever touching this worker's fds. See the rationale + // on APT_* in PgSQL_Thread.h. + // + // The window check is the CPU guard: without demonstrated throughput + // there is no evidence a connection is about to be returned, so the + // full timeout is kept rather than spinning. + apt_update_window(); + if ( + partition_pool_nulls > 0 + && apt_window_total >= APT_MIN_QUERIES + && ttw > APT_SHORT_TTW_MS + ) { + ttw = APT_SHORT_TTW_MS; + } #ifdef IDLE_THREADS if (GloVars.global.idle_threads && idle_maintenance_thread) { memset(events, 0, sizeof(struct epoll_event) * MY_EPOLL_THREAD_MAXEVENTS); // let's make valgrind happy. It also seems that needs to be zeroed anyway @@ -3403,7 +3424,62 @@ void PgSQL_Thread::run() { unsigned int w = rand_fast() % (GloPTH->num_threads); PgSQL_Thread* thr = GloPTH->pgsql_threads[w].worker; if (resume_mysql_sessions->len) { - idle_thread_assigns_sessions_to_worker_thread(thr); + // Split the batch across two consecutive workers instead of + // dropping all of it on one. Handing the whole queue to a + // single randomly-chosen worker made the distribution + // unstable: measured 1598/2 across two workers at 1600 + // clients, established during the connection ramp and never + // corrected, because migration stops once sessions are no + // longer idle. + // + // This is not affinity -- sessions still go to arbitrary + // workers by design. It only stops a single draw from + // deciding the fate of an entire batch. + // + // Cost is one extra mutex pair and one extra pipe write per + // handoff. At the observed handoff rate that is far below + // measurement noise. + // + // The two locks are taken sequentially, never held together, + // so two idle threads picking overlapping pairs in opposite + // order cannot deadlock. + unsigned int nthr = GloPTH->num_threads; + if (nthr > 1) { + unsigned int w2 = (w + 1) % nthr; + PgSQL_Thread* thr2 = GloPTH->pgsql_threads[w2].worker; + + // Power of two choices: sample two workers and give the + // whole batch to the less loaded one. + // + // Splitting the batch evenly was tried first and did not + // work: the light worker received its share and re-exported + // it within a second, because a worker exports all of its + // idle sessions every loop iteration and a light worker + // iterates far more often. Measured 1598/2 either way. + // Sending everything to the lighter worker gives a + // restoring force instead of a fair coin -- it keeps + // winning until it is no longer the lighter one. + // + // Dirty reads are deliberate. This is a load hint, not an + // invariant; a stale value costs at most one misdirected + // batch, and taking locks to read two counters would cost + // more than it could ever save. + // + // resume_mysql_sessions is included in the estimate because + // those sessions are already promised to that worker but + // not yet absorbed. Without it, several handoffs in quick + // succession all see the same low mysql_sessions->len and + // pile onto the same 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); @@ -3505,14 +3581,21 @@ void PgSQL_Thread::idle_thread_check_if_worker_thread_has_unprocess_resumed_sess pthread_mutex_unlock(&thr->myexchange.mutex_resumes); } -void PgSQL_Thread::idle_thread_assigns_sessions_to_worker_thread(PgSQL_Thread * thr) { +void PgSQL_Thread::idle_thread_assigns_sessions_to_worker_thread(PgSQL_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; a bound lets one batch be split across + // several workers instead of landing entirely on one. + unsigned int to_move = resume_mysql_sessions->len; + if (max_sessions && max_sessions < to_move) { + to_move = max_sessions; + } + while (to_move--) { PgSQL_Session* mysess = (PgSQL_Session*)resume_mysql_sessions->remove_index_fast(0); thr->myexchange.resume_mysql_sessions->add(mysess); } @@ -3852,6 +3935,38 @@ void PgSQL_Thread::ProcessAllSessions_MaintenanceLoop(PgSQL_Session * sess, unsi } } +void PgSQL_Thread::apt_update_window() { + const unsigned long long q = status_variables.stvar[st_var_queries]; + // Cumulative counter; only the delta since the last iteration is new work. + const unsigned long long delta = (q >= apt_last_queries) ? (q - apt_last_queries) : 0; + apt_last_queries = q; + + if (apt_bucket_start == 0) { + apt_bucket_start = curtime; + } + + const unsigned long long elapsed = (curtime > apt_bucket_start) ? (curtime - apt_bucket_start) : 0; + if (elapsed >= APT_BUCKET_US * APT_BUCKETS) { + // Idle longer than the whole window: everything in it is stale, and + // advancing bucket by bucket would be a pointless loop. + memset(apt_buckets, 0, sizeof(apt_buckets)); + apt_window_total = 0; + apt_idx = 0; + apt_bucket_start = curtime; + } else { + unsigned long long advance = elapsed / APT_BUCKET_US; + while (advance--) { + apt_idx = (apt_idx + 1) % APT_BUCKETS; + apt_window_total -= apt_buckets[apt_idx]; + apt_buckets[apt_idx] = 0; + apt_bucket_start += APT_BUCKET_US; + } + } + + apt_buckets[apt_idx] += delta; + apt_window_total += delta; +} + void PgSQL_Thread::process_all_sessions() { unsigned int n; unsigned int total_active_transactions_ = 0; @@ -3973,6 +4088,72 @@ void PgSQL_Thread::process_all_sessions() { } } } + // Second pass over the sessions that failed to get a backend connection. + // + // Connections are released from inside handler() during the scan above, so + // a session that gave up at index 50 may be servable by a release that + // happened at index 700 -- in this same pass. Without this it waits for + // something to set to_process again, and for a session parked on a failed + // checkout that is ProcessAllSessions_MaintenanceLoop(), which runs on a + // hardcoded 1 second interval. That 1Hz retry is what bounded the observed + // tail (p99 ~1.8s, max ~1.9s) while p50 stayed near 1ms. + // + // Retrying across iterations does not help: nothing changes between polls. + // The releases happen during the scan, which is why the retry has to be + // here rather than a shorter poll timeout or a session-level deadline. + // + // Gated on partition_pool_nulls: zero means nobody failed a checkout in + // this pass, so there is nothing to retry and this costs one branch. + // Deliberately a single extra pass, not a loop to fixpoint -- bounded work + // per iteration, and a session that still cannot be served falls back to + // the existing path. + // TEMPORARY INSTRUMENTATION -- remove before merge. Counts how many + // sessions the rescan actually finds and how many it manages to serve, so + // "no difference" can be attributed to the right cause: never firing, + // finding no candidates, or finding them and still getting no connection. + unsigned int rescan_cand = 0; + unsigned int rescan_served = 0; + unsigned int rescan_skip_noproc = 0; + unsigned int rescan_skip_hasconn = 0; + unsigned int rescan_skip_nombe = 0; + + if (partition_pool_nulls > 0) { + for (n = 0; n < mysql_sessions->len; n++) { + PgSQL_Session* sess = (PgSQL_Session*)mysql_sessions->index(n); + // Still wants processing, is not paused, and has a backend that + // never got a connection: exactly the failed-checkout state. + if (sess->to_process != 1) { rescan_skip_noproc++; continue; } + if (sess->pause_until > curtime) continue; + if (sess->mybe == NULL || sess->mybe->server_myds == NULL) { rescan_skip_nombe++; continue; } + if (sess->mybe->server_myds->myconn != NULL) { rescan_skip_hasconn++; continue; } + + rescan_cand++; + rc = sess->handler(); + if (rc != -1 && sess->killed == false + && sess->mybe && sess->mybe->server_myds && sess->mybe->server_myds->myconn) { + rescan_served++; + } + if (rc == -1 || sess->killed == true) { + char _buf[1024]; + if (sess->client_myds && sess->killed) + proxy_warning("Closing killed client connection %s:%d\n", sess->client_myds->addr.addr, sess->client_myds->addr.port); + snprintf(_buf, sizeof(_buf), "%s:%d:%s()", __FILE__, __LINE__, __func__); + GloPgSQL_Logger->log_audit_entry(PGSQL_LOG_EVENT_TYPE::AUTH_CLOSE, sess, NULL, _buf); + unregister_session(n); + n--; + delete sess; + } + } + } + + // TEMPORARY INSTRUMENTATION -- one line per second (maintenance tick). + if (maintenance_loop) { + proxy_info("PGSQL_RESCAN thr=%p sessions=%u nulls=%u attempts=%u cand=%u served=%u skip[noproc=%u hasconn=%u nombe=%u]\n", + this, mysql_sessions->len, partition_pool_nulls, partition_pool_attempts, + rescan_cand, rescan_served, + rescan_skip_noproc, rescan_skip_hasconn, rescan_skip_nombe); + } + if (maintenance_loop) { unsigned int total_active_transactions_tmp; total_active_transactions_tmp = __sync_add_and_fetch(&status_variables.active_transactions, 0); @@ -5883,10 +6064,18 @@ void PgSQL_Thread::push_MyConn_local(PgSQL_Connection * c) { PgSQL_SrvC* mysrvc = (PgSQL_SrvC*)c->parent; if (mysrvc->status == MYSQL_SERVER_STATUS_ONLINE) { if (c->async_state_machine == ASYNC_IDLE) { - unsigned int n = (GloPTH && GloPTH->num_threads > 0) ? GloPTH->num_threads : 1; - if ((push_local_counter++ % n) == 0) { - cached_connections->add(c); - return; + // Never cache locally while somebody is waiting for a connection. + // return_local_connections() only publishes the cache at the end + // of the pass, and a pass walks every session this worker owns -- + // so the time a cached connection stays invisible to peer workers + // grows with client count, precisely when starvation is worst. + // The 1-in-N ratio below is fixed and does not scale with that. + if (!pool_has_waiters()) { + unsigned int n = (GloPTH && GloPTH->num_threads > 0) ? GloPTH->num_threads : 1; + if ((push_local_counter++ % n) == 0) { + cached_connections->add(c); + return; + } } } }