CUBRID Thread Manager — High-Concurrency Code-Level Deep Dive

Where this document fits: The high-level analysis cubrid-thread-manager.md covers design intent, motivation, and the two-phase (CBRD-26177 + CBRD-26662) evolution; the base engine is framed in cubrid-thread-worker-pool.md. This document traces every branch and field at the code level. Each chapter is self-contained, but reading in order follows the full lifecycle of a single client request under high concurrency — accepted by the epoll reactor, placed by the coordinator, handed to a back-end task worker, and (Phase 2) admitted under a per-core concurrency slot it surrenders on any logical wait.

Source provenance: Chapters 1–11 trace the merged develop checkout. Chapters 12–14 trace PR #7323 (feature/worker_pool_elastic), which is not yet merged — those symbols and line numbers are relative to that branch and will shift before it lands. Position-hint rows for PR-only files are marked with a PR: path prefix.

Contents:

Ch	Title	Status
1	Data Structure Map I the Connection Reactor	✅
2	Data Structure Map II the Coordinator	✅
3	Initialization and Memory	✅
4	The epoll Connection Worker Loop	✅
5	Send Recv Budgets and IO Bounding	✅
6	The Coordinator Placement Rebalance and Auto Scale	✅
7	Context Lifecycle and Per Worker Freelist	✅
8	From Bytes to a Task	✅
9	Back End Pool Engine I Structure and Dispatch	✅
10	Back End Pool Engine II the Worker Run Loop	✅
11	Atomic Free Statistics and Observability	✅
12	Phase 2 Data Structures Elastic Pool and Concurrency Slots	✅
13	Phase 2 Flows Slot Acquire Release and the Elastic Run Loop	✅
14	Phase 2 Flows the Slot Daemon and Logical Wait Wiring	✅

---

Chapter 1: Data Structure Map I the Connection Reactor

An accepted connection lives in the new front end as a triangle of three long-lived C++ objects: a context (the connection’s byte-level state), a worker (the thread servicing many contexts on one epoll loop), and a pool (the owner that allocates contexts and holds the worker vector). This chapter maps every field of that triangle plus the two CV primitives (thread_watcher, message_blocker) and the control vocabulary a worker consumes (message and its enums, timer slots, backpressure record). Later chapters trace the motion; here we fix the shapes and the invariants that keep the pointers honest.

Cross-link: why this reactor exists — the C10K/reactor rationale, the connection-worker vs back-end-pool split — is in the companion cubrid-thread-manager.md (“Reactor front end”).

1.1 The reactor triangle at a glance

Ownership runs one way; back-references the other. The pool owns the worker objects (by std::unique_ptr) and the context storage (an intrusive freelist). A worker holds a raw pool * and a set of raw context * it services. A context records only the index of its owning worker — indices survive a context migrating between workers (Ch 7), where a raw pointer would dangle.

Figure 1-1. The reactor triangle — ownership (solid) versus back-reference / index (dashed). The pool sits at the apex owning both the worker vector and the context freelist; the worker and context at the base corners point back with a raw pointer and an integer index respectively.

INVARIANT — a context never owns a raw pointer to its worker. context::m_worker is an int index into pool::m_workers. On rebalance (Ch 6/7) only this integer is rewritten; nothing is re-pointed and no worker * can outlive a worker teardown.

1.2 context — one accepted connection

// connection::context -- src/connection/connection_context.hpp
struct context {
  /* THIS MUST BE THE FIRST */
  css_conn_entry *m_conn;
  int m_worker;  uint64_t m_id;  ignore_level m_ignore;  bool m_removed;
  struct { state m_state; receiver m_receiver; cubbase::span<std::byte> m_header;
           int m_request_id; bool m_command; } m_recv;
  struct { transmitter m_transmitter; std::shared_ptr<message_blocker> m_blocker; } m_send;
  statistics::metrics<statistics::context> m_stats;
  /* copy+move deleted; ctors (capacity) and (); reset() */
};

Field	Role	Why it exists
m_conn	Pointer to the legacy css_conn_entry wrapped	Bridge to all pre-existing css_conn_entry code; kept first so context * aliases &m_conn
m_worker	Index of owning worker in pool::m_workers	Survives handoff; an index cannot dangle
m_id	Monotonic id assigned at claim	Keys m_exhausted and message routing without a pointer
m_ignore	ignore_level guard on stale ERR/HUP events	After close is decided, later error events are swallowed, not re-run
m_removed	Torn down but still in m_context	Marks it for purge_stale_contexts; distinguishes closing from gone
m_recv.m_state	state — HEADER/DATA/ERROR	Read side is a two-phase parser: header, then body sized by it
m_recv.m_receiver	receiver — pooled reassembly buffer + drain state	Owns socket-to-span reassembly
m_recv.m_header	span<std::byte> over parsed header	Header stays addressable while its body streams in
m_recv.m_request_id	Request id from header	Correlates reply to request
m_recv.m_command	Header was a command packet	If true, a back-end task is pushed once the body fully arrives
m_send.m_transmitter	transmitter — pooled outbound buffer + deleters	Owns partial-write state
m_send.m_blocker	shared_ptr<message_blocker>	A back-end worker doing a blocking send parks here; signalled on drain. Source notes it becomes a vector if concurrent blocking sends are allowed
m_stats	Per-context metrics	Rolled up to worker/coordinator (Ch 11)

INVARIANT — m_conn is first so context * aliases css_conn_entry **. Both connection::context and master::context mark m_conn /* THIS MUST BE THE FIRST */; at offset 0, legacy code holding &ctx->m_conn recovers the enclosing context by address, and the freelist trick (§1.8) depends on the same guarantee. Reorder and every such cast breaks silently.

context is non-copyable/non-movable (all four special members = delete) — it holds pooled buffers and is used only by pointer. Ctor context (std::size_t capacity) sizes the receiver pool; default context () defers it. reset() restores pristine state (m_conn=null, m_worker=-1, m_id=0, both sub-structs, m_stats) for freelist reuse (Ch 7).

// connection::state / ignore_level -- src/connection/connection_context.hpp
enum class state { HEADER, DATA, ERROR };
enum class ignore_level : uint8_t { DONT_IGNORE = 0, IGNORE_ALL };

state is the read-side phase. ignore_level is uint8_t-backed because it also travels inside a message (§1.5); DONT_IGNORE = 0 is the default so a zero-initialized context never accidentally suppresses events.

1.3 The sync primitives — thread_watcher, message_blocker

Both are CV bundles declared one namespace up (cubconn) because pool, workers, and coordinator all share them.

// thread_watcher / message_blocker -- src/connection/connection_context.hpp
struct thread_watcher { std::mutex mtx; std::condition_variable cv; int active; };
struct message_blocker { std::mutex m; std::condition_variable cv; bool done; };

Struct.Field	Role	Why it exists
thread_watcher.mtx	Guards active	active must move atomically vs waiters
thread_watcher.cv	Signals startup/shutdown quorum	Pool waits while workers spin up/down
thread_watcher.active	Count of live worker threads	Pool blocks until all N reach READY / all exit
message_blocker.m	Guards done	Predicate lock for the CV
message_blocker.cv	Signals the blocked producer	Producer of a blocking message parks here
message_blocker.done	Predicate: worker finished this request	Guards spurious wakeups and lost signals

INVARIANT — a message_blocker outlives both parties via shared_ptr. It is always held by shared_ptr (in message::waiter_handle and context::m_send.m_blocker); producer and worker each hold a reference and whoever releases last destroys it. So the worker can signal cv after the producer timed out, and the producer can destruct while the worker still holds its copy — a raw pointer would let a timed-out producer free the blocker under the worker.

1.4 worker — the servicing thread

One OS thread, one epoll loop, many contexts.

// connection::worker (member layout) -- src/connection/connection_worker.hpp
pool *m_parent;  std::shared_ptr<coordinator> m_coordinator;
std::shared_ptr<thread_watcher> m_watcher;
std::thread m_thread;  std::size_t m_core;  status m_status;  bool m_stop;
cubthread::entry *m_entry;  std::unordered_set<context *> m_context;  std::size_t m_index;
cubsocket::epoll m_events;  int m_eventfd;  int m_timerfd;  uint64_t m_timens;
std::array<timer_handle, (size_t) timer_type::TYPE_COUNT> m_timer_handler;
bool m_has_retry;
tbb::concurrent_queue<message> m_queue[(size_t) queue_type::TYPE_COUNT];
std::atomic<uint64_t>          m_queue_size[(size_t) queue_type::TYPE_COUNT];
std::vector<context *> m_removed_context;
size_t m_recv_budget;  size_t m_send_budget;
std::unordered_map<uint64_t, exhausted_context> m_exhausted;
statistics::metrics<statistics::worker> m_stats;

Field	Role	Why it exists
m_parent	Raw pool * owner	Retire contexts / read config; pool outlives every worker
m_coordinator	shared_ptr<coordinator>	Reports metrics, receives placement (Ch 6); shared by many workers
m_watcher	shared_ptr<thread_watcher>	Bumps active at start/exit for the pool quorum
m_thread	The std::thread running run()	OS handle; joined at finalize
m_core	CPU core pinned to	Affinity for cache locality (Ch 3)
m_status	status — READY/RUNNING/HIBERNATING/TERMINATING	Drives the loop and hibernation (Ch 4)
m_stop	Hard-stop flag	Breaks the loop on shutdown regardless of status
m_entry	cubthread::entry *	Ties this thread into the legacy thread-manager
m_context	unordered_set<context *> serviced	Membership set the loop iterates; O(1) migration
m_index	This worker’s index in pool::m_workers	Matches context::m_worker; routing/handoff identity
m_events	cubsocket::epoll	epoll fd over client sockets + control fds
m_eventfd	eventfd for cross-thread wakeups	Producer signals it to break epoll_wait
m_timerfd	timerfd in the epoll set	One fd multiplexes all periodic timers
m_timens	Current timer period (ns)	Cached resolution the timerfd is armed to
m_timer_handler	array<timer_handle, TYPE_COUNT> by timer_type	One slot per timer kind; no alloc on hot path
m_has_retry	Any deferred close awaiting retry	Cheap skip of the retry scan when clear
m_queue[]	Two concurrent_queue<message> (IMMEDIATE, LAZY)	MPSC control channel from any thread
m_queue_size[]	atomic<uint64_t> per queue	Bounds one drain pass to messages present at entry
m_removed_context	Contexts marked removed, pending reap	Deferred cleanup, never erase mid-iteration
m_recv_budget / m_send_budget	Per-pass byte caps	Fairness so one connection can’t monopolize (Ch 5)
m_exhausted	map<uint64_t, exhausted_context>	Contexts that hit budget mid-IO, keyed by m_id (§1.7)
m_stats	Per-worker metrics	Aggregated to coordinator (Ch 11)

INVARIANT — m_queue_size snapshots bound each drain pass. The worker reads m_queue_size and processes only that many messages; MPSC arrivals during the pass wait for next wakeup. Draining “until empty” would let a hot producer pin the single consumer forever. The counter is atomic: producers increment from other threads while the one consumer decrements.

// worker::status -- src/connection/connection_worker.hpp
enum class status { READY, RUNNING, HIBERNATING, TERMINATING };

READY = constructed, parked pre-run; RUNNING = active loop; HIBERNATING = idle, timer-woken only, sockets parked (Ch 4); TERMINATING = draining to exit. m_stop is orthogonal — it forces exit from any status.

1.5 The control vocabulary — queue_type, message_type, message

Every instruction is a message on one of two queues. The queue says when, the type says what.

// worker::queue_type / message_type -- src/connection/connection_worker.hpp
enum class queue_type : uint8_t { IMMEDIATE, LAZY, TYPE_COUNT };
enum class message_type {
  START, HIBERNATE /*lazy*/, AWAKEN /*lazy*/, SHUTDOWN,
  NEW_CLIENT, HANDOFF_CLIENT /*lazy*/, TAKEOVER_CLIENT,
  SHUTDOWN_CLIENT /*lazy*/, SEND_PACKET, RELEASE_PACKET, TYPE_COUNT };

IMMEDIATE messages drain before IO; LAZY ones (HIBERNATE, AWAKEN, HANDOFF_CLIENT, SHUTDOWN_CLIENT, tagged /* lazy queue */ in the enum) run after IO so they see settled state. TYPE_COUNT in both enums is the array-sizing sentinel (m_queue, m_queue_size, m_timer_handler), not a value. message_type groups into worker-control (START, HIBERNATE, AWAKEN, SHUTDOWN) and client-control (the rest); their handlers are Ch 7/8.

// worker::message -- src/connection/connection_worker.hpp
struct message {
  message_type type;  uint64_t id;  context *ctx;  css_conn_entry *conn;
  std::vector<cubbase::span<std::byte>> packet;   /* SEND/RELEASE_PACKET */
  std::function<void ()> deleter;                 /* SEND_PACKET */
  worker *worker_ptr;  int worker_index;          /* HANDOFF_CLIENT */
  ignore_level ignore;  bool retry;               /* SHUTDOWN_CLIENT */
  std::shared_ptr<message_blocker> waiter_handle; /* START/SHUTDOWN_CLIENT/SEND_PACKET */
#if !defined (NDEBUG)
  uint64_t message_id;
#endif
};

Field	Role	Used by	Why it exists
type	Discriminator	all	Selects handler and live payload fields
id	Target context id	client msgs	Routes to a context without a pointer (matches m_id)
ctx	Target context	client msgs	Direct handle when producer already holds it
conn	Legacy connection entry	NEW_CLIENT etc.	Hands the raw css_conn_entry to wrap
packet	Byte spans to send/free	SEND/RELEASE_PACKET	Outbound payload, or buffers to release
deleter	Frees packet after send	SEND_PACKET	Producer keeps buffer-lifetime ownership
worker_ptr	Destination worker	HANDOFF_CLIENT	Peer worker receiving the migrated context
worker_index	Destination index	HANDOFF_CLIENT	Index form for routing/logging
ignore	Suppress-level to stamp	SHUTDOWN_CLIENT	Sets context::m_ignore during teardown
retry	Retry close if blocked	SHUTDOWN_CLIENT	Defers a close that still has pending tasks
waiter_handle	Blocker to signal on done	START/SHUTDOWN_CLIENT/SEND_PACKET	Lets a producer block until worker finishes
message_id	Debug monotonic id	all (debug)	Trace ordering; compiled out in release

INVARIANT — message is move-only. Copy = delete, move = default noexcept. It owns a vector, a std::function deleter, and a shared_ptr blocker — copying would double-free the buffers. It is always std::move-d into the queue (enqueue (type, std::move (item))); the noexcept move is what lets tbb::concurrent_queue store it without a copy fallback.

1.6 The timer machinery — timer_type, timer_latency, timer_handle

One timerfd drives up to four periodic jobs. timer_type names the job (and indexes m_timer_handler); timer_latency is the period in ns; timer_handle is one armed slot.

// worker timers -- src/connection/connection_worker.hpp
enum class timer_type : uint32_t { NA, HIBERNATE, STATISTICS, QUEUE, HA, TYPE_COUNT };
enum class timer_latency : uint64_t {
  NA = 0,                                /* off */
  LOW_LATENCY    = (uint64_t)(1 * 1e6),  /* 1 msec */
  MEDIUM_LATENCY = (uint64_t)(1 * 1e9),  /* 1 sec  */
  HIGH_LATENCY   = (uint64_t)(2 * 1e9) };/* 2 sec  */
struct timer_handle { bool valid; timer_latency latency;
                      std::function<bool ()> function; uint64_t last_time; };

timer_type: NA = unused slot 0; HIBERNATE fires the idle check; STATISTICS flushes metrics; QUEUE re-checks the message queues; HA drives heartbeat/HA close. timer_latency encodes the period directly (0 = off, else 1 ms / 1 s / 2 s).

timer_handle field	Role	Why it exists
valid	Slot armed	A fixed array needs per-slot occupancy
latency	Period for this timer	Compared vs elapsed to decide firing
function	std::function<bool ()> callback	The job; returns whether to keep the timer
last_time	ns timestamp of last fire	Elapsed = now − last_time; gates next fire

INVARIANT — the timerfd period is the minimum active latency. One fd multiplexes all slots, so it is armed to the smallest latency among valid slots, then each handler runs only when its own last_time shows enough elapsed. Arm coarser and a LOW_LATENCY job misses its deadline; last_time is what stops coarser jobs over-firing.

1.7 The backpressure record — exhausted_context

When a context hits m_recv_budget/m_send_budget mid-transfer, the worker parks it in m_exhausted (keyed by m_id) to resume where it stopped.

// worker::exhausted_context -- src/connection/connection_worker.hpp
struct exhausted_context { bool prepared; uint32_t events; context *ctx; };

Field	Role	Why it exists
prepared	Resume set up	Distinguishes a fresh park from one already re-armed
events	epoll mask pending when budget ran out	Resume replays the exact EPOLLIN/EPOLLOUT intent
ctx	The stalled context	Direct resume handle, no re-lookup

Keying by uint64_t m_id (not a pointer) means a context closed while exhausted is dropped by id without chasing a possibly-reaped pointer (Ch 5).

1.8 pool — owner and allocator

pool owns everything and alone touches the freelist and worker vector under a lock.

// connection::pool (members) -- src/connection/connection_pool.hpp
std::mutex m_mutex;
#if !defined (NDEBUG)
std::atomic<std::thread::id> m_mutex_holder { std::thread::id () };
#endif
std::vector<std::unique_ptr<worker>> m_workers;
std::shared_ptr<coordinator> m_coordinator;  std::shared_ptr<thread_watcher> m_watcher;
std::uint32_t m_max_connections, m_max_connection_workers, m_min_connection_workers;
struct { freelist *m_head; std::size_t m_max; std::size_t m_claim; } m_freelist;

Field	Role	Why it exists
m_mutex	Guards all pool mutable state	Contexts/workers touched from many threads
m_mutex_holder	Debug current lock owner	Cheap correct-thread assertions in debug
m_workers	vector<unique_ptr<worker>>	Sole owner of worker lifetime; clean teardown
m_coordinator	shared_ptr<coordinator>	Handed to every worker (Ch 2/6)
m_watcher	shared_ptr<thread_watcher>	Start/stop quorum shared with workers
m_max_connections	Hard cap on live connections	Sizes the freelist
m_max_connection_workers	Worker-count ceiling	Auto-scale upper bound (Ch 6)
m_min_connection_workers	Worker-count floor	Auto-scale lower bound / always-on
m_freelist.m_head	Head of free context list	O(1) claim/retire
m_freelist.m_max	Total contexts allocated	Capacity accounting
m_freelist.m_claim	Contexts currently claimed	Detects exhaustion vs m_max

The freelist node is intrusive:

// pool::freelist -- src/connection/connection_pool.hpp
struct freelist {
  /* THIS MUST BE THE FIRST */
  context m_context;   /* embedded by value */
  freelist *m_next;
  freelist (std::size_t capacity) : m_context (capacity), m_next (nullptr) {}
};

Field	Role	Why it exists
m_context	The context storage, embedded by value	No per-connection alloc; the node is the context
m_next	Next free node	Singly-linked free chain off m_head

INVARIANT — freelist::m_context is first so context * round-trips to freelist *. claim_context() pops m_head and returns &node->m_context; because m_context is at offset 0, retire_context (context *ctx) recovers the node with reinterpret_cast<freelist *> (ctx). Put m_next first and retire corrupts the free chain. Same address-alias trick as m_conn-first (§1.2).

flowchart LR
  H["m_freelist.m_head"] --> N0["freelist #0<br/>m_context (offset 0)<br/>m_next"]
  N0 --> N1["freelist #1<br/>m_context<br/>m_next"]
  N1 --> NIL["nullptr"]
  CLAIM["claim_context()<br/>returns &head-&gt;m_context"] -.-> N0

Figure 1-2. The intrusive freelist: each node embeds a context at offset 0; claim/retire cast between freelist * and context *.

The sibling master::context and the controller<RX,TX> template (the Unix-socket control channel in controller.hpp) belong to the master-connector path, not the reactor, but share the result enum every IO call here returns — Ok, Partial, Error, Reset, Pending, BudgetExhausted, PeerReset, RefuseConnection, ClosedConnection, Aborted, Skewed (buffer.hpp) — which later IO-branch chapters lean on.

1.9 Chapter summary — key takeaways

1. The triangle is pool owns worker points-to many context. Ownership is unique_ptr (workers) and an intrusive freelist (contexts); back-references are a raw pool * and an index context::m_worker, never a worker *.
2. context holds the whole per-connection byte state — m_conn bridge, m_id/m_worker identity, the m_recv (two-phase HEADER→DATA parser + receiver) and m_send (transmitter + optional blocker) sub-structs. It is non-copyable/non-movable and reset for reuse.
3. Two layout invariants are load-bearing: context::m_conn first (aliases css_conn_entry **) and freelist::m_context first (lets retire_context cast context * back to freelist *). Reorder either and casts corrupt silently.
4. A worker is a control-message consumer, not an RPC target: every instruction is a move-only message on the IMMEDIATE or LAZY tbb::concurrent_queue, and m_queue_size snapshots bound each drain pass to prevent producer starvation.
5. One timerfd multiplexes four jobs via m_timer_handler[timer_type]; armed to the minimum active timer_latency, each timer_handle::last_time gates its own firing.
6. Backpressure is explicit state: a budget-exhausted context is parked in m_exhausted keyed by m_id, remembering the exact epoll events to replay next pass.
7. thread_watcher and message_blocker are the only two CV primitives — the former counts live workers for the pool’s start/stop quorum; the latter, always shared_ptr, lets a producer block on a specific message until the worker signals done.

Chapter 2: Data Structure Map II the Coordinator

Chapter 1 mapped the connection reactor — the per-worker epoll loop that moves bytes. This chapter maps its counterpart: the single coordinator thread that never touches a client socket, yet decides which worker owns each connection, when to hand one off, and how many workers should exist. This is a field-level map only; the algorithms that read the state — score, rebalance selection, trial-based auto-scaling — are Chapter 6.

Cross-link: why a single-writer control plane exists is argued in the companion cubrid-thread-manager.md, sections Auto scaling (CBRD-26406), Score-based connection assignment, and Coordinator + context freelist.

2.1 The coordinator at a glance

coordinator (in cubconn::connection) is a plain member-owning class whose members form six functional clusters, walked field-by-field below.

Figure 2-1 — coordinator ownership map.

The control cluster (m_controller, control_*) is compile-time optional; see §2.7.

2.2 The statistics primitive — metrics<T, VT>

Every counter is a fixed-size, key-typed vector from connection_statistics.hpp:

// metrics -- src/connection/connection_statistics.hpp
template <class T, typename VT = std::uint64_t>
class metrics {
    inline metrics<T, double> operator- (const metrics &other); /* <- delta is double */
    inline metrics<T, double> operator* (double multiplier);
  private:
    VT m_values[static_cast<std::size_t> (T::STATS_COUNT)];      /* <- one slot per key */
};

T is an enum class whose last real value STATS_COUNT sizes the array — no per-key allocation. Two vocabularies instantiate it:

Key enum	Members (each /* count */ unless noted)
statistics::context	BYTES_IN_TOTAL, BYTES_OUT_TOTAL, OPEND_NS, LAST_ACTIVE_NS, LAST_MOVED_NS, MOVE_COUNT, RECV_BUDGET_HIT, SEND_BUDGET_HIT
statistics::worker	PACKET_COUNT, CLIENT_NUM, MQ_REQUESTED, MQ_COMPLETED, MQ_NEW_CLIENT, MQ_HANDOFF_CLIENT, MQ_TAKEOVER_CLIENT, MQ_SHUTDOWN_CLIENT, MQ_SEND_PACKET, MQ_RELEASE_PACKET, BLOCKED_RMUTEX (us)

worker has a trailing NA past STATS_COUNT (never allocated); worker_to_string[] names each worker key for printing.

INVARIANT — raw/derived type split. Counters collect as VT = uint64_t, but operator-/operator* return metrics<T, double>. The coordinator keeps two copies of a worker’s counters — a uint64 “previous” snapshot and a double EWMA accumulator — because the type system forbids storing a rate in a raw slot, which is what would overflow the rate math or lose fractional decay.

2.3 worker_statistics — the per-worker scorecard

m_statistics is a std::vector<worker_statistics>, one entry per connection worker — the coordinator’s model of that worker’s load.

// worker_statistics -- src/connection/coordinator.hpp
struct worker_statistics {
  double m_score;                 /* score */
  double m_core;  uint64_t m_last_cpu_time;         /* resource */
  uint32_t m_client_num;  uint64_t m_last_updated;  /* immediate */
  statistics::metrics<statistics::context, double> m_sum;   /* sum of contexts */
  std::pair<statistics::metrics<statistics::worker, double>,
            statistics::metrics<statistics::worker>> m_worker;  /* accumulated , previous */
  std::unordered_map<uint64_t,
      std::pair<statistics::metrics<statistics::context, double>,
                statistics::metrics<statistics::context>>> m_contexts;
};

Field	Role	Why it exists
m_score	The single scalar placement/scaling compares across workers	Reduces a load vector to one orderable number (Ch 6)
m_core	Fractional CPU-core share attributed to this worker	Normalizes score by CPU actually received
m_last_cpu_time	Last CPU-time sample (ns)	Next update computes a CPU-time delta, not an absolute
m_client_num	Immediate connection count	Fast exact placement input without walking m_contexts
m_last_updated	Monotonic ns of the last fold	The time_delta denominator for EWMA
m_sum	Per-update double sum of all context metrics	Aggregate view without iterating the map
m_worker	first = EWMA-accumulated worker counters; second = previous raw snapshot	The raw/derived pair (§2.2) at worker level
m_contexts	Map: connection id → per-connection (accumulated, previous) pair	Same pair per live connection, so one hot client is visible

INVARIANT — slot index equals worker id. m_statistics[i] is worker i; the vector is index-parallel to m_current_worker (§2.9). Every score lookup, rebalance source/target, and CLIENT_MOVE target indexes it — if it diverges from the live worker set, placement writes a stale slot and the model rots.

INVARIANT — accumulate .first, replace .second. The EWMA update (Ch 6/11) folds current - m_worker.second into m_worker.first scaled by the delta, then sets m_worker.second = current. .first is only decayed, .second only replaced. m_contexts pairs obey the same rule per connection.

2.4 The lifecycle enum — status

m_status holds one coordinator::status, gating which timer-driven action is legal per phase so the coordinator never drains while still preparing.

status	Meaning
PREPARING	Bring-up before serving; timers not yet armed
STABLE	Steady state; statistics + rebalancing run, no scale change in flight
DRAINING	Scale-down underway; one worker sheds its connections
EXPANDING	Scale-up underway; a new worker is being wired in

Transitions are the scaling algorithm (Ch 6); m_status is just its discriminator.

2.5 The auto-scaling window — trial state

Auto-scaling is a trial-and-observe controller: nudge the worker count, watch whether the aggregate score improves, commit or revert. Two enums, one sample record, and two anonymous members hold the window.

// scaling enums + record -- src/connection/coordinator.hpp
enum class scaling_status    { TRIAL, STABLE };
enum class scaling_direction { DOWN, UP };
struct scaling_statistics { std::size_t scale; double score; };  /* one (count,score) sample */

scaling_statistics is a single trial data point — one dot on the score-vs-workers curve the verdict fits (Ch 6):

Field	Role	Why it exists
scale (std::size_t)	Worker count at this sample	The x-axis of a trial data point
score (double)	Aggregate pool score measured at that worker count	The y-axis a verdict compares across samples

The window that collects those samples is an anonymous member:

// m_scaling_statistics -- src/connection/coordinator.hpp
struct {
  scaling_status status;  std::size_t window_size;
  std::vector<scaling_statistics> history;
  scaling_direction previous_direction;  std::size_t previous_scale;
  scaling_direction direction;  std::size_t count;
} m_scaling_statistics;

Field	Role	Why it exists
status	TRIAL (measuring a change) vs STABLE (settled)	Distinguishes measuring from settled
window_size	Samples a trial averages over	Smooths noisy per-tick scores before judging
history	Bounded buffer of scaling_statistics samples	The evidence a verdict is computed from
previous_direction	Direction of the last committed move	Detects oscillation / back-off
previous_scale	Worker count before the current trial	Baseline the trial’s score compares against
direction	Direction of the trial under evaluation	The hypothesis under test this window
count	Samples collected this window	Trips the verdict at window_size

The mechanics of a move (not its statistics) live in a second anonymous struct:

// m_scaling -- src/connection/coordinator.hpp
struct { uint64_t last_drain_ns; uint64_t last_expand_ns; int draining_worker; } m_scaling;

Field	Role	Why it exists
last_drain_ns	Monotonic ns of the last scale-down	Rate-limits shrink; prevents downward thrash
last_expand_ns	Monotonic ns of the last scale-up	Rate-limits growth symmetrically
draining_worker	Index of the worker shedding, or negative sentinel	Ensures at most one drain in flight

INVARIANT — at most one worker drains at a time. While draining_worker is a valid index (and m_status == DRAINING) no second drain starts, so two workers never race for the same target slots and corrupt m_migrating (§2.9).

2.6 The timer wheel — timer_handle and m_timer_handler

One timerfd drives all periodic work; the coordinator dispatches through a fixed table rather than one fd per task.

// timer types -- src/connection/coordinator.hpp
enum class timer_latency : uint64_t {
  NA = 0,                                           /* off */
  LOW_LATENCY    = static_cast<uint64_t>(1  * 1e9), /* 1 sec */
  MEDIUM_LATENCY = static_cast<uint64_t>(5  * 1e9), /* 5 sec */
  HIGH_LATENCY   = static_cast<uint64_t>(60 * 1e9)  /* 1 min */
};
enum class timer_type : uint32_t { NA, STATISTICS, REBALANCING, SCALING, TYPE_COUNT };
struct timer_handle { bool valid; timer_latency latency; std::function<bool ()> function; uint64_t last_time; };

m_timer_handler is std::array<timer_handle, TYPE_COUNT>, indexed by timer_type.

timer_handle field	Role	Why it exists
valid	Is this slot armed?	Register/remove without resizing the array
latency	Desired period (ns timer_latency)	The interval the handler wants
function	bool() callback (stats/rebalance/scale pass)	Type-erased for uniform dispatch; returns continue/ok
last_time	Monotonic ns of last run	Fires only when now - last_time >= latency

Slots: STATISTICS folds samples into m_statistics/m_task_statistics; REBALANCING emits a CLIENT_MOVE; SCALING runs one trial step (all Ch 6/11).

INVARIANT — one fd, many logical timers. m_timens is always the shortest active period; each handler self-throttles via last_time. Arm too slowly and a handler starves; drop the last_time check and every handler fires every tick.

2.7 The control channel — controller + control_* (compile-time optional)

Callout — guarded subsystem. In the PR #7323 worktree the entire control cluster — control_type / control_recv / control_send, the m_controller and m_ctrlfd fields (§2.9), and the handle_controller* methods — is wrapped in #if defined (ENABLE_CONTROLLER) and is not compiled in a default build. Everything in this section exists only under that macro.

Statistics and messages flow into the coordinator. The control channel is the reverse: imperative commands out to a specific worker plus a reply, over an AF_UNIX SOCK_DGRAM socket wrapped by controller<RX, TX>.

// controller -- src/connection/controller.hpp
template <typename RX, typename TX>
class controller {
    bool open (std::string path, int flags);   /* asserts SOCK_NONBLOCK */
    result recv (RX &data, sockaddr_un &peer, socklen_t &peerlen);
    result send (TX &data, sockaddr_un &peer, socklen_t &peerlen);
  private:
    int m_ctrlfd;  std::string m_path;
};

The coordinator holds controller<control_recv, control_send> m_controller (plus the cached m_ctrlfd for its epoll set). Payloads are fixed-size PODs:

// control_type / control_recv / control_send -- src/connection/coordinator.hpp
#if defined (ENABLE_CONTROLLER)
enum class control_type : uint32_t {
  SHOW_STATS, SCALE_UP, SCALE_DOWN, CLIENT_MOVE,  /* RECV: coordinator -> worker */
  OK, NOK,                                        /* SEND: worker -> coordinator */
  TYPE_COUNT
};
struct control_recv { control_type type; int from; int to; int id; };
struct control_send { control_type type; };
#endif

Naming is from the worker’s view: control_recv = command received, control_send = ack returned.

control_recv field	Role	Why it exists
type	SHOW_STATS / SCALE_UP / SCALE_DOWN / CLIENT_MOVE	Discriminates the four imperatives
from	Source worker index	CLIENT_MOVE: who gives up the connection
to	Target worker index	CLIENT_MOVE: who takes it over
id	Connection id to act on	Identifies the specific connection

control_send field	Role	Why it exists
type	The OK/NOK verdict returned by the worker	No other payload is needed — the coordinator already knows which command it issued

INVARIANT — a control datagram is exactly sizeof(RX)/sizeof(TX). recv/send return result::Error unless recvfrom/sendto moved exactly the struct size; a truncated datagram is never partially applied. SOCK_DGRAM + SOCK_NONBLOCK preserves message boundaries and returns result::Pending (shared result enum, buffer.hpp) instead of blocking the lone coordinator thread — hence datagrams here, not the byte-stream reactor of Chapter 1.

2.8 The message inbox — message_type and message

The high-volume async path into the coordinator is the message queue: new client, contexts returning, hand-off result, statistics batch, shutdown.

// message_type -- src/connection/coordinator.hpp
enum class message_type { START, NEW_CLIENT, RETURN_TO_POOL, HANDOFF_REPLY, STATISTICS, SHUTDOWN, TYPE_COUNT };

message is a move-only tagged struct (copies = delete, moves defaulted); type selects which fields are live. It is not a union — all fields coexist — so type is the reader’s contract, not the compiler’s.

// message -- src/connection/coordinator.hpp
struct message {
  message (const message &) = delete;              /* <- move-only: owns conn / vectors */
  message (message &&) noexcept = default;
  message_type type;
  css_conn_entry *conn;                            /* NEW_CLIENT */
  std::vector<context *> resource;                 /* RETURN_TO_POOL */
  bool transferred; int from; int to; uint64_t id; /* HANDOFF_REPLY */
  struct {                                         /* STATISTICS */
    uint64_t cpu_time_ns; uint64_t time_ns;
    std::pair<std::size_t, statistics::metrics<statistics::worker>> worker;
    std::vector<std::pair<uint64_t, statistics::metrics<statistics::context>>> contexts;
  } statistics;
};

Role matrix — which fields are live per type (START/SHUTDOWN carry none):

Field	NEW_CLIENT	RETURN_TO_POOL	HANDOFF_REPLY	STATISTICS
conn	accepted css_conn_entry*	–	–	–
resource	–	contexts to recycle	–	–
transferred/from/to/id	–	–	move result + endpoints	–
statistics.cpu_time_ns/time_ns	–	–	–	sample window
statistics.worker	–	–	–	(worker idx, raw worker metrics)
statistics.contexts	–	–	–	(conn id, raw context metrics)[]

The statistics.* metrics arrive raw (uint64); the coordinator folds them into the double EWMA accumulators of §2.3 on receipt — the raw/derived boundary is exactly this queue.

Figure 2-2 — the message inbox.

flowchart LR
  A["accept path"] -->|NEW_CLIENT| Q
  W["connection worker"] -->|HANDOFF_REPLY / STATISTICS / RETURN_TO_POOL| Q
  P["pool / shutdown"] -->|START / SHUTDOWN| Q
  Q["m_queue<br/>tbb::concurrent_queue&lt;message&gt;<br/>+ m_queue_size (atomic)"]
  Q -->|single consumer| C["coordinator thread<br/>handle_message_queue"]

// m_queue / m_queue_size -- src/connection/coordinator.hpp
tbb::concurrent_queue<message> m_queue;   /* multi-producer, single-consumer */
std::atomic<uint64_t> m_queue_size;

INVARIANT — MPSC discipline + bounded drain. Producers enqueue from anywhere; consumption is only from the coordinator thread. m_queue_size bounds one drain pass to the messages present when it started, so a steady producer cannot pin the consumer and starve the timer/control work sharing the epoll iteration. Because message is move-only, enqueue(message &&) transfers ownership of the embedded conn and vectors — the producer must not touch them after std::move.

2.9 The remaining coordinator fields

Identity, reactor plumbing, and worker accounting, for completeness.

Field	Type	Role
m_parent	pool *	Back-pointer to the owning connection pool
m_watcher	shared_ptr<thread_watcher>	Liveness/health registration
m_thread	std::thread	The coordinator’s own OS thread
m_core	std::size_t	CPU core the thread is pinned to
m_status	status	Lifecycle phase (§2.4)
m_stop	bool	Set by SHUTDOWN to break the run loop
m_entry	cubthread::entry *	Per-thread engine context (companion doc)
m_events	cubsocket::epoll	The coordinator’s own epoll set
m_eventfd	int	Event-based wakeup fd (queue/control)
m_timerfd	int	The single timer fd (§2.6)
m_timens	uint64_t	Current timerfd interval (ns)
m_ctrlfd	int	Cached m_controller fd — guarded, ENABLE_CONTROLLER only (§2.7)
m_max_worker / m_min_worker	uint32_t	Scale ceiling / floor
m_current_worker	uint32_t	Live worker count; parallels m_statistics.size()
m_migrating	unordered_set<uint64_t>	Connection ids currently in a hand-off/take-over
m_controller	controller<control_recv, control_send>	Outbound control channel — guarded, ENABLE_CONTROLLER only (§2.7)
m_task_statistics	anon struct	Pool-wide throughput EWMA (below)
m_statistics	vector<worker_statistics>	Per-worker scorecards (§2.3)

INVARIANT — m_migrating brackets an in-flight move. An id is inserted when CLIENT_MOVE is issued and erased on the matching HANDOFF_REPLY; while present it is excluded from placement/rebalance, so no connection is handed to two targets at once. The reply id must match, or an id leaks and that connection is frozen out of all future rebalancing.

m_task_statistics is the pool-wide analogue of §2.3, one struct for the whole pool:

// m_task_statistics -- src/connection/coordinator.hpp
struct {
  std::uint32_t workers; uint64_t time_ns;
  std::pair<double, uint64_t> requested;   /* first: EWMA rate, second: previous raw */
  std::pair<double, uint64_t> started;
  std::pair<double, uint64_t> completed;
  std::pair<double, uint64_t> depth;
} m_task_statistics;

Field	Role
workers / time_ns	Worker count + timestamp the last sample covered (EWMA normalizer/denominator)
requested	(rate, prev) of tasks handed to the back-end pool — scale-up pressure
started	(rate, prev) of tasks the pool began — pickup rate
completed	(rate, prev) of tasks finished — throughput
depth	(rate, prev) of queue depth — backlog signal

Each pair is the §2.2 split (first = decayed double rate, second = previous raw uint64); these pool-wide rates drive the back-end elastic pool (Ch 12–14) and the auto-scaler (Ch 6).

2.10 Chapter summary — key takeaways

1. The coordinator owns six clusters: per-worker scorecards (m_statistics), the auto-scaling trial window (m_scaling_statistics + m_scaling), the timer table (m_timer_handler), the outbound control channel (m_controller), the inbound message queue (m_queue), and pool-wide throughput (m_task_statistics).
2. Every counter is a metrics<T, VT> fixed array kept as a raw uint64 previous plus a double EWMA accumulator; the type system stops you mixing them.
3. m_statistics[i] is worker i — index-parallel to the live worker set, and every placement/rebalance/scale decision indexes it.
4. The control channel is compile-time optional (ENABLE_CONTROLLER): when present it flows commands out as fixed-size AF_UNIX datagrams; messages always flow in on an MPSC tbb::concurrent_queue, drained bounded by m_queue_size.
5. message is a move-only tagged struct: type selects live fields, and moving it transfers ownership of the embedded conn and vectors.
6. Two membership/count invariants keep moves safe: at most one draining_worker at a time, and an id sits in m_migrating for exactly the span of its hand-off.
7. This chapter is structure only — the score math, rebalancing selection, and trial scaling that read it are Chapter 6; the back-end pool consuming m_task_statistics is Chapters 9–14.

Chapter 3: Initialization and Memory

The reader question: from main() to a fully wired reactor, who allocates what, on which thread, and in what order? The companions explain why the reactor exists (cubrid-thread-manager.md — “Connection worker (CBRD-26212)”, “Coordinator + context freelist (CBRD-26407)”) and what a worker pool is (cubrid-thread-worker-pool.md). This chapter is bring-up mechanics: every new, every std::thread, every epoll_ctl, and every teardown branch.

Two object graphs are stood up back to back inside css_init (server_support.c):

• the back-end request pool — css_Server_request_worker_pool, a cubthread::stats_worker_pool_type built by thread_create_stats_worker_pool. This is the task engine dissected in Ch 9-11.
• the connection reactor — a stack-local cubconn::connection::pool connections, brought up by pool::initialize. This is the epoll front end of Ch 1-8.

Both draw their thread entries from one pre-sized pool that boot allocated earlier. We trace the entry pool first, then the two graphs, then teardown.

3.1 The bring-up call chain — main to css_init

main (server.c) registers fatal-signal handlers, saves the executable path, setsid()s, then calls net_server_start (database_name) and returns its status. All reactor bring-up is downstream.

// main -- src/executables/server.c  (condensed)
  register_fatal_signal_handler (SIGSEGV); /* ... SIGILL/FPE/BUS/SYS/ABRT ... */
  database_name = argv[1];
  ret_val = net_server_start (database_name);   /* <- never returns until shutdown */

net_server_start (network_sr.c) is the real orchestrator. Ordered: er_init, cubthread::initialize (thread_p) (creates the manager singleton and the main entry, §3.2), sysprm_load_and_init, css_initialize_server_interfaces (net_server_request), then boot_restart_server (which runs recovery and calls cubthread::initialize_thread_entries to size and allocate the entry pool, §3.2), and only on success css_init (thread_p, packed_name, name_length, TCP_PORT_ID). So by the time css_init runs, the DB is recovered and the entry pool exists; css_init only builds the two thread graphs on top.

flowchart TD
  M["main -- server.c"] --> NS["net_server_start -- network_sr.c"]
  NS --> TI["cubthread::initialize<br/>create manager + main entry"]
  NS --> BR["boot_restart_server<br/>recovery + initialize_thread_entries"]
  NS --> CI["css_init -- server_support.c"]
  CI --> WP["thread_create_stats_worker_pool<br/>back-end request pool"]
  CI --> RP["connections.initialize<br/>connection reactor"]
  CI --> RUN["connector.run port name<br/>blocks until shutdown"]

Figure 3-1. From process entry to the two thread graphs. net_server_start sequences entry-pool sizing before css_init builds the pools.

3.2 Sizing the thread universe — the entry pool

Thread entries (cubthread::entry, Ch 2 of the companion’s data map) are heavyweight; the manager pre-allocates a fixed array and hands them out by claim/retire rather than constructing per thread. The count is derived, not configured, by manager::set_max_thread_count_from_config:

// manager::set_max_thread_count_from_config -- src/thread/thread_manager.cpp
  m_max_threads = cubbase::count_registry<connection>::total () + cubbase::count_registry<worker_pool>::total () +
                  cubbase::count_registry<daemon>::total () + 1 /* PAD */;

Each subsystem registers its demand at static-init time via macros. The reactor contributes two lines: REGISTER_CONNECTION (coordinator, 1) (coordinator.cpp) reserves one entry, and REGISTER_CONNECTION (connection_worker, [](){ return prm_get_integer_value (PRM_ID_CSS_MAX_CONNECTION_WORKER); }) (connection_worker.cpp) reserves max_connection_worker entries — the getter is evaluated when count_registry::total() sums the registry. The back-end pool registers through REGISTER_WORKERPOOL (transaction, ...PRM_ID_TASK_WORKER). initialize_thread_entries then calls set_max_thread_count_from_config followed by alloc_entries:

// manager::alloc_entries -- src/thread/thread_manager.cpp  (SERVER_MODE branch)
  m_available_entries_count = m_max_threads;
  m_all_entries = new entry[m_max_threads];                       /* the entire entry pool, one shot */
  m_entry_dispatcher = new entry_dispatcher (m_all_entries, m_max_threads);

init_lockfree_system sizes the lock-free transaction system to m_max_threads + 1, and init_entries stamps each entry’s index and assigns an LF-tran index. From then on every reactor thread pulls its context with manager::claim_entry (which stores it in tl_Entry_p) and returns it with retire_entry.

INVARIANT — reserved entries equal registered demand. m_available_entries_count starts at m_max_threads and is debited by create_and_track_resource (§3.3) exactly by the count each pool reserved via its REGISTER_* macro. The coordinator’s claim_entry and each worker’s claim_entry consume from the same array. If PRM_ID_CSS_MAX_CONNECTION_WORKER changed between registration and pool::initialize, the reactor would create more workers than entries reserved and claim_entry would eventually return nullptr — which every caller treats as assert_release (false).

3.3 The back-end request pool — create_worker_pool

css_init builds the task engine first, through the inline helper thread_create_stats_worker_pool which forwards to the manager template:

// manager::create_worker_pool -- src/thread/thread_manager.hpp  (SERVER_MODE)
  static_assert (std::is_base_of_v<worker_pool, Res>);
  workerpool = create_and_track_resource<Res> (m_worker_pools, pool_size,           /* reserve pool_size entries */
               pool_size, core_count, std::forward<CtArgs> (args)...);
  if (workerpool)
    { workerpool->initialize (pool_size, core_count); }                             /* <- only if reservation succeeded */
  return workerpool;

create_and_track_resource is the gatekeeper. Under m_entries_mutex, if m_available_entries_count < entries_count it returns NULL without constructing anything (this is how an over-subscribed thread budget surfaces); otherwise it debits the count, new Res (...)s the resource, and pushes it onto m_worker_pools.

Res here is stats_worker_pool_type = worker_pool_impl<true> — the true template argument selects the atomic-stats specialization (Ch 11). Its ctor assert (core_count > 0 && core_count <= pool_size) and records m_max_workers = pool_size, then initialize splits the pool into cores: allocate_cores (core_count) + assign_workers_to_cores (worker_count) (the per-core thread spin-up is Ch 9-10). Back in css_init, a NULL return is fatal:

// css_init -- src/connection/server_support.c
  if (css_Server_request_worker_pool == NULL)
    { assert (false); er_set (ER_FATAL_ERROR_SEVERITY, ARG_FILE_LINE, ER_GENERIC_ERROR, 0);
      status = ER_FAILED; goto shutdown; }   /* <- skips reactor bring-up, jumps to teardown */

pool_size = PRM_ID_TASK_WORKER, core_count = PRM_ID_TASK_GROUP; pooling and idle-timeout come from css_get_server_request_thread_pooling_configuration / _timeout_configuration. Ownership: the raw pointer lives in the file-static css_Server_request_worker_pool and is destroyed by destroy_worker_pool at the tail of css_init.

3.4 connection::pool::initialize — the reactor spine

With the task engine up, css_init calls connections.initialize (MAX_CONNECTIONS, max_connection_workers, min_connection_workers). MAX_CONNECTIONS is css_get_max_conn () + 1. Branch-complete:

// pool::initialize -- src/connection/connection_pool.cpp
  (void) os_set_signal_handler (SIGPIPE, SIG_IGN);                       /* 1. writes to dead sockets must not kill us */
  (void) os_set_signal_handler (SIGFPE, SIG_IGN);
  max_connection_workers = this->initialize_topology (max_connection_workers);   /* 2. clamp to real cores */
  if (min_connection_workers > max_connection_workers)                  /* 3. floor cannot exceed ceiling */
    { min_connection_workers = max_connection_workers; }
  this->lock_resource ();                                               /* 4. take the pool baton */
  this->initialize_freelist (max_connections);                         /* 5. context slab */
  this->initialize_coordinator (max_connection_workers, min_connection_workers); /* 6. spawn coordinator thread */
  this->initialize_workers (max_connection_workers, min_connection_workers);     /* 7. spawn N worker threads */
  this->release_resource ();                                           /* 8. hand the baton off */
  this->start_coordinator ();                                          /* 9. ping the coordinator */
  m_max_connections = max_connections; /* ...record the three sizes... */

Step 2, initialize_topology, is where max_connection_workers becomes physical. It reads os::resources::cpu::effective () — whose adjusted_effective is an optional core-id vector and adjusted_max the usable core count — takes the first min(effective.size(), max_connection_workers) cores, maps NICs onto them (net::map_nic_to_index), and returns min(adjusted_max, max_connection_workers). So the configured max is silently reduced to the number of cores the process may actually run on.

flowchart TD
  A["ignore SIGPIPE/SIGFPE"] --> B["initialize_topology<br/>clamp workers to adjusted_max"]
  B --> C{"min > max?"}
  C -- yes --> D["min = max"]
  C -- no --> E["lock_resource<br/>take pool mutex"]
  D --> E
  E --> F["initialize_freelist<br/>slab of contexts"]
  F --> G["initialize_coordinator<br/>make_shared spawns thread"]
  G --> H["initialize_workers<br/>N unique_ptr workers + prewarm"]
  H --> I["release_resource<br/>baton to coordinator"]
  I --> J["start_coordinator<br/>enqueue START + notify"]

Figure 3-2. pool::initialize, every step. Steps 4-8 are one critical section on the css_init thread.

INVARIANT — m_mutex is a baton, not a scoped lock. pool::initialize calls lock_resource() (step 4) but the matching release_resource() (step 8) does not end the story: coordinator::initialize (§3.6) re-acquires the same mutex at its tail and holds it for the coordinator’s entire lifetime, releasing it only in coordinator::finalize. The setup section runs on the css_init thread; the moment it releases, the freshly spawned coordinator thread (blocked in its own lock_resource) grabs the mutex and owns it forever. This is why claim_context/retire_context/get_workers all assert (m_mutex_holder == std::this_thread::get_id ()) — they are only ever called from the coordinator thread, which provably holds the baton. Violating the handoff (e.g. a second thread calling claim_context) trips the assert.

3.5 initialize_freelist — the context slab

The freelist is where “every connection object” is pre-allocated so the steady-state path never mallocs. Its node embeds a full context:

// pool::freelist -- src/connection/connection_pool.hpp
  struct freelist { context m_context; freelist *m_next;
                    freelist (std::size_t capacity) : m_context (capacity), m_next (nullptr) {} };

Field	Role	Why it exists
m_context	embedded per-connection context, constructed with a capacity byte buffer	embedding it in the node means one allocation per connection, not two
m_next	intrusive free-chain link	O(1) push/pop while the pool baton is held

The pool tracks the chain in an anonymous struct:

Field	Role	Why it exists
m_head	LIFO head of the free chain	claim_context pops it, retire_context pushes onto it
m_max	high-water cap = max_connections * 1.1	retire_context deletes nodes past this instead of hoarding memory
m_claim	count of outstanding (claimed) contexts	must be 0 at finalize; the assert catches context leaks

// pool::initialize_freelist -- src/connection/connection_pool.cpp
  m_freelist.m_head = nullptr; m_freelist.m_claim = 0;
  m_freelist.m_max = static_cast<std::size_t> (static_cast<float> (max_connections) * /* margin */ 1.1);
  for (i = 0; i < m_freelist.m_max; i++)
    { head = m_freelist.m_head; m_freelist.m_head = new freelist (32 * 1024);   /* 32 KB context buffer each */
      m_freelist.m_head->m_next = head; }

Every node’s context is sized 32 * 1024 bytes — the same literal reused in claim_context’s overflow path, where an empty freelist mints a fresh new freelist (32 * 1024) rather than blocking. So the 10% margin is a soft target, not a hard cap: exceed it and claim_context allocates on demand, and retire_context frees back down to m_max.

INVARIANT — m_claim == 0 at teardown. finalize_freelist opens with assert (m_freelist.m_claim == 0). Every claim_context does m_claim++, every retire_context does m_claim--; a non-zero count at shutdown means a connection context was claimed and never returned to the coordinator’s RETURN_TO_POOL path (Ch 7). The assert converts that leak into a debug-build crash at the exact shutdown boundary.

3.6 initialize_coordinator — spawning the epoll coordinator

initialize_coordinator picks the coordinator’s home core (adjusted_effective[0], else 0) and make_shared<coordinator> — the ctor does the heavy lifting on the css_init thread, then spawns the coordinator thread:

// coordinator::coordinator -- src/connection/coordinator.cpp  (condensed)
  m_controller.open ("/tmp/cub_server_" + std::to_string (getpid ()) + "_coordinator.sock", SOCK_NONBLOCK | SOCK_CLOEXEC);
  m_ctrlfd = m_controller.get_fd ();
  m_eventfd = eventfd (0, EFD_NONBLOCK | EFD_CLOEXEC);          /* wake-on-message */
  m_timerfd = timerfd_create (CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
  if (m_eventfd < 0 || m_timerfd < 0) { assert_release (false); }
  if (!this->eventfd_register (m_eventfd) || !this->eventfd_register (m_timerfd) ||
      !this->eventfd_register (m_ctrlfd)) { assert_release (false); }   /* all three into epoll */
  /* three timers: STATISTICS/REBALANCING/SCALING via eventfd_addtimer */
  m_thread = std::thread (&coordinator::attach, this);          /* <- the coordinator thread begins */

eventfd_register adds each fd to the coordinator’s private cubsocket::epoll m_events in edge-triggered read mode:

// coordinator::eventfd_register -- src/connection/coordinator.cpp
  if (!m_events.add_descriptor (fd, EPOLLET | EPOLLIN)) { return false; }

The epoll wrapper (epoll.cpp) is a thin RAII shell: its ctor is m_epoll = epoll_create1 (EPOLL_CLOEXEC) (asserting on -1), and add_descriptor packs an epoll_event whose data.ptr is the caller’s cookie (or data.fd when none) before epoll_ctl (EPOLL_CTL_ADD). The coordinator registers bare fds (cookie nullptr → data.fd); §3.7 shows the worker registers context pointers instead.

The spawned thread runs coordinator::attach = initialize(); run(); finalize();. coordinator::initialize is where the coordinator claims its OS identity and grabs the baton:

// coordinator::initialize -- src/connection/coordinator.cpp
  m_watcher->mtx.lock (); m_watcher->active++; m_watcher->mtx.unlock ();   /* announce liveness */
  pthread_setname_np (pthread_self (), "coordinator");
  os::resources::cpu::setaffinity (m_core);                               /* pin to home core */
  m_entry = cubthread::get_manager ()->claim_entry ();                    /* pull a thread entry */
  if (m_entry == nullptr) { assert_release (false); }
  m_entry->register_id (); m_entry->type = TT_SERVER; m_entry->tran_index = -1;
  m_entry->m_status = cubthread::entry::status::TS_RUN; m_entry->shutdown = false;
  m_entry->get_error_context ().register_thread_local ();
  m_status = status::STABLE;
  m_parent->lock_resource ();                                             /* <- takes the baton for life */

m_watcher is a thread_watcher { std::mutex mtx; std::condition_variable cv; int active; } shared (via shared_ptr) by pool, coordinator, and every worker. active is the count of live reactor threads; the pool’s finalize waits on cv for it to drain.

graph TD
  P["pool<br/>owns via value"] -->|shared_ptr| C["coordinator<br/>m_thread + epoll m_events"]
  P -->|"vector unique_ptr"| W["worker[0..N-1]<br/>each m_thread + epoll m_events"]
  P -->|shared_ptr| WA["thread_watcher<br/>mtx cv active"]
  C -->|shared_ptr| WA
  W -->|shared_ptr| WA
  C -->|shared_ptr held by workers| C2["coordinator<br/>(for handoff msgs)"]

Figure 3-3. Ownership after pool::initialize. The coordinator is shared_ptr (workers hold it for handoff); workers are unique_ptr owned only by the pool; the watcher is shared by all.

3.7 initialize_workers — connection worker creation and pre-warm

initialize_workers reserves the vector, resolves the core list, and constructs exactly max_connection_workers workers, one per core:

// pool::initialize_workers -- src/connection/connection_pool.cpp
  m_workers.reserve (max_connection_workers);
  cores = ctx.adjusted_effective ? *ctx.adjusted_effective : /* iota 0..adjusted_max */ ...;
  assert (cores.size () >= max_connection_workers);
  for (i = 0; i < max_connection_workers; i++)
    { m_workers.emplace_back (std::make_unique<worker> (this, m_coordinator, m_watcher, cores[i], i)); }

Each worker ctor (connection_worker.cpp) mirrors the coordinator’s: it reserves m_context, reads the RECV/SEND_BUDGET params, creates its own eventfd/timerfd, registers them, installs three timers (HIBERNATE/STATISTICS/HA), and spawns m_thread = std::thread (&worker::attach, this). The worker’s eventfd_register differs from the coordinator’s — it wraps each fd in a fake context so the run loop can treat wakeups uniformly with real connections:

// worker::eventfd_register -- src/connection/connection_worker.cpp
  ctx = new context ();
  conn = reinterpret_cast<css_conn_entry *> (new int { fd });   /* a heap int masquerading as a conn */
  ctx->m_conn = conn;
  if (!m_events.add_descriptor (fd, EPOLLET | EPOLLIN, ctx)) { delete ctx; delete conn; return false; }

The cookie is the context*; worker::run reads events[i].data.ptr back as a context* and compares ctx->m_conn->fd against m_eventfd/m_timerfd to route wakeups vs. client I/O (Ch 4). worker::initialize (on the worker thread) does the same watcher/affinity/claim_entry sequence as the coordinator, minus the baton, and clears m_context/m_removed_context.

After all workers are constructed, initialize_workers pre-warms every queue to dodge a startup race:

// pool::initialize_workers -- src/connection/connection_pool.cpp  (prewarm)
  for (std::unique_ptr<worker> &worker : m_workers)
    for (i = 0; i < static_cast<std::size_t> (worker::queue_type::TYPE_COUNT); i++)
      { worker::message request; request.type = worker::message_type::START;
        if (!worker->enqueue_and_notify ((worker::queue_type) i, std::move (request), nullptr, -1 /* infinite */))
          { assert_release (false); } }

enqueue_and_notify with -1 blocks until the worker acknowledges, so every worker’s IMMEDIATE and LAZY queue is exercised before any real client arrives.

INVARIANT — one worker per core, and all max are created up front. assert (cores.size () >= max_connection_workers) guarantees the topology (§3.4) supplied enough cores; worker i is pinned to cores[i]. Note that all max_connection_workers are constructed here regardless of min — the coordinator’s m_current_worker is seeded to max_worker in its ctor, so the reactor boots fully expanded and the auto-scaler (Ch 6) later drains workers down toward min. min_connection_workers is only the floor for scale_down; it never limits the initial allocation.

3.8 start_coordinator — the wake-up handshake

The coordinator thread is already looping in run by now (spawned in §3.6). start_coordinator merely posts a START message and rings the eventfd:

// pool::start_coordinator -- src/connection/connection_pool.cpp
  request.type = coordinator::message_type::START;
  m_coordinator->enqueue (std::move (request));
  if (!m_coordinator->notify ()) { assert_release (false); }

enqueue pushes onto the coordinator’s TBB queue and bumps m_queue_size (release order); notify writes 1 to the eventfd, which the epoll loop already registered. handle_message_queue_start is a no-op returning true — START exists so the loop makes one full pass and the handler table (indexed by message_type) is proven wired before NEW_CLIENT traffic. After this returns, css_init attaches the connector and calls connector.run, which blocks until shutdown.

3.9 Teardown — pool::finalize and the reverse baton

Shutdown runs connections.finalize () on the css_init thread. It reverses bring-up but must repatriate the mutex baton from the coordinator. Branch-complete:

// pool::finalize -- src/connection/connection_pool.cpp
  this->finalize_workers ();          /* 1. SHUTDOWN every worker, wait until active == 1 */
  this->finalize_coordinator ();      /* 2. SHUTDOWN coordinator, wait until active == 0 */
  this->try_to_lock_resource ();      /* 3. reclaim the baton the coordinator just dropped */
  m_workers.clear ();                 /* 4. ~worker joins each thread */
  this->finalize_freelist ();         /* 5. assert m_claim==0; delete the chain */
  this->release_resource ();
  this->finalize_topology ();         /* no-op */
  m_max_connections = -1; /* ... */

finalize_workers posts SHUTDOWN to each worker’s IMMEDIATE queue and notifys, then blocks on the watcher condition variable with the shutdown timeout: cv.wait_for (lock, wait_for, [this]{ return m_watcher->active == 1; }) — the surviving 1 is the coordinator. Each worker thread, on receiving SHUTDOWN, runs worker::finalize: it forces every live context closed (ignore_level::IGNORE_ALL + handle_connection_close), drains the LAZY queue and purge_stale_contexts in a 1 ms loop until m_context is empty, then retire_entry, and finally active-- + cv.notify_one. If the deadline passes with active != 1, finalize_workers logs and _exit (0) — a stuck worker aborts the process rather than hanging shutdown.

finalize_coordinator then posts SHUTDOWN to the coordinator and waits for active == 0. The coordinator’s handle_message_queue_shutdown sets m_stop = true, breaking run; coordinator::finalize runs m_parent->release_resource () first (dropping the baton), then retire_entry, active--, cv.notify_one. Same _exit (0) on timeout.

Only now does the css_init thread reclaim the mutex via try_to_lock_resource — 1000 tries at 10 ms each, _exit (0) if the coordinator somehow never released it. With the baton back, m_workers.clear() destroys each unique_ptr<worker>; ~worker joins the thread and closes the worker’s eventfd/timerfd. finalize_freelist asserts m_claim == 0 and walks m_head deleting nodes. ~coordinator (when the last shared_ptr drops) joins the coordinator thread and closes its fds.

flowchart TD
  F0["finalize_workers<br/>SHUTDOWN + notify each"] --> F1{"active == 1<br/>before timeout?"}
  F1 -- no --> X1["_exit 0"]
  F1 -- yes --> F2["finalize_coordinator<br/>SHUTDOWN + notify"]
  F2 --> F3{"active == 0<br/>before timeout?"}
  F3 -- no --> X2["_exit 0"]
  F3 -- yes --> F4["try_to_lock_resource<br/>reclaim baton"]
  F4 --> F5["m_workers.clear<br/>join threads"]
  F5 --> F6["finalize_freelist<br/>assert claim==0"]

Figure 3-4. pool::finalize. Workers must reach active == 1 before the coordinator is stopped, and the baton is reclaimed only after active == 0.

INVARIANT — teardown order: workers before coordinator. The reactor stops workers first (waiting for active to fall to 1), then the coordinator (waiting for 0). The coordinator outlives the workers because worker SHUTDOWN/context-return still funnels RETURN_TO_POOL messages the coordinator must drain (Ch 7). Reversing this — killing the coordinator first — would strand the pool mutex it holds and abandon in-flight context returns, corrupting m_claim and tripping the finalize_freelist assert. The back-end request pool (css_Server_request_worker_pool) is torn down after the reactor, via destroy_worker_pool at the end of css_init, because active client tasks must finish before their task engine disappears.

3.10 Chapter summary — key takeaways

1. Bring-up is a two-graph sequence inside css_init: the back-end request pool (create_worker_pool → worker_pool_impl<true>::initialize) first, then the connection reactor (pool::initialize). net_server_start sequences DB recovery and entry-pool allocation before either.
2. The entry pool is sized by registration, not config-at-init. set_max_thread_count_from_config sums the REGISTER_CONNECTION/REGISTER_WORKERPOOL counts (coordinator = 1, connection_worker = max_connection_worker, transaction = task_worker) plus a PAD; alloc_entries makes one new entry[] array that every thread claims from.
3. The pool mutex is a baton, not a scope. pool::initialize locks it on the css_init thread and releases it so the spawned coordinator thread can grab it for life; claim_context/retire_context assert the holder is the coordinator. Teardown repatriates it via try_to_lock_resource.
4. The freelist pre-allocates max_connections * 1.1 contexts of 32 KB each; overflow allocates on demand and retire trims back to m_max. m_claim == 0 is the leak-detecting shutdown assert.
5. The coordinator and each worker build private epoll instances and register eventfd + timerfd; the coordinator also registers a control unix socket. Workers wrap each fd in a fake context (new int{fd} as a pseudo-css_conn_entry) so the run loop routes wakeups and client I/O uniformly.
6. All max workers boot up front, pinned one-per-core (assert cores.size() >= max); min_connection_worker is only the auto-scaler’s drain floor. Every queue is pre-warmed with a blocking START to close a startup race.
7. Teardown is strictly ordered: workers to active == 1, coordinator to active == 0, reclaim baton, join and free — with _exit(0) guards on every shutdown-timeout branch, and the task engine destroyed last.

Chapter 4: The epoll Connection Worker Loop

Chapter 3 built and initialized a worker: it owns an epoll set, an eventfd, a timerfd, a per-worker timer table, and two tbb::concurrent_queue<message> inboxes. worker::attach — the constructor’s std::thread entry — runs initialize (); run (); finalize ();. This chapter dissects run: once running, how does one worker thread wait for events across hundreds of client sockets and dispatch them without blocking on any one connection? The why (reactor pattern, C10K) lives in the “Reactor and the single event loop” section of cubrid-thread-worker-pool.md; this is the line-by-line how.

Invariant 4-A (reactor non-blocking rule). A worker services every connection assigned to it and blocks in exactly one place: m_events.wait () (epoll_wait) at the top of run. Every socket is edge-triggered non-blocking, so no per-connection read/write stalls the thread, and sockets are drained only up to a budget (Chapter 5). If any handler blocked on one connection, every other client on that worker would freeze.

4.1 Three event sources, one wait

A worker’s epoll set holds three descriptor kinds, all registered with epoll_event.data.ptr pointing at a context *: client sockets (EPOLLET | EPOLLIN | EPOLLRDHUP, plus EPOLLOUT when a send is pending); m_eventfd, the wakeup counter other threads poke; and m_timerfd, armed to the shortest active timer latency. The two control fds were each wrapped in a throwaway context during construction (eventfd_register does new context () and stores the raw fd via reinterpret_cast<css_conn_entry *> (new int { fd })), so the callback data is uniformly a context *. run tells them apart by comparing ctx->m_conn->fd against m_eventfd / m_timerfd.

flowchart LR
  C1["client fd 1<br/>EPOLLIN/OUT/RDHUP"] --> EP["epoll m_events"]
  C2["client fd N"] --> EP
  EF["m_eventfd<br/>(queue doorbell)"] --> EP
  TF["m_timerfd<br/>(shortest latency)"] --> EP
  EP -- "data.ptr = context*" --> RUN["worker::run demux"]
  RUN -- "eventfds[0]" --> MQ["handle_message_queue"]
  RUN -- "eventfds[1]" --> TMR["timer callbacks"]
  RUN -- "client fd" --> IO["reception / transmission"]

Figure 4-1. Every source funnels through one epoll instance; run demuxes on the fd identity carried in context.

4.2 worker::run — branch-complete

The frame: a stack array of 512 epoll_events, a two-slot bool eventfds[2] deferral latch, and while (!m_stop).

// worker::run -- src/connection/connection_worker.cpp
  nfds = m_events.wait (events.data (), events.size (),
                        m_exhausted.empty () ? TIMEOUT_INFINITE : TIMEOUT_NOWAIT); /* <- 4.2.1 */
  if (nfds < 0)
    { if (errno == EINTR) continue; assert_release (false); continue; }  /* signal: retry; else log+loop */
  m_timens = this->get_time_ns (CLOCK_MONOTONIC);              /* <- one clock read, reused all loop */
  // ... per-fd demux (4.2.2) ...
  if (m_exhausted.size () > 0) this->handle_exhausted ();      /* <- Chapter 5 */
  if (eventfds[0] || eventfds[1]) this->eventfd_handler (eventfds);

4.2.1 The timeout branch. TIMEOUT_INFINITE (-1) when m_exhausted is empty, else TIMEOUT_NOWAIT (0). m_exhausted holds contexts that hit their per-connection IO budget with work still pending; if any exist the worker must poll (return at once, possibly nfds == 0) to revisit them via handle_exhausted — Chapter 5’s machinery. Empty map → sleep until an fd is ready.

4.2.2 The per-fd demux. For each of nfds descriptors: (1) ctx = reinterpret_cast<context *> (events[i].data.ptr) (asserted non-null). (2) Hangup/error: if events[i].events carries any of EPOLLHUP | EPOLLRDHUP | EPOLLERR and the fd is neither m_eventfd nor m_timerfd, call handle_hangup_or_error (ctx, events[i].events & EPOLLERR) and continue — the control fds are excluded so a spurious HUP is not read as a client disconnect. (3) Stamp LAST_ACTIVE_NS = m_timens. (4) EPOLLIN: fd == m_eventfd → eventfds[0] = true; continue;; fd == m_timerfd → eventfds[1] = true; continue;; else client → handle_reception (ctx, false): ClosedConnection/PeerReset → continue, Error → log + return false (fatal), Ok/Pending → fall through. (5) EPOLLOUT: handle_transmission (ctx, false), same three-way handling.

The control fds are only latched inside the fd loop; their work defers to eventfd_handler after every client fd in the batch is serviced — client IO always precedes control-plane bookkeeping.

flowchart TD
  A["epoll_wait<br/>timeout = exhausted? 0 : -1"] --> B{"nfds < 0?"}
  B -- "EINTR / other" --> A
  B -- "no" --> C["m_timens = now"]
  C --> D{"for each ready fd"}
  D --> E{"HUP/RDHUP/ERR<br/>and not eventfd/timerfd?"}
  E -- yes --> F["handle_hangup_or_error; continue"]
  E -- no --> G{"EPOLLIN?"}
  G -- "fd==eventfd" --> H["eventfds[0]=true; continue"]
  G -- "fd==timerfd" --> I["eventfds[1]=true; continue"]
  G -- "client" --> J["handle_reception"]
  J -- "Closed/PeerReset" --> D
  J -- "Error" --> Z["return false"]
  G -- no --> K{"EPOLLOUT?"}
  K -- yes --> L["handle_transmission"]
  L -- "Error" --> Z
  D -- "done" --> M{"m_exhausted>0?"}
  M -- yes --> N["handle_exhausted"]
  M -- no --> O{"eventfds set?"}
  N --> O
  O -- yes --> P["eventfd_handler"]
  O -- no --> A
  P --> A

Figure 4-2. worker::run, every branch. A fatal Error from reception/transmission is the only non-m_stop exit; on return, attach proceeds to finalize.

4.3 worker::eventfd_handler — the deferred demux

Called once per iteration when either latch is set; it consumes both latches and resets m_has_retry at entry.

// worker::eventfd_handler -- src/connection/connection_worker.cpp
  m_has_retry = false;
  if (eventfds[0])
    { eventfds[0] = false;
      if (!this->eventfd_clear (m_eventfd)) return false;    /* <- drain the counter */
      if (!this->handle_message_queue ()) return false;      /* <- 4.5 */
      m_timer_handler[(size_t) timer_type::QUEUE].last_time = m_timens; }
  if (eventfds[1])
    { eventfds[1] = false;
      for (i = 0; i < timer_type::TYPE_COUNT; i++)
        { if (!m_timer_handler[i].valid) continue;
          if (m_timens - m_timer_handler[i].last_time > (uint64_t) m_timer_handler[i].latency)
            { if (!m_timer_handler[i].function ()) return false;   /* <- fire callback */
              m_timer_handler[i].last_time = m_timens; } }
      if (!this->eventfd_clear (m_timerfd)) return false;
      if (!m_has_retry) return this->eventfd_removetimer (timer_type::QUEUE); }
  return true;

Eventfd branch (eventfds[0]). eventfd_clear runs a non-blocking read loop on m_eventfd (retry EINTR, break EAGAIN) to zero the 64-bit counter — the counterpart to notify’s write. Then handle_message_queue drains both inboxes, and the QUEUE timer’s last_time is refreshed so the retry timer does not immediately re-fire.

Timerfd branch (eventfds[1]). The timerfd is armed to a single period equal to the shortest active latency (eventfd_starttimer), so on expiry the handler walks all slots and fires each whose own elapsed time exceeds its latency. Any callback returning false propagates false and kills the loop. The four slots (set in the constructor):

timer_type	Latency	Callback	Purpose
HIBERNATE	MEDIUM 1 s	hibernate_check	park an idle worker
STATISTICS	MEDIUM 1 s	statistics_metrics_to_coordinator	push metrics to coordinator
HA	HIGH 2 s	ha_close_all_connections	drop connections on HA standby
QUEUE	LOW 1 ms	handle_message_queue	fast retry for deferred close

The QUEUE-timer retry. handle_connection_close, when it must retry a client shutdown, re-enqueues SHUTDOWN_CLIENT on LAZY, sets m_has_retry = true, and adds a QUEUE timer bound to handle_message_queue. A pending retry keeps the 1 ms timer alive; the final if (!m_has_retry) eventfd_removetimer (QUEUE) tears it down once no retry is outstanding — a stuck-closing connection is polled aggressively without a permanent 1 ms wheel.

flowchart TD
  S["m_has_retry=false"] --> A{"eventfds[0]?"}
  A -- yes --> B["eventfd_clear(eventfd)"] --> C["handle_message_queue"] --> D["QUEUE last_time=now"]
  A -- no --> E{"eventfds[1]?"}
  D --> E
  E -- yes --> F["fire each valid timer<br/>whose elapsed > latency"] --> G["eventfd_clear(timerfd)"] --> H{"m_has_retry?"}
  H -- no --> I["eventfd_removetimer(QUEUE)"] --> J["return true"]
  H -- yes --> J
  E -- no --> J

Figure 4-3. eventfd_handler: queue drain, timer fan-out, QUEUE-timer teardown.

4.4 The producer side: enqueue, notify, enqueue_and_notify

The queues are MPSC — coordinator, peer workers, and the network layer push; only the owning worker pops. enqueue pushes into the selected TBB queue then bumps the count with memory_order_release, pairing with the consumer’s exchange (acquire) (4.5) to publish message contents:

// worker::enqueue -- src/connection/connection_worker.cpp
  m_queue[(size_t) type].push (std::move (item));                        /* MPSC producer push */
  m_queue_size[(size_t) type].fetch_add (1, std::memory_order_release);  /* publishes the push */

Its leading assert requires a routable target (live conn fd, live ctx fd, positive id, or a control type START/SHUTDOWN/HIBERNATE/AWAKEN). enqueue does not wake the worker — kept separate so a caller may batch pushes and notify once. notify writes 1 to m_eventfd, retrying EINTR; a short write returns false; EAGAIN (counter saturated) is treated as success (break) since the worker is already scheduled:

// worker::notify -- src/connection/connection_worker.cpp
  bytes = ::write (m_eventfd, &u /*=1*/, sizeof (u));
  if (bytes == sizeof (u)) break;
  if (bytes == 0 || (bytes > 0 && bytes < sizeof (u))) return false;  /* short write */
  if (errno == EINTR) continue;
  if (errno == EAGAIN) break;                                          /* saturated == fine */
  return false;

Invariant 4-C (notify coalescing is loss-safe). The message lives in the queue; the eventfd is only a doorbell. Because the consumer snapshots the count with exchange (0) rather than trusting the eventfd value, any number of notify writes — even a saturating one — guarantees pending messages are seen. The one forbidden order is notify before enqueue; every path enqueues first.

enqueue_and_notify combines them, with an optional synchronous wait via a message_blocker:

// worker::enqueue_and_notify -- src/connection/connection_worker.cpp
  if (wait_time)
    { handle = std::make_shared<message_blocker> (); handle->done = false;
      lock = std::unique_lock<std::mutex> (handle->m);    /* LOCK BEFORE enqueue: no lost wakeup */
      item.waiter_handle = handle; }
  this->enqueue (type, std::move (item));
  if (!this->notify ()) { if (func) func (); return false; }
  if (func) func ();                                       /* post-notify hook, runs on success too */
  if (wait_time)
    { if (wait_time < 0) handle->cv.wait (lock, [&]{ return handle->done; });
      else handle->cv.wait_for (lock, std::chrono::seconds (wait_time), [&]{ return handle->done; }); }

wait_time is asserted to only START, SHUTDOWN_CLIENT, SEND_PACKET — the only handlers that call wakeup_blocked_worker (item.waiter_handle) to set done and signal. wait_time < 0 waits forever, > 0 bounds it, 0 (default) allocates no handle at all.

Invariant 4-D (waiter lock ordering). The caller holds message_blocker::m from before enqueue until it enters cv.wait; the consumer’s wakeup_blocked_worker takes the same mutex before setting done. This bracketing makes the cross-thread wait safe even if the worker completes the request first — the re-checked predicate handle->done closes the race.

4.5 handle_message_queue and the dispatch table

handle_message_queue drains both queues in fixed order, then reclaims dead contexts via purge_stale_contexts (batch-return to the coordinator). queue_type has two values: IMMEDIATE (drained first — NEW_CLIENT, TAKEOVER_CLIENT, SEND_PACKET, RELEASE_PACKET) and LAZY (HIBERNATE, AWAKEN, HANDOFF_CLIENT, SHUTDOWN_CLIENT); the producer picks the queue at enqueue time per each message_type’s tag. handle_message_queue_by_index is the dispatch core — a constexpr array of { member-fn-ptr, statistics-slot } indexed directly by message_type, with static_asserts pinning it to exactly 10 entries starting at 0:

// worker::handle_message_queue_by_index -- src/connection/connection_worker.cpp
  size = m_queue_size[(size_t) type].exchange (0, std::memory_order_acquire); /* SNAPSHOT + reset */
  while (i++ < size && m_queue[(size_t) type].try_pop (request))              /* bounded by snapshot */
    { if (! (message_type::START <= request.type && message_type::TYPE_COUNT > request.type))
        { assert_release (false); continue; }                                 /* corrupt type: skip */
      if (! (this->*handler[(size_t) request.type].first) (request)) return false; /* member fn ptr */
      /* ... stats: MQ_<slot> + MQ_COMPLETED ... */ }

Any handler returning false aborts the drain and propagates false to run, killing the loop.

Invariant 4-B (bounded drain, no producer starvation). size = exchange (0, acquire) snapshots the count at this instant and zeroes it atomically. The while (i++ < size && ...) processes at most size messages even if producers keep pushing mid-drain; new pushes re-increment the counter and carry their own notify, so they serve on a later pass. This caps one drain’s work. The acquire pairs with enqueue’s release so the size messages’ contents are fully visible.

4.6 handle_message_queue_send_packet — branch-complete

SEND_PACKET transmits bytes on an owned connection. All work runs under the connection’s recursive cmutex, shared with the epoll EPOLLOUT path.

1. assert (item.conn); rmutex_lock (&item.conn->cmutex).
2. ctx = item.conn->context. If null (context torn down): unlock, run item.deleter () if present, wakeup_blocked_worker (item.waiter_handle), return true — send silently dropped.
3. Push each span in item.packet via push_for_send, then stamp (), then push_for_deleter (std::move (item.deleter)) (buffer freed once fully sent).
4. status = ctx->m_send.m_transmitter.fill (fd) — immediate non-blocking write.
   • PeerReset / Error: unlock, wake waiter, set ctx->m_ignore = IGNORE_ALL (main loop force-removes it), return true — fatal to the connection, not the worker.
   • assert (Ok || Pending).
   • Ok: m_transmitter.clear (), unlock, wake waiter, return true.
   • Pending (kernel buffer full): m_events.modify_descriptor (fd, EPOLLET|EPOLLIN|EPOLLOUT|EPOLLRDHUP, ctx). Fail → unlock, wake waiter, return false (epoll broken, fatal). Ok → unlock, then ctx->m_send.m_blocker = std::move (item.waiter_handle) — the waiter is not woken; the blocker parks on the context and is released later when the pending send drains.

flowchart TD
  A["lock cmutex"] --> B{"ctx==null?"}
  B -- yes --> C["unlock; run deleter;<br/>wake waiter; return true"]
  B -- no --> D["push spans + deleter; fill(fd)"] --> E{"status?"}
  E -- "PeerReset/Error" --> F["unlock; wake waiter;<br/>m_ignore=IGNORE_ALL; return true"]
  E -- "Ok" --> G["clear; unlock;<br/>wake waiter; return true"]
  E -- "Pending" --> H["modify_descriptor +EPOLLOUT"]
  H -- fail --> I["unlock; wake waiter; return false"]
  H -- ok --> J["unlock;<br/>m_blocker = move(waiter); return true"]

Figure 4-4. handle_message_queue_send_packet. Every terminal branch wakes the waiter except Pending, which parks the blocker on the context — so a wait_time caller blocks until its bytes actually leave the machine.

4.7 handle_message_queue_release_packet

RELEASE_PACKET returns received-packet memory to the connection’s receiver pool after a back-end task finished with it (the task ran on a pool thread and may not touch the receiver directly — only the owning worker may). Branches: (1) assert (item.conn), assert (item.packet.size () > 0), rmutex_lock (&item.conn->cmutex). (2) ctx = item.conn->context; if null → unlock, log, return true (connection and its arena already gone). (3) for each span, ctx->m_recv.m_receiver.release (packet.data ()). (4) unlock, return true. No waiter, no failure path beyond the null guard — releasing memory cannot fail the worker. This closes the “bytes to task” ownership loop of Chapter 8.

4.8 handle_message_queue_new_client

NEW_CLIENT installs a freshly accepted connection (the coordinator, Chapter 6, hands off a pre-built context). Branches:

1. assert (item.conn && item.conn->fd != -1); ctx = item.ctx; ctx->m_conn = item.conn;.
2. rmutex_lock (&ctx->m_conn->cmutex); publish ownership: conn->worker = this; conn->context = ctx; — now peer threads’ SEND_PACKET/RELEASE_PACKET can resolve this context via conn->context.
3. m_events.add_descriptor (fd, EPOLLET|EPOLLIN|EPOLLRDHUP, ctx). Fail → null worker/context, unlock, m_removed_context.push_back (ctx), return false.
4. m_context.insert (ctx). Duplicate (.second == false) → null worker/context, unlock, m_events.remove_descriptor (fd) (undo step 3), push to removed, return false.
5. Unlock. Seed stats (OPEND_NS, LAST_ACTIVE_NS = m_timens, LAST_MOVED_NS = 0, MOVE_COUNT = 0); bump CLIENT_NUM.
6. Defensive: if m_status == HIBERNATING (which “theoretically cannot be true” for a new client), restart the timer wheel via eventfd_starttimer so a mis-ordered queue does not strand a live client on a parked worker. return true.

Invariant 4-E (ownership publication order). conn->worker/conn->context are set under cmutex and before the fd joins epoll, and m_context insertion is under the same lock. From the moment a client fd can produce an epoll event, conn->context already resolves to a live, registered context — and a failed registration rolls both back (steps 3–4) so no half-installed client survives.

m_context is the worker’s private roster; the coordinator never touches it directly. Context teardown and the per-worker freelist are Chapter 7.

4.9 Chapter summary — key takeaways

1. One thread, one blocking point. run blocks only in epoll_wait; edge-triggered non-blocking sockets keep any one connection from stalling the reactor (Invariant 4-A), and the wait polls (timeout 0) whenever m_exhausted is non-empty so budget-throttled contexts are revisited.
2. Three fd kinds, uniform callback. Client sockets, m_eventfd, and m_timerfd all carry a context * in data.ptr; run demuxes by comparing against the two control fds and defers both to eventfd_handler after all client IO in the batch.
3. The eventfd is a doorbell, not a mailbox. notify writes a coalescing counter (a saturating EAGAIN counts as success); messages live in the queue, and the consumer snapshots the count with exchange (0, acquire) — making wakeups loss-safe (4-C) and drains bounded against producer starvation (4-B), with enqueue’s fetch_add (release) completing the pairing.
4. Synchronous callers ride a message_blocker. enqueue_and_notify locks handle->m before enqueuing (4-D); only START/SHUTDOWN_CLIENT/SEND_PACKET signal it, and SEND_PACKET defers the signal to real send completion on the Pending path.
5. Handlers guard on a vanished context and the QUEUE timer self-disarms. send_packet/release_packet/new_client resolve conn->context first and bail if null; new_client publishes ownership under cmutex before the fd enters epoll, rolling back on failure (4-E). A deferred close re-arms the 1 ms QUEUE timer with m_has_retry; eventfd_handler removes it once no retry is outstanding, so aggressive polling never becomes permanent.

Chapter 5: Send Recv Budgets and IO Bounding

Chapter 4 established the epoll loop: a single connection worker owns hundreds of sockets and, on each wakeup, walks the list of ready file descriptors calling handle_reception and handle_transmission. That design has one obvious failure mode. A client that keeps its socket perpetually readable — a bulk LOAD, a runaway result-set fetch, a slow-loris that trickles bytes forever — could sit inside handle_reception draining its socket until the kernel buffer empties, while every other connection the same worker owns waits. The reactor turns one thread per many connections (see the reactor-pattern motivation in cubrid-thread-manager.md) into one greedy connection starving its siblings.

This chapter answers: what stops one busy connection from starving the others inside a single worker? The answer is a per-connection byte budget charged per epoll tick, a partial-progress protocol that yields the worker the instant the budget is spent, and a two-phase re-arm that guarantees every other ready fd is serviced before the yielded connection runs again. We trace every branch of the budget check, the requeue-vs-continue decision, and the three close-path helpers (is_wait_required, has_remaining_tasks, purge_stale_contexts) that keep the budget bookkeeping consistent when a connection dies mid-flight. Pool sizing and the C10K backdrop are covered in cubrid-thread-worker-pool.md; this chapter is the per-tick IO discipline built on top of it.

5.1 The two budget parameters

Each worker reads two server parameters once in its constructor and caches them as plain size_t members — they are never re-read per tick, so the cost of the fairness check is a register compare.

// worker::worker -- src/connection/connection_worker.cpp
m_recv_budget = static_cast<size_t> (prm_get_integer_value (PRM_ID_CSS_RECV_BUDGET_PER_CONNECTION));
m_send_budget = static_cast<size_t> (prm_get_integer_value (PRM_ID_CSS_SEND_BUDGET_PER_CONNECTION));
m_exhausted.reserve (128);   /* <- the yielded-context sidebar, sized for churn */

The prm_Def[] entries fix the value envelope (member order is default_value, value, upper_limit, lower_limit):

Parameter	Default	Upper	Lower	Meaning of the value
recv_budget_per_connection	16 KB	1 GB	0	Max bytes recv()’d for one connection per tick
send_budget_per_connection	32 KB	1 GB	0	Max bytes sendmsg()’d for one connection per tick

The send default is double the recv default: replies are typically larger than requests, and the send path holds no reply-parsing cost, so it can afford a looser cap.

INVARIANT (0 means unlimited). A budget of 0 disables bounding. Every accounting site guards with limit > 0 && consumption >= limit — when limit == 0 the left conjunct short-circuits and the connection is never yielded for budget reasons. Because lower_limit is 0, an operator can legally set either budget to 0 and restore the pre-fairness “drain until EAGAIN” behaviour. Do not “simplify” the guard to consumption >= limit; that would turn 0 into “yield after every byte” and wedge the worker.

Note the width mismatch worth remembering before you widen these: m_send_budget is a size_t, but transmitter::fill (int fd, int limit) takes an int. The 1 GB upper_limit fits a signed 32-bit int (0x40000000), so the current envelope is safe; raising upper_limit past INT_MAX would silently make fill see a negative limit and disable bounding on the send side.

5.2 exhausted_context — the yielded-connection record

When a connection spends its budget, the worker cannot simply forget it — it must remember which half (recv, send, or both) still has pending work so it can resume exactly there next tick. That memory is exhausted_context, stored in a map keyed by the context id.

// worker::exhausted_context -- src/connection/connection_worker.hpp
struct exhausted_context
{
  bool prepared;      /* <- has this record survived one full loop yet? */
  uint32_t events;    /* <- EPOLLIN and/or EPOLLOUT still owed */
  context *ctx;       /* <- the connection to resume */
};
// ...
std::unordered_map<uint64_t, exhausted_context> m_exhausted;   /* keyed by ctx->m_id */

Field	Role	Why it exists
prepared	One-tick deferral latch. false on insert; flipped true on the first handle_exhausted pass, which then skips the record	Guarantees at least one full epoll pass over every other ready fd before a just-yielded connection is retried — the core anti-starvation delay
events	Bitmask of the IO directions still owed (EPOLLIN, EPOLLOUT, or both)	A connection can exhaust recv and send in the same tick; each direction is cleared independently as it completes
ctx	Back-pointer to the context (Chapter 7) whose IO is deferred	The resume target; validated against the map key on re-add

The map is keyed by ctx->m_id (a monotonic per-context id, Chapter 7), not by the fd or the pointer. That matters: fds are recycled by the kernel and context pointers by the freelist, but m_id is unique for the life of the process, so a stale entry can never alias a freshly reused connection.

INVARIANT (one map entry per live context). handle_exhausted_add_context looks up m_id before inserting. On a hit it asserts it->ctx == ctx and OR-folds the new direction into events rather than creating a second record. Two records for one connection would let handle_exhausted process the same ctx twice per tick and double-charge — or worse, resume after the context was freed. The single-entry rule is what makes m_exhausted.erase (ctx->m_id) in the close path (§5.7) a complete removal.

flowchart LR
  subgraph W["worker (one thread)"]
    MAP["m_exhausted<br/>unordered_map&lt;uint64_t, exhausted_context&gt;"]
    RB["m_recv_budget"]
    SB["m_send_budget"]
  end
  E1["exhausted_context<br/>prepared / events / ctx"]
  C1["context (Ch 7)<br/>m_id, m_recv, m_send"]
  MAP -->|"value"| E1
  E1 -->|"ctx"| C1
  MAP -.->|"key = ctx-&gt;m_id"| C1

Figure 5-1. m_exhausted maps a context id to the record that remembers which IO directions it still owes and whether it has been deferred a tick.

5.3 Charging the budget inside drain and fill

The budget is spent by the IO primitives, not by the worker. receiver::drain runs its recv state machine (Chapter 8 turns the bytes into tasks) accumulating a local consumption, and every state that advances re-checks the cap:

// receiver::receive_in_allocated -- src/connection/receiver.cpp
m_received += bytes;
consumption += bytes;                       /* <- charge the tick */
// ... condensed: parse the now-complete packet ...
if (limit > 0 && consumption >= limit)
  {
    return result::BudgetExhausted;         /* <- stop; do not loop for more */
  }
return result::Ok;

drain’s dispatch loop treats BudgetExhausted as a terminal status that first records a metric, then returns it up unchanged:

// receiver::drain -- src/connection/receiver.cpp
case result::BudgetExhausted:
  m_stats->add (statistics::context::RECV_BUDGET_HIT, 1);   /* <- observability, Ch 11 */
  [[fallthrough]];
case result::Pending:
case result::Error:
case result::PeerReset:
  return status;

The send side mirrors this. transmitter::fill walks the iovec array of the msghdr, and checks the cap before each sendmsg rather than after — so it never issues a syscall it has no budget to pay for:

// transmitter::fill -- src/connection/transmitter.cpp
while (msg->msg_iovlen)
  {
    if (limit > 0 && consumption >= limit)
      {
        m_stats->add (statistics::context::SEND_BUDGET_HIT, 1);
        return result::BudgetExhausted;
      }
    bytes = ::sendmsg (fd, msg, MSG_NOSIGNAL);
    // ... condensed: advance iov_base/iov_len by bytes, consumption += bytes ...
  }
return result::Ok;   /* <- iovec fully drained; nothing left to send */

INVARIANT (budget is checked at frame boundaries, never mid-syscall). Neither primitive interrupts a recv/sendmsg in progress; the cap is examined only between kernel calls. A single syscall can therefore overshoot the budget by up to one socket-buffer’s worth of bytes. The budget bounds the number of tick-hogging iterations, not the exact byte count. This is deliberate: partial syscalls would require restartable framing the receiver (Chapter 8) does not have. Both statuses — Ok (nothing left) and BudgetExhausted (cap hit with work remaining) — are distinct from Pending (EAGAIN, kernel buffer drained). The requeue decision in §5.4 turns on exactly this three-way distinction.

5.4 The requeue-vs-continue decision

handle_reception and handle_transmission each take a boolean in_exhausted — false when the call comes from the fresh epoll pass in run, true when it comes from the deferred re-run in handle_exhausted. That flag is the entire requeue gate. On the recv side:

// worker::handle_reception -- src/connection/connection_worker.cpp
io_status = ctx->m_recv.m_receiver.drain (ctx->m_conn->fd, m_recv_budget);
if (io_status == result::PeerReset || io_status == result::Error)
  {
    ctx->m_send.m_transmitter.empty ();
    this->handle_connection_close (ctx);          /* <- socket is dead: close, do not requeue */
    return io_status;
  }
assert (io_status == result::Pending || io_status == result::BudgetExhausted);

if (!in_exhausted && io_status == result::BudgetExhausted)
  {
    handle_exhausted_add_context (ctx, EPOLLIN);  /* <- yield: remember to resume recv */
  }
// still dispatch whatever WAS parsed this tick:
if (ctx->m_recv.m_receiver.get_result ()->empty ())
  {
    return io_status;
  }
// ... condensed: for each parsed packet -> handle_packet (Ch 8) ...

Three things to read off this. First, partial progress is never discarded: even when the budget is hit, any whole packets drain already parsed are handed to handle_packet in the same call. The connection yields the worker, not its already-received data. Second, the requeue happens only when !in_exhausted — a call already originating from the exhausted list does not re-add itself; it just returns BudgetExhausted and stays in the map for the next tick (its events bit is still set). Third, Pending never requeues: it means the kernel buffer is empty, so epoll level/edge triggers will wake the worker naturally when more bytes arrive.

handle_transmission adds a fourth status, Ok, because a send can finish:

// worker::handle_transmission -- src/connection/connection_worker.cpp
status = ctx->m_send.m_transmitter.fill (ctx->m_conn->fd, m_send_budget);
if (status == result::PeerReset || status == result::Error) { /* close, return */ }
assert (status == result::Ok || status == result::Pending || status == result::BudgetExhausted);
if (status == result::Ok)
  {
    this->wakeup_blocked_worker (ctx->m_send.m_blocker);          /* <- reply fully flushed */
    // ... condensed: if CONN_CLOSING -> handle_connection_close; else rearm EPOLLIN ...
    ctx->m_send.m_transmitter.clear ();
  }
else if (!in_exhausted && status == result::BudgetExhausted)
  {
    handle_exhausted_add_context (ctx, EPOLLOUT);                 /* <- yield: resume send */
  }
return status;

flowchart TD
  A["handle_reception / handle_transmission(ctx, in_exhausted)"]
  A --> B["drain / fill charged against budget"]
  B --> C{"io_status?"}
  C -->|"PeerReset / Error"| D["handle_connection_close<br/>return (no requeue)"]
  C -->|"Ok (send only)"| E["wakeup blocker, clear<br/>rearm EPOLLIN or close"]
  C -->|"Pending"| F["dispatch parsed packets<br/>return (epoll will re-wake)"]
  C -->|"BudgetExhausted"| G{"in_exhausted?"}
  G -->|"true"| H["stay in m_exhausted<br/>events bit still set"]
  G -->|"false"| I["handle_exhausted_add_context<br/>set EPOLLIN / EPOLLOUT"]
  I --> J["dispatch parsed packets, return"]

Figure 5-2. Every terminal status of the per-tick IO call and whether it requeues the connection onto the exhausted list.

5.5 The two-phase re-arm and the busy-wait tick

handle_exhausted_add_context is the only writer that grows the map. It sets prepared = false on a brand-new record and OR-folds events on an existing one:

// worker::handle_exhausted_add_context -- src/connection/connection_worker.cpp
if (m_exhausted.find (ctx->m_id) == m_exhausted.end ())
  {
    m_exhausted[ctx->m_id].prepared = false;   /* <- must survive one loop before resume */
    m_exhausted[ctx->m_id].events = event;
    m_exhausted[ctx->m_id].ctx = ctx;
  }
else
  {
    assert (m_exhausted[ctx->m_id].ctx == ctx);
    m_exhausted[ctx->m_id].events |= event;    /* <- recv+send owed at once */
  }

The run loop closes the fairness circuit in two places. First, the epoll wait timeout collapses to non-blocking whenever the exhausted list is non-empty, so the worker keeps spinning instead of sleeping on epoll_wait while it owes work:

// worker::run -- src/connection/connection_worker.cpp
nfds = m_events.wait (events.data (), events.size (),
                      m_exhausted.empty () ? TIMEOUT_INFINITE : TIMEOUT_NOWAIT);
// ... condensed: for each of nfds ready fds -> handle_reception / handle_transmission ...
if (m_exhausted.size () > 0)
  {
    if (!this->handle_exhausted ()) { return false; }   /* <- after the fresh fds */
  }

The ordering is the whole point: on every tick the fresh epoll set is serviced first, then the exhausted list. A greedy connection that yielded last tick cannot cut in front of a sibling that just became ready. handle_exhausted then applies the second delay via prepared:

// worker::handle_exhausted -- src/connection/connection_worker.cpp
for (auto it = m_exhausted.begin (); it != m_exhausted.end (); )
  {
    if (!it->second.prepared)
      {
        it->second.prepared = true;   /* <- first sighting: arm, but skip this tick */
        it++;
        continue;
      }
    ctx = it->second.ctx;
    if (it->second.events & EPOLLIN)
      {
        status = this->handle_reception (ctx, true);   /* <- in_exhausted = true */
        if (status == result::ClosedConnection || status == result::PeerReset)
          { it = m_exhausted.erase (it); continue; }   /* <- gone: drop record */
        if (status == result::Error) { return false; } /* <- fatal: kill worker */
        if (status == result::Pending)                 /* <- recv fully drained now */
          {
            it->second.events &= ~EPOLLIN;             /* <- clear the owed direction */
            if (!it->second.events)
              { it = m_exhausted.erase (it); continue; }
          }
        /* BudgetExhausted again: fall through, EPOLLIN stays set, retry next tick */
      }
    if (it->second.events & EPOLLOUT)
      {
        status = this->handle_transmission (ctx, true);
        // ... symmetric: erase on Closed/PeerReset, return false on Error ...
        if (status == result::Ok || status == result::Pending)
          {
            it->second.events &= ~EPOLLOUT;
            if (!it->second.events)
              { it = m_exhausted.erase (it); continue; }
          }
      }
    it++;
  }

Trace every exit for one record: (a) first pass — prepared flips false→true, continue, record untouched; (b) connection died — erase and move on; (c) Error — return false, run tears the worker down; (d) direction completed (Pending for recv, Ok/Pending for send) — clear that bit, and if events is now zero, erase; (e) budget hit again — no branch matches, the bit stays set, it++, retried next tick still bounded to one budget’s worth. The it = m_exhausted.erase (it) idiom is the correct erase-while-iterating pattern for unordered_map (erase returns the next valid iterator).

INVARIANT (bounded work per connection per tick). Between two consecutive service opportunities for any one connection there is at least one full pass over every other ready fd (ordering) plus, on first exhaustion, one extra tick of deferral (prepared). No connection can consume more than recv_budget + send_budget bytes of socket IO before the worker turns to its siblings. Violate either half — service the exhausted list before the fresh fds, or skip the prepared skip — and the starvation this chapter exists to prevent returns.

5.6 Close-path helpers: is_wait_required and has_remaining_tasks

The send budget’s flush counterpart shows up when a connection is being torn down: the worker must not drop bytes the client is still owed. Two predicates inside handle_connection_close gate that.

is_wait_required decides whether a connection whose transaction index is not yet assigned (a client still inside boot_client_register) is worth sleeping 50 ms for:

// worker::is_wait_required -- src/connection/connection_worker.cpp
bool worker::is_wait_required (context *ctx)
{
  if (ctx->m_conn->fd == cdc_Gl.conn.fd)   /* <- the change-data-capture pseudo-connection */
    {
      return false;                        /* <- CDC has no boot handshake: never wait */
    }
  return true;                             /* <- every real client: wait once, then re-read tran index */
}

Only two outcomes. The single special case is the CDC global connection, which never runs the client boot handshake, so waiting on its transaction index would be a pure 50 ms stall for nothing; every ordinary client returns true and the caller sleeps once before re-reading tran_index/client_id.

has_remaining_tasks is the send-flush gate — it decides whether close must be retried because a reply is still buffered:

// worker::has_remaining_tasks -- src/connection/connection_worker.cpp
bool worker::has_remaining_tasks (context *ctx)
{
  this->handle_message_queue_by_index (queue_type::IMMEDIATE);   /* <- drain queued RELEASE_PACKET first */
  if (ctx->m_ignore < ignore_level::IGNORE_ALL && !ctx->m_send.m_transmitter.empty ())
    {
      return true;    /* <- graceful close: unsent bytes remain, retry later */
    }
  return false;       /* <- forced close (IGNORE_ALL) OR send buffer empty: proceed */
}

Three branches. It first services the IMMEDIATE queue, because a pending RELEASE_PACKET message may free memory the transmitter is about to inspect. Then: a graceful close (m_ignore < IGNORE_ALL) with a non-empty transmitter returns true, sending handle_connection_close down its retry path so the reply gets one more flush attempt. A forced close (m_ignore == IGNORE_ALL, set on Error/PeerReset in §5.4) short-circuits the && and returns false immediately — pending sends are abandoned because the peer is already gone.

5.7 purge_stale_contexts — scrubbing the budget map on close

handle_connection_close does not free a context inline; it marks ctx->m_removed = true and appends it to m_removed_context. The deferred reaper, called at the end of handle_message_queue, ha_close_all_connections, and finalize, is where the context leaves both the live set and the exhausted map:

// worker::purge_stale_contexts -- src/connection/connection_worker.cpp
coordinator::message message;
message.type = coordinator::message_type::RETURN_TO_POOL;
message.resource = m_removed_context;             /* <- hand the batch to the coordinator (Ch 6) */

for (context *ctx : m_removed_context)
  {
    if (m_context.erase (ctx) == 0)
      {
        er_log_conn (__FILE__, __LINE__, "... context not found\n");
        continue;                                 /* <- already gone: skip the exhausted scrub */
      }
    m_exhausted.erase (ctx->m_id);                /* <- THE critical line: drop any owed IO */
  }
m_removed_context.clear ();
m_coordinator->enqueue (std::move (message));
if (!m_coordinator->notify ()) { assert_release (false); }

Two branches per context. If m_context.erase returns 0 the context was never in this worker’s live set (a double-removal race) — log and continue, skipping the exhausted scrub. Otherwise m_exhausted.erase (ctx->m_id) removes any deferred-IO record. The else branch matters more than it looks: without it, a connection that hit its budget (§5.4) and then got closed would leave a dangling exhausted_context whose ctx points at memory the coordinator is about to recycle into the freelist (Chapter 7). The next handle_exhausted would dereference it. The same one-line scrub appears wherever a context leaves a worker abruptly — handle_message_queue_handoff_client runs it too — which is exactly why §5.2’s single-entry invariant makes the erase-by-m_id complete.

INVARIANT (no exhausted record outlives its context). Every path that removes a context from m_context also erases ctx->m_id from m_exhausted in the same critical section (the worker thread, no lock needed — both maps are worker-private). The exhausted map may only reference live contexts owned by this worker; a violation is a use-after-free the very next tick.

5.8 Chapter summary — key takeaways

1. Two cached budgets bound per-tick IO. recv_budget_per_connection (16 KB) and send_budget_per_connection (32 KB) are read once into m_recv_budget/m_send_budget; drain and fill charge a local consumption and return BudgetExhausted once limit > 0 && consumption >= limit. A budget of 0 disables bounding.

2. The budget is checked at frame boundaries, so it bounds iterations, not exact bytes. A syscall in progress is never interrupted; overshoot up to one socket-buffer is expected and intentional.

3. Partial progress is preserved. Hitting the budget yields the worker, never the connection’s data — already-parsed packets are dispatched in the same call before the connection is deferred.

4. exhausted_context remembers the owed directions. Keyed by the immortal ctx->m_id, it carries events (EPOLLIN/EPOLLOUT still owed) and prepared (a one-tick deferral latch); one entry per live context is an enforced invariant.

5. Fairness comes from ordering plus deferral. run services the fresh epoll set before the exhausted list and spins with TIMEOUT_NOWAIT while the list is non-empty; handle_exhausted skips each record for one tick via prepared. Together they guarantee bounded work per connection per tick.

6. The requeue gate is in_exhausted. A call from the fresh pass requeues on BudgetExhausted; a call already from the exhausted list does not re-add, leaving its events bit set for the next tick. Pending never requeues; Ok (send only) finishes and clears.

7. Close-path helpers keep the accounting consistent. is_wait_required special-cases the CDC connection; has_remaining_tasks retries a graceful close while the transmitter holds unsent bytes but lets a forced (IGNORE_ALL) close proceed; and purge_stale_contexts erases ctx->m_id from m_exhausted on every removal so no deferred record ever outlives its context.

Chapter 6: The Coordinator Placement Rebalance and Auto Scale

Every connection worker emits a STATISTICS message to one dedicated thread — the coordinator. This chapter answers: how does that thread decide where a new connection goes, when to move one off an overloaded worker, and when to add or drop a worker? All three reduce to one scalar — the worker score — plus a small state machine on top. The companion cubrid-thread-manager.md sections Score-based connection assignment (CBRD-26424) and Auto scaling (CBRD-26406) explain why; here we trace the arithmetic and every branch. Struct fields (m_scaling, m_scaling_statistics, m_task_statistics, worker_statistics) were tabulated in Chapter 2; this chapter recalls only the ones that drive control flow.

6.1 The control loop — coordinator::run and its three timers

coordinator::run is a single-consumer epoll loop over exactly three fds: m_eventfd (a wakeup from any thread that enqueued a message), m_timerfd (a coalesced periodic tick), and m_ctrlfd (the debug control socket). It never touches connection sockets — those live on the workers.

// coordinator::run -- src/connection/coordinator.cpp
while (!m_stop)
  {
    nfds = m_events.wait (events.data (), events.size (), TIMEOUT_INFINITE);
    if (nfds < 0)
      {
        if (errno == EINTR) { continue; }        /* <- signal, retry */
        assert_release (false); continue;        /* <- other error: fatal in debug */
      }
    m_timens = this->get_monotonic_ns ();        /* <- one clock read per loop */
    for (i = 0; i < nfds; i++)
      if (events[i].events & EPOLLIN)
        {
          if (events[i].data.fd == m_eventfd)      { eventfd_clear (m_eventfd); handle_message_queue (); }
          else if (events[i].data.fd == m_timerfd) { /* dispatch timers, below */ }
          else if (events[i].data.fd == m_ctrlfd)  { handle_controller (); }
        }
  }

m_timerfd is armed to the minimum latency of all registered timers, so one physical tick may only mean the fastest handler is due. Each handler is re-gated against its own last_time:

// coordinator::run (timerfd branch) -- src/connection/coordinator.cpp
for (j = 0; j < static_cast<int> (timer_type::TYPE_COUNT); j++)
  {
    if (!m_timer_handler[j].valid) { continue; }
    if (m_timens - m_timer_handler[j].last_time > static_cast<uint64_t> (m_timer_handler[j].latency))
      {
        if (!m_timer_handler[j].function ()) { return false; }  /* <- handler asked to stop */
        m_timer_handler[j].last_time = m_timens;                /* <- rearm this handler only */
      }
  }
eventfd_clear (m_timerfd);   /* <- drain the timerfd counter last */

The three handlers, installed in the constructor:

timer_type	latency	Handler	Job
STATISTICS	LOW_LATENCY 1 s	statistics_update	Fold new counters into EWMAs and scores
REBALANCING	MEDIUM_LATENCY 5 s	statistics_rebalancing	Move one hot connection max→min
SCALING	HIGH_LATENCY 60 s	statistics_scaling	Step the auto-scale trial machine

INVARIANT (single-consumer coordinator state). m_statistics, m_scaling, m_scaling_statistics, m_migrating are mutated only on the coordinator thread — from handle_message_queue (eventfd) or the timer handlers, both inside run. Producers touch only the lock-free m_queue + m_queue_size. That is why nothing in this chapter takes a lock: one writer.

flowchart TD
  W["epoll wait TIMEOUT_INFINITE"]
  N{"nfds < 0"}
  E{"errno == EINTR"}
  T["m_timens = get_monotonic_ns"]
  L["for each ready fd"]
  EV["eventfd\nclear + handle_message_queue"]
  TM["timerfd\ngate each handler by last_time"]
  CT["ctrlfd\nhandle_controller"]
  W --> N
  N -->|yes| E
  E -->|yes| W
  E -->|no| W
  N -->|no| T --> L
  L --> EV --> W
  L --> TM --> W
  L --> CT --> W

Figure 6-1. coordinator::run — one clock read per wakeup, then fan ready fds to handlers; timers self-gate on elapsed time.

6.2 Turning raw counters into one number — the score

Placement, rebalancing, and scaling all rank by worker_statistics::m_score, recomputed by statistics_update_score whenever a worker’s stats change. Two macros weight it, both on VAL_TO_SCORE (w, m, s) = w * s / m — weight w on sample s normalized by reference magnitude m:

// score macros -- src/connection/coordinator.cpp
#define VAL_TO_SCORE(w, m, s) ((w) * static_cast<double> (s) / (m))
#define EVAL_WORKER(mq, rmutex)     (VAL_TO_SCORE (25, 3.5, (mq)) + VAL_TO_SCORE (500, 1, (rmutex)))
#define EVAL_CONTEXT(bytes, budget) (VAL_TO_SCORE (50, 1000, (bytes)) + VAL_TO_SCORE (10, 1, (budget)))

// coordinator::statistics_update_score -- src/connection/coordinator.cpp
m_statistics[worker].m_score =
        1 * static_cast<double> (m_statistics[worker].m_client_num) / 1 +   /* <- one point per client */
        EVAL_WORKER (EWMA_WORKER (MQ_COMPLETED), EWMA_WORKER (BLOCKED_RMUTEX)) +
        EVAL_CONTEXT (EWMA_CONTEXT (BYTES_IN_TOTAL) + EWMA_CONTEXT (BYTES_OUT_TOTAL),
                      EWMA_CONTEXT (RECV_BUDGET_HIT) + EWMA_CONTEXT (SEND_BUDGET_HIT));

Three additive pressures: headcount (m_client_num), worker pressure (completed MQ work at 25/3.5, plus recv-mutex blocking at a heavy 500 — a blocked worker is expensive), and context pressure (bytes in+out at 50/1000, plus send/recv budget-hit frequency at 10). The EWMA_* operands are exponentially-weighted moving averages, so the score tracks sustained load, not spikes. The STATISTICS timer body drives the fold:

// coordinator::statistics_update -- src/connection/coordinator.cpp
this->handle_message_queue ();   /* <- absorb STATISTICS messages queued since last tick */
this->statistics_update_task (); /* <- fold the global task-worker counters */

Per-worker EWMA folding happens in handle_message_queue_statistics (Ch 7/11), which calls statistics_update_connection then statistics_update_score. The kernel is statistics_EWMA:

// coordinator::statistics_EWMA (scalar overload) -- src/connection/coordinator.cpp
diff = 0;
if (current > prev) { diff = static_cast<double> (current - prev); }  /* <- counters only rise; clamp */
acc = acc * (1 - alpha) + diff * (alpha / (time_delta * 1e-6));       /* <- rate per us, alpha = EWMA_ALPHA */
prev = current;

statistics_update_connection branches on m_last_updated: first sample (== 0) copies raw values into both accumulator and “previous” slot (no phantom delta); every later sample resets m_sum, folds the worker metric and each context metric, then rebuilds m_sum as the summation of every context’s smoothed value — that m_sum is what EVAL_CONTEXT reads.

INVARIANT (no first-sample spike). A counter’s first observation must not be a delta from zero, or a fresh worker shows a huge fake rate. m_last_updated (and the parallel requested.second gauge in statistics_update_task) gate the first fold; break it and every new worker looks overloaded and repels arrivals.

Caveat from the merged checkout: css_get_task_stats is a stub (body commented out in server_support.c), so m_task_statistics folds zeros — the task-completion term that later feeds scale_selection is inert until that plumbing lands (see parent doc Cross-check Notes).

6.3 Placement — statistics_find_score_extremes

Placement is the min side of one two-extreme pass:

// coordinator::statistics_find_score_extremes -- src/connection/coordinator.cpp
max = 0; min = 0;
for (i = 1; i < m_current_worker; i++)   /* <- only ACTIVE workers; index 0 seeds both */
  {
    if (m_statistics[i].m_score < m_statistics[min].m_score)       { min = i; }
    else if (m_statistics[i].m_score >= m_statistics[max].m_score) { max = i; }
  }
return { min, max };

Two subtleties. The bound is m_current_worker, not m_max_worker: hibernated workers above the current count are invisible, so a new client never lands on a sleeping worker. The else if means no worker is min and max in one iteration, and >= biases max-ties to the higher index while < keeps the lowest index for min — a deterministic tie-break. New-client placement (handle_message_queue_new_client, Ch 7) calls this, hands the client to min, bumps m_client_num, and recomputes the score. That is all of “where new connections go”: the score-minimum active worker.

coordinator::random_bit (a bernoulli(0.5) over a static mt19937) is the intended randomized tie-break, but in the merged checkout it has no caller — the deterministic >=/< stands in for it. If you wire real tie randomization, note the generator is a function-local static, safe only because random_bit too runs solely on the coordinator thread.

6.4 Rebalancing — statistics_rebalancing and transfer_connection

The 5 s REBALANCING timer moves at most one connection hottest→coldest, only if the imbalance is worth the migration cost:

// coordinator::statistics_rebalancing -- src/connection/coordinator.cpp
constexpr double threshold = 0.2;
std::tie (min, max) = statistics_find_score_extremes ();
diff = m_statistics[max].m_score - m_statistics[min].m_score;
if (diff <= m_statistics[max].m_score * threshold) { return true; }  /* <- within 20%: leave it */

score = -1; id = 0;
for (auto &stats : m_statistics[max].m_contexts)
  {
    target = EVAL_CONTEXT (/* this context's smoothed bytes + budget hits */);
    if (target <= diff / 2 && score < target) { score = target; id = stats.first; }  /* <- biggest that fits */
  }
if (id != 0) { this->transfer_connection (id, max, min); }

Branches: (1) scores within 20% → return, no churn. (2) Else scan the hot worker’s contexts for the connection whose own EVAL_CONTEXT is the largest that still fits under diff / 2 — the anti-oscillation ceiling; moving more than half the gap would make min the new max. (3) If none qualifies (id stays 0, e.g. one giant connection dominates), nothing moves.

transfer_connection performs the hand-off (also reached from scale_down and the controller’s CLIENT_MOVE):

// coordinator::transfer_connection -- src/connection/coordinator.cpp
if (m_migrating.find (id) != m_migrating.end ()) { return false; }  /* <- already in flight, skip */
m_migrating.insert (id);
auto stats = m_statistics[from].m_contexts.find (id);
m_statistics[to].m_contexts.emplace (stats->first, /* copy both EWMA slots to destination */);
/* the stats in worker[from] are removed when the worker responds. */
request.type = connection::worker::message_type::HANDOFF_CLIENT;
request.id = stats->first; request.worker_ptr = workers[to].get (); request.worker_index = to;
workers[from]->enqueue (queue_type::LAZY, std::move (request));   /* <- tell SOURCE to hand off */
if (!workers[from]->notify ()) { assert_release (false); }

Optimistic-copy-then-confirm: the destination map gets the stats immediately (a copy of both accumulator and previous metrics), but the source copy is erased only when the worker replies HANDOFF_REPLY (Ch 7), which flips m_client_num on both sides and, if the hand-off failed (!transferred), reverts by erasing the speculative destination copy. The message goes on the source’s LAZY queue (Ch 5): a hand-off is not latency-critical.

INVARIANT (one hand-off per connection). m_migrating is the in-flight set keyed by connection id. transfer_connection refuses a second migration of the same id; it is erased only in handle_message_queue_handoff_reply. Drop this and scale_down looping over contexts while a rebalance is mid-flight could emit two HANDOFF_CLIENTs for one connection, double-counting m_client_num.

flowchart TD
  R["statistics_rebalancing (5s)"]
  X["find score extremes -> min,max"]
  D{"diff <= max_score * 0.20"}
  S["scan max.contexts\npick biggest EVAL_CONTEXT <= diff/2"]
  F{"id != 0"}
  T["transfer_connection id max->min"]
  M{"id in m_migrating"}
  C["copy stats to dest\nHANDOFF_CLIENT to source LAZY"]
  Z["return, no move"]
  R --> X --> D
  D -->|balanced| Z
  D -->|no| S --> F
  F -->|no fit| Z
  F -->|yes| T --> M
  M -->|in flight| Z
  M -->|new| C

Figure 6-2. statistics_rebalancing — one bounded move per tick, gated by the 20% band, the diff/2 fit ceiling, and the in-flight set.

6.5 Auto-scaling — the trial-and-observe state machine

The coordinator cannot compute the “right” worker count, so it experiments. Two enums drive it: m_status (STABLE/DRAINING/EXPANDING) is the physical state — is a worker mid-migration; m_scaling_statistics.status (STABLE/TRIAL) is the experiment state — are we probing. The experiment fields (from Ch 2):

Field	Role
status	STABLE idle, TRIAL probing this many more ticks
window_size	Max probe length; auto_scaling_window_size default 4
history	Per-probe {scale, score} samples, one per TRIAL tick
direction	Which way this trial steps (UP/DOWN)
count	Ticks remaining in this trial
previous_direction	Direction of the last completed trial
previous_scale	Worker count when the last trial began

Arming — scale_trial (called when scaling is STABLE, and after each trial to chain the next):

// coordinator::scale_trial -- src/connection/coordinator.cpp
m_scaling_statistics.history.clear ();
if (m_scaling_statistics.previous_scale == m_current_worker)          /* <- last trial changed nothing */
  direction = (previous_direction == DOWN) ? UP : DOWN;              /* <- so try the other way */
else
  direction = m_scaling_statistics.previous_direction;              /* <- it helped, keep going */
if (direction == DOWN) count = min (m_current_worker - m_min_worker, window_size);
else                   count = min (m_max_worker - m_current_worker, window_size);
if (m_scaling_statistics.count == 0)                                 /* <- no headroom this way */
  { previous_direction = flip (previous_direction); status = STABLE; }
else
  { m_scaling_statistics.status = TRIAL; }

Direction flips if the last trial ended where it began (fruitless), else persists; count clamps to remaining headroom and window_size; zero headroom aborts to STABLE after flipping previous_direction so the next scale_trial probes the other way.

Stepping — statistics_scaling (the 60 s SCALING timer):

// coordinator::statistics_scaling -- src/connection/coordinator.cpp
if (m_scaling_statistics.status == scaling_status::STABLE) { this->scale_trial (); return true; }
assert (m_scaling_statistics.status == scaling_status::TRIAL);
bytes_inout = /* sum of BYTES_IN_TOTAL + BYTES_OUT_TOTAL over all workers' m_sum */;
m_scaling_statistics.history.push_back ({ m_current_worker,
        VAL_TO_SCORE (50, 1000, bytes_inout) + m_task_statistics.completed.first * 2 });  /* <- throughput score */
m_scaling_statistics.count--;
if (m_scaling_statistics.count == 0)
  {
    m_scaling_statistics.previous_scale = m_current_worker;
    selected = this->scale_selection ();
    if (selected < m_current_worker)      { previous_direction = DOWN; scale_down (); scale_trial (); }
    else if (selected > m_current_worker) { previous_direction = UP;
                                            for (...) scale_up (); scale_trial (); }
    else                                  { m_scaling_statistics.status = STABLE; }  /* <- current won */
  }
else
  {
    if (m_scaling_statistics.direction == DOWN) { this->scale_down (); }
    else                                        { this->scale_up (); }
  }

Branches: a STABLE tick just arms a trial and returns. A TRIAL tick records the current count and a throughput score (aggregate bytes 50/1000 plus twice the smoothed task-completion rate) into history, then decrements count. If ticks remain, it steps once in direction. If this was the last tick (count == 0), scale_selection picks the best-scoring recorded scale and it moves toward it — scale_down, a loop of scale_up, or drop to STABLE if current already won — then rearms via scale_trial.

// coordinator::scale_selection -- src/connection/coordinator.cpp
max_score = max over history of stats.score;
for (auto &stats : history) if (max_score * 0.95 < stats.score) candidates.push_back (stats.scale);
if (candidates.size () != 0) return candidates[uniform_int (0, size-1)];  /* <- random among top 5% */
return m_current_worker;   /* <- empty history: stay put */

Physical steps. scale_up wakes the next dormant worker:

// coordinator::scale_up -- src/connection/coordinator.cpp
if (m_current_worker >= m_max_worker || m_status != status::STABLE) { return false; }  /* <- no room / busy */
m_scaling.draining_worker = -1;
m_status = status::EXPANDING;
m_statistics[m_current_worker].m_score = 0;   /* <- zero score => wins placement, absorbs new clients */
/* enqueue AWAKEN (LAZY) to workers[m_current_worker]; notify */
m_current_worker++;
m_scaling.last_expand_ns = get_monotonic_ns ();
m_status = status::STABLE;

Zeroing the new worker’s score before incrementing m_current_worker steers the next statistics_find_score_extremes min to the fresh worker, filling it fast. scale_down is the only two-phase op:

// coordinator::scale_down -- src/connection/coordinator.cpp
if (m_current_worker <= m_min_worker || m_status != status::STABLE) { return false; }
m_current_worker--;                                   /* <- worker now invisible to placement */
for (auto &stats : m_statistics[m_current_worker].m_contexts)   /* <- evict every connection */
  { std::tie (newhome, std::ignore) = statistics_find_score_extremes ();
    transfer_connection (stats.first, m_current_worker, newhome); }
m_scaling.draining_worker = m_current_worker;
m_status = status::DRAINING;                          /* <- stay DRAINING until it empties */

It decrements first so placement and the extremes scan (i < m_current_worker) stop targeting the doomed worker, then hands off every connection to the current score-min home. It cannot finish synchronously — workers must confirm each hand-off — so it parks in DRAINING. Completion is async, from handle_message_queue_statistics: when the draining worker’s next report shows an empty context set, scale_down_finish fires:

// coordinator::scale_down_finish -- src/connection/coordinator.cpp
/* enqueue HIBERNATE (LAZY) to workers[draining_worker]; notify */
m_statistics[draining_worker].m_score = 0; /* ...clear m_core, m_client_num, m_sum, m_worker, m_contexts... */
m_scaling.last_drain_ns = get_monotonic_ns ();
m_scaling.draining_worker = -1;
m_status = status::STABLE;

INVARIANT (only STABLE scales). Both scale_up/scale_down refuse to start unless m_status == STABLE, and each leaves it non-STABLE while a hand-off is outstanding. scale_up restores STABLE synchronously; scale_down waits for scale_down_finish. So a trial that issues scale_down cannot issue another step until the drain completes — the m_status gate serializes structural changes. Remove it and a second scale_down could pick the still-draining worker as a hand-off target.

stateDiagram-v2
  [*] --> STABLE
  STABLE --> TRIAL: scale_trial \n count > 0
  STABLE --> STABLE: scale_trial \n count == 0 no headroom
  TRIAL --> TRIAL: tick \n count > 0 step up or down
  TRIAL --> STABLE: count == 0 \n selected == current
  TRIAL --> TRIAL: count == 0 \n selected differs then rearm

Figure 6-3. m_scaling_statistics.status — the experiment machine. Each TRIAL tick appends one history sample; count == 0 commits to the best-scoring scale and rearms.

flowchart TD
  S["statistics_scaling (60s)"]
  Q{"scaling status"}
  A["scale_trial arm next"]
  REC["record history\nscore = bytes + 2*completed\ncount--"]
  C{"count == 0"}
  STEP{"direction"}
  SU["scale_up"]
  SD["scale_down"]
  SEL["scale_selection\ntop-5% band pick"]
  CMP{"selected vs current"}
  DN["dir=DOWN; scale_down; rearm"]
  UP["dir=UP; scale_up xN; rearm"]
  ST["status = STABLE"]
  S --> Q
  Q -->|STABLE| A
  Q -->|TRIAL| REC --> C
  C -->|no| STEP
  STEP -->|UP| SU
  STEP -->|DOWN| SD
  C -->|yes| SEL --> CMP
  CMP -->|less| DN
  CMP -->|greater| UP
  CMP -->|equal| ST

Figure 6-4. statistics_scaling — every branch. A STABLE tick arms a trial; a TRIAL tick records a sample and either steps once more or, at count == 0, commits toward the best-scoring scale.

6.6 Chapter summary — key takeaways

1. One thread, one number. coordinator::run is a lock-free single-consumer epoll loop over eventfd/timerfd/ctrlfd; all decisions rank workers by m_score, and all coordinator state is mutated only here — the invariant that lets the chapter run lock-free.
2. Score = headcount + worker pressure + context pressure, each EWMA-smoothed (VAL_TO_SCORE/EVAL_WORKER/EVAL_CONTEXT), with BLOCKED_RMUTEX weighted 500 so a stalled worker repels load hardest; first-sample folding is gated by m_last_updated.
3. Placement is the min side of statistics_find_score_extremes, scanned only over [0, m_current_worker) so sleeping workers are unreachable; >=/< is the current deterministic tie-break and random_bit is wired in name only.
4. Rebalancing moves at most one connection per 5 s, only past a 20% band, choosing the largest context under diff/2 to avoid overshoot; transfer_connection copies to the destination optimistically and erases the source only on HANDOFF_REPLY, guarded by m_migrating.
5. Scaling is trial-and-observe. scale_trial picks a direction (flip if the last trial was fruitless, else persist), clamps count to headroom and window_size (default 4); statistics_scaling records a throughput score per tick and, at count == 0, commits toward the best via scale_selection’s top-5% band.
6. Two status flavors serialize structure changes. m_status (STABLE/DRAINING/EXPANDING) gates physical changes; scale_up restores STABLE synchronously and zeroes the new worker’s score so it absorbs arrivals, while scale_down stays DRAINING until scale_down_finish fires from the emptied worker’s next report.
7. Known inert path: css_get_task_stats is stubbed, so the task-completion term in both the score and the scaling throughput metric folds zero — restore it before tuning weights.

Chapter 7: Context Lifecycle and Per Worker Freelist

The reader question: how is a connection::context allocated and recycled without touching new/delete on the hot path? The companions (“The connection reactor”, “Why a freelist”) explain why a per-connection object must never hit the general allocator during a connect/disconnect storm; this chapter is the how — the exact ownership hops, every branch of claim/retire, and the close protocol that lazily returns a context to the pool. Three collaborators share the work:

• pool owns the memory: a LIFO of freelist nodes, each embedding one context. It exposes claim_context / retire_context.
• coordinator is the only thread that calls claim/retire, and it holds the pool’s m_mutex for its whole lifetime (lock_resource at the tail of initialize, release_resource at the head of finalize), so the freelist is single-writer by construction — no per-node lock exists.
• worker never allocates a context. It receives one already-claimed via a NEW_CLIENT message, tracks it in an m_context roster, and when the connection dies stages the pointer in m_removed_context for a batched RETURN_TO_POOL hand-back.

So “per-worker freelist” is really two layers: a shared, coordinator-serialized memory pool, plus a per-worker roster + retire-staging buffer.

7.1 The three structs

The freelist node. A node is a context plus a next-pointer, with the context deliberately first:

// pool::freelist -- src/connection/connection_pool.hpp
struct freelist {
  context m_context;   /* THIS MUST BE THE FIRST -- so reinterpret_cast<freelist*>(&m_context) round-trips */
  freelist *m_next;
  freelist (std::size_t capacity) : m_context (capacity), m_next (nullptr) {}
  bool prepare () { return m_context.prepare (); }   /* <- allocates the receiver backing store */
};
// ... the freelist control block, an anonymous struct member of pool ...
struct { freelist *m_head; std::size_t m_max; std::size_t m_claim; } m_freelist;

Field	Role	Why it exists
freelist::m_context	The recycled context payload	Placed first so retire_context recovers the node from a bare context* by a zero-offset reinterpret_cast.
freelist::m_next	Intrusive LIFO link	Threads free nodes without a side container; valid only while the node sits on m_head.
m_freelist.m_head	Top of the free stack	nullptr means the reserve is drained — claim then falls back to new.
m_freelist.m_max	Retain ceiling = max_connections * 1.1	Above this, retired nodes are freed instead of pooled, so a transient spike does not permanently bloat RSS.
m_freelist.m_claim	Count of contexts currently out	Compared against m_max to decide recycle-vs-free; asserted == 0 at finalize_freelist.

The connection::context struct — the object being recycled. Every field:

// connection::context -- src/connection/connection_context.hpp
struct context {
  css_conn_entry *m_conn;                   /* THIS MUST BE THE FIRST */
  int m_worker;  uint64_t m_id;
  ignore_level m_ignore;  bool m_removed;
  struct { state m_state; receiver m_receiver; NET_HEADER m_header;
           int m_request_id; bool m_command; } m_recv;
  struct { transmitter m_transmitter; std::shared_ptr<message_blocker> m_blocker; } m_send;
  statistics::metrics<statistics::context> m_stats;
};

Field	Role	Why it exists
m_conn	Back-pointer to css_conn_entry	Bridge to the legacy engine; m_conn->context points back here, so either side recovers the other. First member by contract.
m_worker	Index of the owning worker	Stamped by the coordinator at claim time; lets RETURN_TO_POOL decrement the right worker’s stats bucket.
m_id	Process-unique context id	Keys the coordinator’s per-worker stats maps and the worker’s m_exhausted map; survives handoff between workers.
m_ignore	ERR/HUP suppression level	Governs whether a dying connection still drains its send buffer (role matrix below).
m_removed	”Already torn down” flag	Idempotency guard so a duplicate SHUTDOWN_CLIENT cannot double-close.
m_recv.m_state	Reception sub-state (HEADER/DATA/ERROR)	Selects the packet handler in handle_packet; detailed in Ch 8.
m_recv.m_receiver	Owned receive buffer/parser	Holds partially-received packets; backing store allocated by prepare(), recycled (not freed) on reset.
m_recv.m_header	The parsed fixed-size packet header (NET_HEADER)	A 9-field struct packet_header (type/version/host_id/transaction_id/request_id/db_error/function_code/flags/buffer_size); default-initialised to DEFAULT_HEADER_DATA.
m_recv.m_request_id	Id of the in-flight request	Correlates a command header with its data body.
m_recv.m_command	”A command packet is mid-assembly”	When the data body completes, gates whether a task is pushed into the back-end pool.
m_send.m_transmitter	Owned send queue + iovec state	Buffers outbound bytes; drained or abandoned before recycle.
m_send.m_blocker	Optional message_blocker handle	When a task worker issues a blocking send, this cond-var wakes it once the bytes leave or the conn dies.
m_stats	Per-context metrics	Snapshotted to the coordinator each stats tick; reset on recycle.

INVARIANT — two stacked “first member” contracts. m_conn is first in context, and m_context is first in freelist. m_conn at offset 0 lets C-side code treat a context* and its css_conn_entry* interchangeably; m_context at offset 0 lets retire_context do reinterpret_cast<freelist*>(ctx) to reach the node header. Reorder either and the cast silently reads the wrong bytes — the freelist links corrupt, or css dereferences garbage.

The two enums. state { HEADER, DATA, ERROR } is the reception state machine (Ch 8 owns its transitions). ignore_level : uint8_t { DONT_IGNORE = 0, IGNORE_ALL } is a small ordered scale — code compares with <, so intermediate levels could slot between. Its meaning shifts by teardown phase:

m_ignore	On a healthy connection	During close
DONT_IGNORE	Normal I/O; errors surface	has_remaining_tasks still waits for the send buffer to drain before closing.
IGNORE_ALL	Set on EPOLLERR/RDHUP, PeerReset, protocol error	Skips the drain — the peer is gone, so has_remaining_tasks returns false immediately.

flowchart TD
  subgraph POOL["pool - coordinator-serialized"]
    H["m_freelist.m_head"] --> N1["freelist node<br/>m_context : context<br/>m_next"]
    N1 --> N2["freelist node<br/>m_context : context<br/>m_next = null"]
  end
  subgraph WK["worker - per-thread"]
    R["m_context : set of context*"]
    RM["m_removed_context : vector of context*"]
  end
  CO["coordinator<br/>holds m_mutex for life"]
  CO -- "claim / retire" --> H
  CO -- "NEW_CLIENT ctx" --> R
  RM -- "RETURN_TO_POOL batch" --> CO
  N1 -. "css_conn_entry.context" .-> R

Figure 7-1. The context lives inside a pool freelist node; the coordinator moves it into a worker’s roster and back.

7.2 Acquire: claim_context and the exhaustion fallback

claim_context runs only on the coordinator (which already holds m_mutex; the assert documents that). Every branch:

// pool::claim_context -- src/connection/connection_pool.cpp
context *pool::claim_context () {
  freelist *head;
  assert (m_mutex_holder == std::this_thread::get_id ());   /* <- caller must own m_mutex */
  head = m_freelist.m_head;
  if (head) { m_freelist.m_head = m_freelist.m_head->m_next; }  /* pop the LIFO */
  else {
    head = new freelist (32 * 1024);      /* <- reserve drained: grow */
    if (!head->prepare ()) {              /* allocate the receiver backing store */
      delete head;
      return nullptr;                     /* <- allocation failure: caller must cope with null */
    }
  }
  m_freelist.m_claim++;
  return &head->m_context;
}

1. Reserve hit (m_head != nullptr): pop the top node, advance m_head to m_next. No allocation — the hot path. The popped node’s m_next is now stale but harmless (overwritten on the next retire).
2. Reserve drained (m_head == nullptr): new freelist (32 * 1024) constructs a node whose context has a 32 KiB-capacity receiver object, then prepare() allocates that receiver’s backing buffer. This is the only place the hot path can allocate, and only when live connections exceed the pre-warmed reserve (initialize_freelist seeds m_max nodes up front, each also prepare()-d).
3. prepare() failure (out of memory on grow): delete head; return nullptr. m_claim is not incremented — the count stays honest. This is the freelist’s only null return.
4. Both success branches m_claim++ and return &head->m_context. Ownership of the node stays with the pool; the caller gets a borrowed context*.

Downstream of the null return. The only caller is coordinator::handle_message_queue_new_client. It checks the result and, on nullptr, css_free_conns the incoming connection and silently drops the client — this is the terminal case of exhaustion + allocation failure:

// coordinator::handle_message_queue_new_client -- src/connection/coordinator.cpp
ctx = m_parent->claim_context ();
if (!ctx) { css_free_conn (item.conn); return true; }   /* failed to allocate: just ignore */
// ... else: pick worker, stamp ctx->m_worker/m_id, enqueue NEW_CLIENT ...

On success the coordinator stamps the still-stale identity fields (request.ctx->m_worker = worker; request.ctx->m_id = id++;) before the context is any worker’s concern.

flowchart TD
  A["claim_context - holds m_mutex"] --> B{"m_freelist.m_head?"}
  B -- "non-null" --> C["head = m_head<br/>m_head = m_head->m_next"]
  B -- "null - drained" --> D["head = new freelist 32K"]
  D --> P{"head->prepare?"}
  P -- "false" --> Q["delete head<br/>return nullptr"]
  P -- "true" --> E
  C --> E["m_claim++"]
  E --> F["return &head->m_context"]
  Q -. "caller" .-> G["coordinator: css_free_conn; drop client"]

Figure 7-2. claim_context: pop, grow, or fail-null — the last handled by the coordinator’s if (!ctx).

7.3 Release: retire_context, rotation, and the shrink branch

Retire is the mirror. It reconstructs the node from the context pointer, scrubs the context, then decides whether the node rejoins the reserve or dies:

// pool::retire_context -- src/connection/connection_pool.cpp
void pool::retire_context (context *ctx) {
  freelist *head;
  assert (m_mutex_holder == std::this_thread::get_id ());
  head = reinterpret_cast<freelist *> (ctx);   /* <- valid only because m_context is first */
  head->m_context.reset ();                     /* scrub all fields before reuse */
  if (m_freelist.m_claim > m_freelist.m_max) { delete head; }   /* over ceiling: shrink */
  else { head->m_next = m_freelist.m_head; m_freelist.m_head = head; }  /* push back: rotate */
  m_freelist.m_claim--;
}

1. Node recovery: reinterpret_cast<freelist*>(ctx) — a no-op offset, legal precisely because m_context is first (7.1 invariant).
2. Scrub: head->m_context.reset () (7.4) returns every field to its constructed default so the next claimer sees a pristine object. This is what lets reuse skip a destructor/constructor round-trip.
3. Shrink branch (m_claim > m_max): live population is above the retain ceiling, so this node is surplus — delete head.
4. Rotate branch (else): push onto m_head (LIFO). “Rotation” is exactly this pop-in-claim / push-in-retire cycle — the same nodes circulate, so steady-state churn touches no allocator.
5. Both branches m_claim--.

INVARIANT — m_claim is the exact count of contexts handed out. Every claim_context success does +1, every retire_context does -1, both single-threaded on the coordinator, so no atomic is needed. finalize_freelist asserts m_claim == 0; a leaked context drifts the count and trips the assert at shutdown. It is also the sole input to the shrink decision — a drifted m_claim would leak memory (never shrinking) or thrash (new/delete every cycle).

initialize_freelist pre-builds the reserve (m_max = max_connections * 1.1 nodes, each new freelist (32*1024) then prepare(); if any prepare() fails it deletes the node and returns false). finalize_freelist walks m_head deleting every node after asserting the claim count is zero.

flowchart TD
  A["retire_context - holds m_mutex"] --> B["head = reinterpret_cast to freelist*"]
  B --> C["head->m_context.reset"]
  C --> D{"m_claim > m_max?"}
  D -- "yes - over ceiling" --> E["delete head"]
  D -- "no" --> F["head->m_next = m_head<br/>m_head = head"]
  E --> G["m_claim--"]
  F --> G

Figure 7-3. retire_context: scrub, then shrink-or-rotate.

7.4 context::reset — scrubbing for reuse

Reuse hinges on reset restoring the exact constructed state — re-seeding scalars, resetting the owned I/O machinery, zeroing stats:

// connection::context::reset -- src/connection/connection_context.cpp
void context::reset () {
  m_conn = nullptr; m_worker = -1; m_id = 0;
  m_ignore = ignore_level::DONT_IGNORE; m_removed = false;
  m_recv.m_state = state::HEADER; m_recv.m_receiver.reset ();   /* buffers recycled, not freed */
  m_recv.m_header = DEFAULT_HEADER_DATA;                        /* {0,0,0,NULL_TRAN_INDEX,0,0,0,0,0} */
  m_recv.m_request_id = -1; m_recv.m_command = false;
  m_send.m_transmitter.clear (); m_send.m_blocker = nullptr;    /* drop shared_ptr ref */
  m_stats.reset ();
}

This mirrors the two constructors field-for-field. Both ctors set the same scalars and m_recv.m_header = DEFAULT_HEADER_DATA; they differ only in how the receiver/transmitter are sized — the capacity ctor uses receiver (capacity, &m_stats) / transmitter (&m_stats), the default ctor uses receiver () / transmitter (). Because reset calls m_recv.m_receiver.reset () rather than freeing, a recycled context keeps its 32 KiB buffer warm — the allocator is untouched even for the byte buffers. Dropping m_send.m_blocker to nullptr releases the last shared_ptr reference to any blocked task’s handle; by the time close reaches retire, that task has already been woken (7.5).

7.5 Close protocol: handle_connection_close, the retry helper, and lazy return

A context leaves a worker only through handle_connection_close. It is the branchiest function in the subsystem: it must quiesce transactions, drain or abandon I/O, and it may retry itself via a lazy message rather than block the reactor. Branch-complete:

1. Idempotency guard: if (ctx->m_removed) — already torn down. Wake any waiter (wakeup_blocked_worker (handle)), log, return true. This makes a duplicate SHUTDOWN_CLIENT safe. Then assert_release (ctx->m_conn).
2. Registering-client early-retry. Before touching status, five conjuncts are tested: m_status != TERMINATING and !css_is_shutdowning_server () and requires_client_info (ctx) and ctx->m_conn->get_tran_index () == NULL_TRAN_INDEX and is_registering_client (ctx). If all hold, the client is still inside boot_client_register (no transaction index yet) but has a pending request or a working task (has_pending_request() || has_working_task()). Closing now would race the registration, so under rmutex the code sets ctx->m_conn->stop_talk = true (signals the registration path to abort), then return retry_connection_close (ctx, is_retry, handle) — defer and let the register finish. requires_client_info returns false only for the CDC internal fd (ctx->m_conn->fd == cdc_Gl.conn.fd), which has no client to wait on.
3. Status flip: under rmutex, CONN_OPEN → CONN_CLOSING; snapshot status.
4. Transaction-index acquisition (start_connection_close, under tran_index_lock): reads tran_index/client_id and, if valid, binds the worker’s m_entry to this conn via css_set_thread_info (..., NET_SERVER_SHUTDOWN). If tran_index == NULL_TRAN_INDEX it returns {-1,-1}. In that case, if requires_client_info (ctx), the retry was already skipped (step 2 didn’t fire), so registration will never publish a transaction index — recover client_id = ctx->m_conn->client_id and close without retry. There is no sleep and no re-read. Unlock tran_index_lock.
5. css_end_server_request; set entry status TS_CHECK (breaks a potential infinite xtran_wait_server_active_trans); if session_p is set, ssession_stop_attached_threads.
6. Non-retry wakeup: if (!is_retry) → net_server_wakeup_workers interrupts in-flight task threads. On a retry pass this is skipped (already interrupted).
7. Active-workers gate: net_server_active_workers (...) > 0 → some task thread still runs for this conn → goto retry.
8. Remaining-tasks gate: has_remaining_tasks (ctx) first drains the IMMEDIATE queue (a queued RELEASE_PACKET might free receiver memory), then — only if m_ignore < IGNORE_ALL — returns true if the transmitter is non-empty → goto retry. An IGNORE_ALL context skips this: a dead peer’s unsent bytes are abandoned.
9. Teardown (no remaining work), in source order: m_events.remove_descriptor (fd); if tran_index != NULL_TRAN_INDEX then net_server_conn_down; end_connection_close (below); then under cmutex (via rmutex_lock) null out m_conn->worker and m_conn->context; then m_send.m_transmitter.clear (); css_shutdown_socket (fd), fd = INVALID_SOCKET; wake the close waiter (handle) and any m_send.m_blocker; css_prepare_shutdown_conn.
10. Lazy release: ctx->m_removed = true; m_removed_context.push_back (ctx); — the context is not retired here; it is staged for the batched purge_stale_contexts. m_stats.sub (CLIENT_NUM, 1); return true.
11. retry: label: end_connection_close; return retry_connection_close (ctx, true, handle).

The nulling in step 9 happens before transmitter.clear (), so no message handler can bind new work to a context whose transmitter is being wiped.

INVARIANT — m_conn->context is nulled under cmutex before the context is staged for retire. Step 9 sets m_conn->context = nullptr under cmutex; every message handler that might race (send_packet, release_packet, shutdown_client) re-reads m_conn->context under the same cmutex and bails if null (7.6). Once the pointer is cleared, no new work binds to the context, so staging it for a later retire_context cannot race a concurrent use.

The shared retry helper. Both step 2 and the retry: label funnel into retry_connection_close, which never blocks the reactor: it builds a SHUTDOWN_CLIENT message (carrying conn, ctx, m_id, m_ignore, retry, and the waiter handle), enqueue (queue_type::LAZY, ...), sets m_has_retry = true, and arms a LOW_LATENCY QUEUE timer bound to handle_message_queue so the retry fires soon without spinning. Its eventfd_addtimer result is returned.

// worker::end_connection_close -- src/connection/connection_worker.cpp
void worker::end_connection_close () {
  pthread_mutex_lock (&m_entry->tran_index_lock);
  css_set_thread_info (m_entry, -1, 0, -1, -1);   /* detach client/tran from this entry */
  m_entry->conn_entry = NULL;
  m_entry->m_status = cubthread::entry::status::TS_RUN;
  pthread_mutex_unlock (&m_entry->tran_index_lock);
}

end_connection_close runs on both exits — teardown (step 9) and retry (step 11) — because either way the worker’s entry must stop impersonating this client before it moves on. It holds tran_index_lock because the entry lives in the thread-manager’s global list and other paths may inspect it.

purge_stale_contexts — the batched hand-back. The main loop and ha_close_all_connections call it after processing, so retire happens once per tick, not once per close:

// worker::purge_stale_contexts -- src/connection/connection_worker.cpp
message.type = coordinator::message_type::RETURN_TO_POOL;
message.resource = m_removed_context;              /* copy the batch to the coordinator */
for (context *ctx : m_removed_context) {
  if (m_context.erase (ctx) == 0) {                /* drop from the live roster */
    er_log_conn (...);  continue;                  /* not in roster: log, skip local cleanup */
  }
  m_exhausted.erase (ctx->m_id);                   /* also drop any IO-budget deferral */
}
m_removed_context.clear ();
m_coordinator->enqueue (std::move (message));
if (!m_coordinator->notify ()) { assert_release (false); }

Branch: if m_context.erase returns 0 the pointer was never in the roster (a bug or double-stage) — log and continue, though the pointer is still in the already-copied message.resource. The coordinator side, handle_message_queue_return_to_pool, is where memory actually returns: for each context it decrements m_statistics[m_worker].m_client_num, erases the context’s stats (m_id key) from every worker map, then css_free_conn (ctx->m_conn) and retire_context (ctx) (7.3). Only there — on the coordinator, under m_mutex — does the context re-enter the freelist.

flowchart TD
  A["handle_connection_close ctx"] --> B{"ctx->m_removed?"}
  B -- yes --> Z["wake waiter; return true"]
  B -- no --> C{"not TERMINATING and not shutdowning<br/>and requires_client_info<br/>and tran_index == NULL<br/>and is_registering_client?"}
  C -- yes --> RH["stop_talk = true<br/>retry_connection_close"]
  C -- no --> D["status -> CONN_CLOSING<br/>start_connection_close"]
  D --> E{"tran_index < 0?"}
  E -- "yes and requires_client_info" --> E2["client_id = m_conn->client_id<br/>close without retry"]
  E -- no --> F["end_server_request; TS_CHECK"]
  E2 --> F
  F --> G{"active_workers > 0?"}
  G -- yes --> RT["retry:"]
  G -- no --> H{"has_remaining_tasks?"}
  H -- yes --> RT
  H -- no --> I["remove fd; conn_down; end_connection_close<br/>null m_conn->worker/context under cmutex<br/>transmitter.clear; shutdown socket; wake waiters"]
  I --> J["m_removed=true<br/>m_removed_context.push_back<br/>CLIENT_NUM--"]
  RT --> RH

Figure 7-4. handle_connection_close: every gate leads to lazy-stage or self-retry, never a blocking spin.

7.6 The null-context guard (CBRD-26412)

Once step 9 nulls m_conn->context, a message enqueued before the close can still arrive referencing a conn whose context is gone. Three handlers — handle_message_queue_send_packet, handle_message_queue_release_packet, handle_message_queue_shutdown_client — recover the context by re-reading m_conn->context under cmutex and null-check it first:

// worker::handle_message_queue_shutdown_client -- src/connection/connection_worker.cpp
r = rmutex_lock (m_entry, &item.conn->cmutex);
ctx = reinterpret_cast<context *> (item.conn->context);
if (ctx == nullptr) {                       /* <- context already cleared by a prior close */
    rmutex_unlock (m_entry, &item.conn->cmutex);
    er_log_conn (..., "context is already cleared for conn = %p", item.conn);
    this->wakeup_blocked_worker (item.waiter_handle);
    return true;                            /* drop the stale request, do not deref null */
}

The send_packet variant additionally runs item.deleter () in the null branch (the buffer it would have queued must still be freed) and wakes item.waiter_handle; release_packet just logs and returns. Without this guard the handler would reinterpret_cast a null pointer and dereference it — a use-after-close crash whenever a SEND/RELEASE/SHUTDOWN lost the race with a concurrent close. Reading the pointer under the same cmutex step 9 used to clear it closes the window: the handler sees either a live context or a definitive nullptr, never a torn value.

Contrast the reactor’s own control-fd allocations: eventfd_register builds its internal context with a bare new context () and frees it with delete, and the master connector likewise uses new/delete. Those are one-per-process control fds, not per-client hot-path objects, so they intentionally bypass the freelist.

7.7 Chapter summary — key takeaways

1. No hot-path new/delete. A context is claimed by popping a pre-warmed freelist node and retired by pushing it back; the allocator is touched only when the reserve is exhausted (claim grows, and may fail to nullptr) or the live count exceeds m_max = max_connections * 1.1 (retire shrinks).
2. Two stacked “first member” contracts. context::m_conn at offset 0 bridges to css_conn_entry; freelist::m_context at offset 0 lets retire_context recover the node via reinterpret_cast. Reordering either corrupts the cast.
3. The freelist is single-writer, not locked per node. Only the coordinator calls claim/retire while holding the pool m_mutex; m_claim is a plain counter asserted to zero at shutdown.
4. Recycling means scrubbing. context::reset re-seeds every field to its constructed default (m_header = DEFAULT_HEADER_DATA) and recycles — not frees — the receiver/transmitter buffers, keeping the 32 KiB store warm.
5. Close is lazy and self-retrying. handle_connection_close never blocks the reactor: a still-registering client, active task threads, or a non-empty send buffer on a live peer all funnel through the shared retry_connection_close helper (a LAZY SHUTDOWN_CLIENT + low-latency timer); the actual retire is batched by purge_stale_contexts → RETURN_TO_POOL.
6. m_removed + null-under-cmutex make it race-safe. The context pointer is cleared under cmutex before staging (and before the transmitter is wiped); the CBRD-26412 guard re-reads it under the same lock and drops stale SEND/RELEASE/SHUTDOWN requests instead of dereferencing null.
7. ignore_level decides drain-vs-abandon. DONT_IGNORE makes close wait for the send buffer; IGNORE_ALL short-circuits has_remaining_tasks so a dead connection tears down immediately.

Chapter 8: From Bytes to a Task

Chapters 4–7 followed a connection worker (the reactor) as it drained a socket and reassembled a full request from epoll fragments. This chapter closes that arc: once a request is fully received, how does it cross out of the reactor and become a runnable job on the back-end task-worker pool — without the reactor ever blocking on query execution?

The crossing is deliberately thin: one wrapper (push_task_into_worker_pool), one free function (css_push_server_task), one heap task (css_server_task), and the generic pool enqueue (push_task_on_core → execute_on_core → core_impl::execute_task). The reactor never links against thread_worker_pool_impl.hpp; the whole pool hides behind the css_push_server_task symbol. The redesign rationale for that split lives in the companion’s Two pools, one process; here we trace the seam line by line, including the stopped-pool reject, the full-pool enqueue, and the separate wakeup_blocked_worker back-pressure handshake.

8.1 The two call sites

Two reactor states declare a request dispatchable, both via the same wrapper. A command header with no data body pushes immediately; a header with a body defers until the body arrives:

// worker::handle_command_header_packet -- src/connection/connection_worker.cpp
if (ntohl (header->buffer_size) > 0)
  ctx->m_recv.m_command = true;             /* <- defer; wait for the data packet */
else
  this->push_task_into_worker_pool (ctx);   /* <- e.g. LOG_CHECKPOINT: dispatch now */

// worker::handle_data_packet -- src/connection/connection_worker.cpp
if (ctx->m_recv.m_command)
  {
    this->push_task_into_worker_pool (ctx);
    ctx->m_recv.m_command = false;          /* <- exactly one dispatch per command/body pair */
  }
NEXT_STATE (ctx, m_recv, HEADER);           /* <- reactor rearms immediately, does not wait */

m_command is the anti-double-dispatch guard (header sets, data consumes-and-clears); NEXT_STATE rearms the recv machine in the same turn, so the reactor never waits for the pushed task.

8.2 worker::push_task_into_worker_pool — the one-line boundary

// worker::push_task_into_worker_pool -- src/connection/connection_worker.cpp
void worker::push_task_into_worker_pool (context *ctx)
{
  css_push_server_task (*ctx->m_conn);
}

Its brevity is the design. It dereferences ctx->m_conn (a CSS_CONN_ENTRY &) and forwards to the C-linkage css_push_server_task. The reactor class cubconn::connection::worker never names cubthread::worker_pool or css_server_task — this one symbol is its only pool import. No return value, no error check, no wait: the call completes as fast as an enqueue, the mechanical form of the §8.6 boundary invariant.

8.3 css_push_server_task — accounting then dispatch

// css_push_server_task -- src/connection/server_support.c
void css_push_server_task (CSS_CONN_ENTRY &conn_ref)
{
  // note: cores are partitioned by connection index ... to avoid pushing tasks to cores that are full.
  //       some of those tasks may belong to threads holding locks ...
  conn_ref.add_pending_request ();          /* <- ++pending_request_count (atomic) */
  conn_ref.add_working_task ();             /* <- ++working_task_count   (atomic) */
  thread_get_manager ()->push_task_on_core (css_Server_request_worker_pool, new css_server_task (conn_ref),
                                            static_cast<size_t> (conn_ref.idx), conn_ref.in_method);
}

Two atomic counters are bumped first. Both live on CSS_CONN_ENTRY as std::atomic<size_t>, touched only through member functions:

Counter	Bumped by (reactor)	Drained by (pool)	Role
pending_request_count	add_pending_request	start_request	Requests received but not yet begun executing.
working_task_count	add_working_task	end_working_task	Tasks in flight; its drain to zero gates teardown of a CONN_CLOSING connection.

They increment on the reactor before the push, so the pool never sees a live task whose connection counters read zero.

The core is chosen by connection index, not round-robin. conn_ref.idx becomes core_hash; the comment gives the reason — hashing a lock-holding task onto a full core could stall unrelated lock waiters for a full queue drain, so every request of one connection pins to one core. (Plain push_task, used by css_push_external_task, instead takes whatever get_next_core round-robins.)

The css_server_task is heap-allocated and its ownership transfers to the pool — from here the pool must delete it (§8.4). The last argument conn_ref.in_method (true for a method/SP callback channel) becomes the pool’s is_temp flag, steering the no-worker branch of §8.5 to a temp thread so a callback never wedges behind a full core.

8.4 css_server_task — the task object and the thread-entry handoff

// css_server_task -- src/connection/server_support.c
class css_server_task : public cubthread::entry_task
{
public:
  css_server_task (void) = delete;                 /* <- must carry a connection */
  css_server_task (CSS_CONN_ENTRY &conn) : m_conn (conn) {}
  void execute (context_type &thread_ref) override final;
  // retire not overwritten; task is automatically deleted
private:
  CSS_CONN_ENTRY &m_conn;                          /* <- only state: which connection this runs for */
};

cubthread::entry_task is the alias task<entry>, so context_type is a cubthread::entry — the pooled per-thread THREAD_ENTRY. The task’s sole state is the reference m_conn. It does not override retire, inheriting the base delete this after execute — the counterpart to the new in §8.3, and why the pushing side must never touch the pointer again.

execute binds the connection to a pool thread and runs the request:

// css_server_task::execute -- src/connection/server_support.c
thread_ref.conn_entry = &m_conn;                   /* <- graft connection onto this pool thread */
session_p = thread_ref.conn_entry->session_p;
m_conn.start_request ();                            /* <- --pending_request_count */
if (session_p != NULL)
  { thread_ref.private_lru_index = session_get_private_lru_idx (session_p); pgbuf_thread_variables_init (&thread_ref); }
thread_ref.m_status = cubthread::entry::status::TS_RUN;
pthread_mutex_lock (&thread_ref.tran_index_lock);
(void) css_internal_request_handler (thread_ref, m_conn);   /* <- the actual query dispatch */
thread_ref.conn_entry = NULL;
thread_ref.m_status = cubthread::entry::status::TS_FREE;
if (m_conn.end_working_task () == 0 && m_conn.status == CONN_CLOSING)   /* <- last task out of a closing conn */
  css_request_shutdown_conn (&m_conn, /*DONT_IGNORE*/, false, 0 /* no wait */);
else
  css_wakeup_handler (&m_conn);                     /* <- otherwise nudge the reactor to flush replies */

end_working_task returns the post-decrement count. Its two branches: if it hits zero and the connection is CONN_CLOSING, this task was the last one out, so it initiates shutdown itself (no wait, to avoid blocking a pool thread on teardown); otherwise it calls css_wakeup_handler to poke the reactor’s message queue so replies produced by the handler get flushed.

Invariant — a request’s counters bracket exactly one pool execution. The increments run once on the reactor before the push; the decrements run once on the pool inside execute, and the push happens-before the execute (release/acquire inside the pool mutex) so they are visible. Push without both increments, or run execute twice, and working_task_count never returns to zero — a CONN_CLOSING connection then leaks, never reaching css_request_shutdown_conn.

flowchart TB
  subgraph reactor["reactor thread"]
    W["push_task_into_worker_pool"] --> C["css_push_server_task"]
    C -->|"new"| T["css_server_task<br/>holds CSS_CONN_ENTRY &"]
    C -->|"++counters"| CE["CSS_CONN_ENTRY<br/>pending_request_count<br/>working_task_count"]
  end
  C -->|"push_task_on_core, ownership moves"| POOL["request pool<br/>core = conn.idx % ncores"]
  subgraph backend["back-end pool thread"]
    POOL -->|"execute (entry &)"| EX["css_server_task::execute<br/>binds conn_entry, runs handler"]
    EX -->|"delete this"| GONE["task retired"]
  end
  T -.->|"references"| CE
  EX -.->|"decrements"| CE

Figure 8-1 — One push creates a heap css_server_task owned by the pool and referencing the connection; the pool thread binds it to a cubthread::entry, runs the handler, decrements the counters, and self-deletes.

8.5 push_task_on_core → execute_on_core → execute_task — every branch

push_task_on_core is the manager façade with one guard branch:

// manager::push_task_on_core -- src/thread/thread_manager.cpp
if (worker_pool_arg == NULL)
  { exec_p->execute (get_entry ()); exec_p->retire (); }   /* <- inline; SA_MODE / null pool only */
else
  { check_not_single_thread (); worker_pool_arg->execute_on_core (exec_p, core_hash, method_mode); }

For cub_server the pool is always non-null, so control reaches execute_on_core, which is pure arithmetic — reduce the index to a slot and forward: core_index = core_hash % m_cores.size (); m_cores[core_index]->execute_task (work_arg, is_temp);.

core_impl::execute_task holds all real branching:

// worker_pool_impl<Stats>::core_impl::execute_task -- src/thread/thread_worker_pool_impl.hpp
if (!m_parent_pool->is_running ())          /* branch 1: pool stopped */
  { task_p->retire (); return; }            /* <- delete and drop silently */
wrapped_task task_ref (task_p);
std::unique_lock<std::mutex> ulock (m_workers_mutex);   /* <- guards m_available_workers */
if (!m_available_workers.empty ())          /* branch 2: idle worker parked */
  {
    refp = static_cast<worker_impl *> (m_available_workers.back ());
    m_available_workers.pop_back ();
    ulock.unlock ();                        /* <- unlock before notify */
    refp->assign_task (std::move (task_ref));
  }
else if (is_temp)                           /* branch 3: no worker, temp task */
  { ulock.unlock (); execute_task_as_temp (std::move (task_ref)); }
else                                        /* branch 4: no worker, normal task */
  m_task_queue.push (std::move (task_ref));

• Branch 1 — stopped pool (reject). is_running() is !m_stopped (std::atomic<bool>, flipped during shutdown). The task is retired and dropped — the only path that discards a received request, reachable only at shutdown. The §8.3 counters are intentionally not undone; the connection is torn down wholesale.
• Branch 2 — idle worker. Pop a parked worker_impl, release m_workers_mutex, then assign_task. Unlock-before-notify keeps the critical section short and avoids waking a thread that would block on the mutex.
• Branch 3 — no worker, temp task. When is_temp (from in_method), execute_task_as_temp spins up a throwaway worker with its own OS thread, so method/SP callbacks progress even when all steady workers are busy — the guard against re-entrant SP self-deadlock.
• Branch 4 — no worker, normal task (full pool). The task is moved into m_task_queue. A full pool queues; it neither blocks the pusher nor drops the task; a steady worker pulls it on finishing its current task. This is how the reactor’s push stays non-blocking under overload — back-pressure becomes queue depth.

assign_task completes the handoff into a worker’s private slot and applies the lazy-thread policy:

// worker_pool_impl<Stats>::core_impl::worker_impl::assign_task -- src/thread/thread_worker_pool_impl.hpp
std::unique_lock<std::mutex> ulock (m_task_mutex);
m_wrapped_task.emplace (std::move (task_ref));   /* <- m_task_mutex guards this optional slot */
if (m_is_temp) { m_has_thread = true; start_thread (); }
if (m_has_thread) { ulock.unlock (); m_task_cv.notify_one (); }   /* <- wake parked worker (Ch 10) */
else { m_has_thread = true; ulock.unlock (); start_thread (); }   /* <- first task: create OS thread */

The OS thread is created only on the first task assigned to a slot; afterward the slot keeps its thread and assign only notifies m_task_cv. Chapter 10 picks up the run loop on the other side of that condition variable.

flowchart TB
  S["execute_task (task, is_temp)"] --> R{"pool is_running?"}
  R -- no --> RJ["retire; drop; return"]
  R -- yes --> L["lock m_workers_mutex"]
  L --> A{"available worker?"}
  A -- yes --> P["pop; unlock;<br/>assign_task -> notify"]
  A -- no --> TE{"is_temp?"}
  TE -- yes --> TT["unlock;<br/>spawn temp thread"]
  TE -- no --> Q["m_task_queue.push<br/>(full pool: queue, do not block)"]

Figure 8-2 — core_impl::execute_task decision tree. The stopped-pool reject (left) is the only task-dropping path; the full-pool case (bottom right) queues rather than blocking the reactor.

8.6 The boundary invariant — the reactor never blocks on execution

Invariant — the reactor thread never waits for a request to execute. push_task_into_worker_pool returns after an enqueue at worst (§8.5 branch 4) and never runs css_internal_request_handler on the reactor thread. Execution happens exclusively on request-pool threads inside css_server_task::execute; the reactor’s only feedback is asynchronous (css_wakeup_handler posting to its message queue). Violate this — e.g. by running the handler inline under SERVER_MODE — and one slow query freezes the epoll loop and stalls every other connection multiplexed onto that reactor worker. This is exactly the head-of-line blocking the reactor/back-end split exists to prevent (companion: Why two pools).

8.7 wakeup_blocked_worker and the back-pressure handshake

The task push of §8.2–8.5 is fire-and-forget — css_server_task carries no blocker. But some control operations (a SEND_PACKET whose socket buffer is full, START, SHUTDOWN_CLIENT) need the originating thread to wait until the reactor acts. That handshake runs through message_blocker, released by wakeup_blocked_worker:

// worker::wakeup_blocked_worker -- src/connection/connection_worker.cpp
void worker::wakeup_blocked_worker (std::shared_ptr<message_blocker> handle)
{
  if (handle)                                 /* <- null handle = fire-and-forget, nothing to wake */
    {
      std::lock_guard<std::mutex> lock (handle->m);
      handle->done = true;                    /* <- set predicate under the lock */
      handle->cv.notify_one ();
    }
}

message_blocker is a three-field rendezvous shared via shared_ptr so it outlives whichever side finishes first:

Field	Role	Why it exists
m (std::mutex)	Guards done, pairs with cv.	Serializes set-predicate and wait-check so a wakeup can’t slip between them.
cv (std::condition_variable)	Wait/notify channel.	Lets the blocked thread sleep instead of spin.
done (bool)	Predicate: request handled.	Distinguishes a real wakeup from a spurious one; cv.wait re-checks it.

The wait side lives in enqueue_and_notify. With a non-zero wait_time it builds the blocker and takes the lock before enqueuing — the lost-wakeup guard:

// worker::enqueue_and_notify -- src/connection/connection_worker.cpp
if (wait_time)
  {
    handle = std::make_shared<message_blocker> (); handle->done = false;
    lock = std::unique_lock<std::mutex> (handle->m);   /* <- hold m before the item is visible */
    item.waiter_handle = handle;
  }
this->enqueue (type, std::move (item));
if (!this->notify ()) { ... return false; }            /* <- notify failed: give up, never wait */
if (wait_time)
  {
    if (wait_time < 0) handle->cv.wait (lock, [&] { return handle->done; });
    else handle->cv.wait_for (lock, std::chrono::seconds (wait_time), [&] { return handle->done; });
  }

Holding m from before enqueue until the wait means a reactor that processes the message and calls wakeup_blocked_worker in between blocks on m until the waiter is parked on cv; the predicate lambda absorbs spurious wakes. The failure branch matters: if notify() fails the message is enqueued but unsignalable, so the function returns false without waiting — waiting on a blocker no one will release would hang forever.

The reactor fires wakeup_blocked_worker wherever the awaited outcome settles, and — critically — on every teardown exit so a waiter is never orphaned. handle_connection_close releases both the close handle and the send blocker:

// worker::handle_connection_close -- src/connection/connection_worker.cpp
this->wakeup_blocked_worker (handle);                 /* <- release the close waiter */
this->wakeup_blocked_worker (ctx->m_send.m_blocker);  /* <- and any thread blocked on an unsent packet */

The send blocker is moved into the context when a SEND_PACKET message is dequeued (ctx->m_send.m_blocker = std::move (item.waiter_handle)), so whether the send succeeds, the epoll re-arm fails, or the connection closes underneath it, exactly one wakeup_blocked_worker (ctx->m_send.m_blocker) site releases the waiter. This synchronous valve is distinct from, and complementary to, the fire-and-forget task push that carries the query.

8.8 css_get_task_stats — the disabled hook

// css_get_task_stats -- src/connection/server_support.c
void css_get_task_stats (UINT64 *stats_out)
{
//  css_Server_request_worker_pool->get_task_stats (stats_out);
}

The body is commented out — in this revision the hook is a no-op stub that leaves stats_out untouched, so a caller that doesn’t pre-zero its buffer reads stale bytes. A reader wiring up task metrics must know this accessor is dead: the live per-core scan for SHOW JOB QUEUES goes through css_wp_core_job_scan_mapper/get_stats, and Chapter 11 covers the atomic free-statistics that are maintained.

8.9 Chapter summary — key takeaways

1. The seam is four thin hops. A completed request calls the one-line push_task_into_worker_pool → css_push_server_task → new css_server_task → push_task_on_core. The reactor never names a pool internal.
2. The task is owned by the pool. css_server_task holds a single CSS_CONN_ENTRY & and does not override retire, so the new in css_push_server_task and the base delete this bracket its life.
3. Counters bracket exactly one execution. add_pending_request / add_working_task run on the reactor before the push; start_request / end_working_task run on the pool inside execute. working_task_count reaching zero is what lets a CONN_CLOSING connection tear down.
4. execute_task has four branches. Stopped pool → drop (shutdown only); idle worker → assign and notify; no worker + temp → spawn temp thread; no worker + normal → queue. Only the stopped branch drops; the full-pool branch queues rather than blocking the pusher.
5. The boundary invariant is mechanical. The push returns after an enqueue at worst and never runs the handler inline, so one slow query cannot freeze the epoll loop multiplexing many connections.
6. wakeup_blocked_worker is a separate synchronous channel. For the control/send paths that must block, message_blocker (mutex + cv + done) plus lock-before-enqueue closes the lost-wakeup window, and teardown always fires it so no waiter is orphaned. The query push never uses it.
7. css_get_task_stats is a stub — its body is commented out; task stats come from the per-core scan, not this accessor.

Chapter 9: Back End Pool Engine I Structure and Dispatch

The connection reactor of Chapters 1–8 turns bytes into a task and calls execute() on a back-end pool. This chapter opens that pool. The reader question is concrete: inside the back-end pool, how is work partitioned across cores and handed to a free worker? The answer is a three-tier nesting — worker_pool_impl owns cores, each core_impl owns workers, each worker_impl owns one OS thread and one context — plus two selection steps: pick a core, then pick a worker. Chapter 10 takes over at the worker’s run() loop; here we stop when a task lands in a worker’s m_wrapped_task.

Cross-link: why the pool shards into cores (single-queue cache-line contention, C10K, pool sizing) is in cubrid-thread-worker-pool.md, section “Core partitioning”; why the redesign leaves this engine intact is in cubrid-thread-manager.md, section “The two-layer split”.

Everything is templated on template <bool Stats>; that boolean makes the statistics machinery either real (worker_pool<true>) or an empty, if constexpr-elided base. Prose drops the <Stats> suffix.

9.1 The three-tier partition

Figure 9-1 is the ownership skeleton: containment (who unique_ptr-owns whom) plus two non-owning back-pointers (m_parent_pool, m_parent_core) that let a leaf reach its root.

flowchart TD
  WP["worker_pool_impl&lt;Stats&gt;<br/>m_cores : vector&lt;unique_ptr&lt;core&gt;&gt;<br/>m_max_workers, m_round_robin_counter, m_stopped"]
  C0["core_impl #0<br/>m_workers / m_available_workers / m_task_queue<br/>m_workers_mutex"]
  C1["core_impl #1<br/>..."]
  W0["worker_impl<br/>m_context_p / m_wrapped_task<br/>m_task_cv / m_task_mutex / m_has_thread"]
  W1["worker_impl<br/>..."]
  WP -->|owns| C0
  WP -->|owns| C1
  C0 -->|owns| W0
  C0 -->|owns| W1
  W0 -.->|m_parent_core| C0
  C0 -.->|m_parent_pool| WP

Figure 9-1 — ownership (solid) and back-pointer (dashed) edges.

The pool selects a core (9.6), the core selects a worker or queues the task (9.7), and the worker owns the running thread. Two mutexes live at two tiers: core_impl::m_workers_mutex guards the per-core lists, worker_impl::m_task_mutex guards one worker’s handoff slot. Keeping their scopes disjoint is the whole correctness story.

INVARIANT — worker/core partition is fixed at init. Every worker belongs to one core for life. assign_workers_to_cores gives the first remainder cores quotient + 1 and the rest quotient, so per-core counts sum to m_max_workers. Only tasks move at runtime; break the sum and get_round_robin_core_hash’s proportional dispatch skews.

9.2 worker_pool_impl — the root

// worker_pool_impl<Stats> -- src/thread/thread_worker_pool_impl.hpp
template <bool Stats>
class worker_pool_impl : public worker_pool
{
  std::vector<std::unique_ptr<core>> m_cores;   /* the cores, owned */
  std::size_t                        m_max_workers;         /* cap across all cores */
  std::atomic<bool>                  m_stopped;             /* one-way latch */
  std::atomic<std::size_t>           m_round_robin_counter; /* dispatch cursor */
};

Field	Role	Why it exists
m_cores	Vector of owned cores; index is the “core hash” domain	Middle tier that shards the worker set; unique_ptr so cores destruct with the pool
m_max_workers	Total concurrent-worker cap (sum over cores)	Bounds thread count; also the wrap modulus of the round-robin cursor (9.6)
m_stopped	false until stop_execution, then true forever	is_running() gate read by execute_task; atomic so any producer sees the latch
m_round_robin_counter	Monotonic dispatch cursor in [0, m_max_workers)	Default core-selection policy; atomic CAS makes concurrent execute() lock-free

Four base-class fields (worker_pool, in thread_worker_pool.hpp) matter downstream: m_name (thread name), m_entry_manager (mints/retires the context — Ch 7), m_pool_threads (prestart at init vs on first task), m_idle_timeout (Ch 10’s condvar wait bound).

INVARIANT — m_stopped is a one-way latch. stop_execution does m_stopped.exchange(true); whoever sees the prior false owns teardown. It never clears, so execute_task treats “not running” as terminal and retires the task. A racing producer reading stale false enqueues one task that retire_queued_tasks later drops — safe, not a leak.

9.3 core_impl — the middle tier

// worker_pool_impl<Stats>::core_impl -- src/thread/thread_worker_pool_impl.hpp
template <bool Stats>
class worker_pool_impl<Stats>::core_impl : public worker_pool::core
{
  std::vector<std::unique_ptr<worker>> m_workers;          /* pooled workers, owned */
  std::vector<worker *>                m_available_workers; /* idle & ready (non-owning) */
  std::queue<wrapped_task>             m_task_queue;        /* overflow when none idle */
  mutable std::mutex                   m_workers_mutex;     /* guards the three above */

  std::vector<std::unique_ptr<worker>> m_temp_workers;      /* throwaway SP/method workers */
  std::vector<std::unique_ptr<worker>> m_free_temp_workers; /* finished temps, pending join */
  mutable std::mutex                   m_temp_workers_mutex;/* guards the two temp lists */
};

Field	Role	Why it exists
m_workers	The core’s fixed roster of pooled workers	Owning store; size never changes after initialize
m_available_workers	Raw pointers to workers currently idle and ready	The free-list execute_task pops from; raw because m_workers owns
m_task_queue	FIFO of tasks that arrived with no idle worker	Absorbs bursts without blocking the producer
m_workers_mutex	Serializes the three lists above	Makes pop-worker / push-task atomic; mutable so const observers can lock
m_temp_workers	Non-pooled workers for a nested SP/method task	A task running inside a task must not steal the running worker; a temp thread runs it (9.7)
m_free_temp_workers	Temps that finished and parked here	Deferred destruction — a temp cannot delete its own unique_ptr while its thread runs
m_temp_workers_mutex	Guards both temp lists	Separate lock so temp churn never contends with the hot pooled path

Inherited from worker_pool::core: m_parent_pool (reaches is_running() and the entry manager) and m_pool_threads.

INVARIANT — availability bounds membership. m_available_workers.size() <= m_workers.size() always; both get_task_or_become_available and become_available assert it after push_back. A worker appears at most once — popped before assign (9.7), re-pushed only when its thread loops back idle (Ch 10). Double-listing would let two tasks clobber one worker’s single m_wrapped_task.

9.4 worker_impl — the leaf that owns a thread

// worker_pool_impl<Stats>::core_impl::worker_impl -- src/thread/thread_worker_pool_impl.hpp
template <bool Stats>
class worker_pool_impl<Stats>::core_impl::worker_impl : public worker_pool::core::worker
{
  context_type               *m_context_p;   /* execution context, lives with the thread */
  std::optional<wrapped_task> m_wrapped_task; /* the one pending/current task */
  std::condition_variable     m_task_cv;      /* wake on assign or stop */
  std::mutex                  m_task_mutex;   /* guards m_wrapped_task / m_has_thread / m_stop */
  bool                        m_stop;         /* pool asked this worker to stop */
  bool                        m_has_thread;   /* an OS thread is live for this worker */
  bool                        m_is_temp;      /* throwaway temp worker */
  stats_base                  m_stats;        /* empty base unless Stats */
};

Field	Role	Why it exists
m_context_p	Pointer to the thread_entry context bound to the thread	Reused across tasks; created in init_run, retired in finish_run (Ch 7/10)
m_wrapped_task	std::optional holding the single assigned task	Handoff slot; has_value() is the “I have work” predicate the condvar waits on
m_task_cv	Condition variable the parked thread sleeps on	Turns assign into a wake instead of a spin; also woken on stop
m_task_mutex	Per-worker lock for the handoff	Scope is one worker, so assigns to different workers never contend
m_stop	Set true when the pool tears down	Breaks the condvar predicate so the thread exits instead of waiting
m_has_thread	True while an OS thread services this worker	The false→true→false flip that decides wake vs spawn in assign_task
m_is_temp	Marks a non-pooled temp worker	Temp path skips the condvar: run once, then self-park in the free-temp list
m_stats	stats_base — empty when Stats==false	Zero-size base under EBO when off; real statset when on

Inherited: m_parent_core (reaches the queue, entry manager, pool name).

INVARIANT — one task per worker. assign_task asserts !m_wrapped_task.has_value() before emplace. A worker gets a task only after being popped from m_available_workers (no second producer can target it), or, for a queued task, inside its own run loop. Violating the optional drops a task.

9.5 wrapped_task and the stats family

wrapped_task is a move-only envelope around the raw task_type *. Its payload is chosen at compile time so the non-stats build carries no timing.

// worker_pool_impl<Stats>::wrapped_task -- src/thread/thread_worker_pool_impl.hpp
struct task_only       { task_type *task; };
struct task_with_stats { task_type *task; cubperf::time_point time; };
using inner_type = typename std::conditional_t<Stats, task_with_stats, task_only>;
// ...
inner_type m_inner;   /* the only data member */

Field / type	Role	Why it exists
task_only::task	Raw task pointer, no timing	The whole envelope when Stats==false
task_with_stats::task	Same pointer	Payload when Stats==true
task_with_stats::time	Enqueue timestamp (cubperf::time_point)	Lets stats measure queue-to-execute latency
inner_type	conditional_t<Stats, ...> alias	Picks the layout at compile time — no runtime branch
m_inner	The sole member	Keeps wrapped_task a thin, movable wrapper

INVARIANT — move-only, no double-retire. Copy ctor/assign are = delete; the move ctor nulls the source and the destructor asserts m_inner.task == nullptr. So a live envelope owns the task once, and retire() nulls task after task->retire(), making any second retire an assert-caught no-op. Every consuming path uses std::move.

The stats family is three cooperating types. stats is an all-static policy class holding the metric taxonomy and helpers (each guarded by if constexpr (Stats)); stats_base is the per-worker storage that the worker_pool<true> specialization fills and the primary template leaves empty. Chapter 11 dissects accumulation; here it is enough that m_stats costs nothing in the false build.

Field / type	Role	Why it exists
stats::id	enum class id : cubperf::stat_id, values start_thread..retire_context (0–7) then type_count sentinel	Names the eight lifecycle metrics; type_count sizes the definition array
stats::statdef	static const cubperf::statset_definition — one COUNTER_AND_TIMER pair per id	The compile-time metric registry (a count and a duration per event)
stats_base (primary)	Empty class — zero-size under EBO when Stats==false	Makes m_stats free in the non-stats build
stats_base<true>::statset	cubperf::statset *, owned; deleted in stats::destroy	Live per-worker counter/timer storage
stats_base<true>::time	cubperf::time_point clock baseline	Anchor for the “time since last event” deltas Ch 11 accumulates

9.6 Selecting a core — execute and get_round_robin_core_hash

execute is two lines: choose a core index, then dispatch onto it.

// worker_pool_impl<Stats>::execute -- src/thread/thread_worker_pool_impl.hpp
void worker_pool_impl<Stats>::execute (task_type *work_arg)
{
  execute_on_core (work_arg, get_next_core ());   /* policy -> core index */
}
// execute_on_core:
core_index = core_hash % m_cores.size ();          /* fold hash into range */
m_cores[core_index]->execute_task (work_arg, is_temp);

get_next_core is virtual, defaulting to get_round_robin_core_hash (Ch 6 subclasses override the policy). The counter wraps at m_max_workers, not at m_cores.size() — deliberate.

// worker_pool_impl<Stats>::get_round_robin_core_hash -- src/thread/thread_worker_pool_impl.hpp
while (true)
  {
    index = m_round_robin_counter;
    next_index = index + 1;
    if (next_index == m_max_workers)   /* <- wrap on TOTAL workers, not core count */
      {
        next_index = 0;
      }
    if (m_round_robin_counter.compare_exchange_strong (index, next_index))
      {
        break;                         /* <- won the CAS, index is mine */
      }
    /* lost the race: CAS reloaded `index`, retry */
  }
return index;                          /* caller does index % m_cores.size() */

Figure 9-2 traces both branches of the CAS loop.

flowchart TD
  A["read index = m_round_robin_counter"] --> B["next = index + 1"]
  B --> C{"next == m_max_workers ?"}
  C -->|yes| D["next = 0 (wrap)"]
  C -->|no| E["keep next"]
  D --> F{"CAS(counter, index -> next) ?"}
  E --> F
  F -->|success| G["return index"]
  F -->|failed race| A
  G --> H["execute_on_core: core = index % m_cores.size()"]

Figure 9-2 — get_round_robin_core_hash CAS loop feeding execute_on_core.

The cursor’s period is m_max_workers and the fold modulus is m_cores.size(), so each core is visited in proportion to its worker count — e.g. 15 workers over 4 cores fold 0..14 under % 4 to give cores 0–2 four visits and core 3 three visits per cycle:

Cursor 0..14	% 4 sequence	Core 0	Core 1	Core 2	Core 3
one full period	0 1 2 3 0 1 2 3 0 1 2 3 0 1 2	4	4	4	3

INVARIANT — dispatch is proportional to core size. Cursor period m_max_workers and fold modulus m_cores.size() give each core a task share equal to its worker share — but only because m_max_workers is the sum of per-core counts (9.1). Wrapping at m_cores.size() would share equally and overload the small core. The strong CAS claims each index once; visibility ordering is irrelevant, so seq_cst is incidental.

9.7 Dispatch inside a core — core_impl::execute_task

Once a core is chosen, execute_task routes the task to one of three destinations — idle worker, fresh temp thread, or overflow queue — after a reject check. Figure 9-3 is branch-complete.

flowchart TD
  S["execute_task(task_p, is_temp)"] --> A{"parent pool is_running() ?"}
  A -->|no| R["task_p->retire(); return"]
  A -->|yes| B["wrap task; lock m_workers_mutex"]
  B --> C{"m_available_workers empty ?"}
  C -->|no| D["refp = back(); pop_back()<br/>unlock; refp->assign_task(move)"]
  C -->|yes| E{"is_temp ?"}
  E -->|yes| F["unlock; execute_task_as_temp(move)"]
  E -->|no| G["m_task_queue.push(move)<br/>(still holding lock)"]

Figure 9-3 — core_impl::execute_task destinations.

// worker_pool_impl<Stats>::core_impl::execute_task -- src/thread/thread_worker_pool_impl.hpp
assert (task_p != nullptr);
if (!m_parent_pool->is_running ())
  {
    task_p->retire ();     /* REJECT: pool stopping, never enqueue */
    return;
  }
wrapped_task task_ref (task_p);
std::unique_lock<std::mutex> ulock (m_workers_mutex);
if (!m_available_workers.empty ())
  {
    refp = static_cast<worker_impl *> (m_available_workers.back ());
    m_available_workers.pop_back ();
    ulock.unlock ();       /* release BEFORE assign to avoid lock nesting */
    refp->assign_task (std::move (task_ref));   /* FAST: hand to idle worker */
  }
else
  {
    if (is_temp)
      {
        ulock.unlock ();   /* drop m_workers_mutex before spawning temp */
        execute_task_as_temp (std::move (task_ref));  /* TEMP path */
      }
    else
      {
        m_task_queue.push (std::move (task_ref));     /* QUEUE path, lock held */
      }
  }

Two subtleties the inline comments flag: the fast path unlocks m_workers_mutex before assign_task (which takes the worker’s m_task_mutex) so the two locks never nest — no core-vs-worker deadlock; the queue path pushes while still holding m_workers_mutex, so enqueue is atomic against a worker’s get_task_or_become_available (Ch 10), which checks the queue under the same lock. The temp path serves a nested SP/method task via a throwaway worker_impl so it never waits behind the worker blocked on it.

INVARIANT — a task is claimed under exactly one lock hold. Pop-worker and push-queue are mutually exclusive under m_workers_mutex; combined with get_task_or_become_available checking the queue before adding to the free-list under the same lock, no window lets a task and an idle worker miss each other. The lost-wakeup is closed by the shared lock.

9.8 Handing off to a worker — assign_task and become_available

assign_task(wrapped_task &&) is where the task crosses from core into worker: place the task, then wake a parked thread or spawn a new one — the choice hinges on m_has_thread.

// worker_impl::assign_task(wrapped_task &&) -- src/thread/thread_worker_pool_impl.hpp
std::unique_lock<std::mutex> ulock (m_task_mutex);
assert (!m_wrapped_task.has_value ());
m_wrapped_task.emplace (std::move (task_ref));   /* place the task */
if (m_is_temp)
  {
    m_has_thread = true;
    assert (m_context_p == nullptr);
    start_thread ();                             /* temp: always a fresh thread */
  }
if (m_has_thread)
  {
    ulock.unlock ();                             /* WAKE: notify outside the lock */
    m_task_cv.notify_one ();
  }
else
  {
    m_has_thread = true;
    ulock.unlock ();
    assert (m_context_p == nullptr);
    start_thread ();                             /* SPAWN: first task, new thread */
  }

The temp branch always start_threads (and its trailing notify_one is a harmless no-op — a temp thread never waits on m_task_cv, Ch 10). The wake branch unlocks before notify_one so the woken thread does not immediately re-block on the just-held mutex. The spawn branch flips m_has_thread under the lock, unlocks, then start_threads a detached thread running worker_impl::run.

The prestart overload assign_task(void) assigns no task — it just guarantees a live parked thread. It is used to prestart pooled threads: from initialize_workers when m_pool_threads is set, and from warmup, which prestarts any available worker that lacks a thread.

// worker_impl::assign_task(void) -- src/thread/thread_worker_pool_impl.hpp
assert (!m_wrapped_task.has_value ());
assert (!m_is_temp);                 /* never prestart a temp */
if (m_has_thread)
  {
    ulock.unlock ();                 /* already has a thread: nothing to do */
  }
else
  {
    m_has_thread = true;
    ulock.unlock ();
    assert (m_context_p == nullptr);
    start_thread ();                 /* spawn a thread that will block waiting */
  }

A prestarted thread enters run with no task, finds the queue empty, and parks — so the first real task takes the cheap wake path instead of paying thread-creation latency on the hot path.

become_available is the inverse micro-op, used by a stopping worker (m_stop set) that must still re-declare availability for consistent teardown accounting:

// core_impl::become_available -- src/thread/thread_worker_pool_impl.hpp
std::unique_lock<std::mutex> ulock (m_workers_mutex);
m_available_workers.push_back (&worker_arg);
assert (m_available_workers.size () <= m_workers.size ());

Unlike get_task_or_become_available (Ch 10) it does not check the queue first — the stopping thread wants no more work, only to keep the pop/push accounting the partition invariant depends on balanced.

INVARIANT — m_has_thread false→true is a one-shot spawn gate. Under m_task_mutex, exactly one path flips it false→true and immediately start_threads; the reverse flip happens only in get_new_task when a parked thread times out empty. So a worker never has more than one live thread, and a task assigned while a thread exists is always a wake. Lose the mutex discipline and two threads could drain one m_wrapped_task.

9.9 Chapter summary — key takeaways

1. Three tiers, two selections. worker_pool_impl owns cores, core_impl owns workers, worker_impl owns one thread+context; dispatch is pick-a-core (get_next_core) then pick-a-worker-or-queue (execute_task).
2. The partition is static; only tasks move. assign_workers_to_cores fixes each worker to one core for life (first remainder cores get one extra); the per-core counts sum to m_max_workers.
3. Round-robin is proportional, not uniform. The cursor wraps at m_max_workers, folded by index % m_cores.size(), so each core’s task share equals its worker share — the smaller core is skipped once a cycle.
4. execute_task has four destinations, one lock. Reject, fast (idle worker, unlock-then-assign), temp (nested SP/method on a throwaway thread), queue — and pop-worker vs push-task is atomic under m_workers_mutex, closing the lost-wakeup window.
5. Handoff is wake-or-spawn on m_has_thread. assign_task sets the worker’s optional, then notify_ones a parked thread or start_threads a new one; prestart via the taskless overload keeps spawn latency off the hot path.
6. Two disjoint mutex scopes. m_workers_mutex guards per-core lists, m_task_mutex guards one worker’s handoff; execute_task unlocks the core lock before entering a worker, so they never nest.
7. Stats is compile-time free. wrapped_task and m_stats collapse to the no-timing layout under worker_pool<false> via std::conditional_t and an empty stats_base; Ch 11 covers the live path.

Chapter 10: Back End Pool Engine II the Worker Run Loop

Chapter 9 traced the push side. This chapter traces the pull side from the worker thread’s own point of view: the function a freshly spawned std::thread runs, worker_impl::run, and everything it reaches — the context bracket (init_run / finish_run), the condition-variable wait that decides “run another task” vs “retire this thread”, the temp-worker single-shot path, pool teardown, and the worker_pool_task_capper admission gate.

All structs were introduced in Chapter 9; the story is motion of the same worker_impl fields across states, so Section 10.1 opens with a matrix.

Cross-link: why the pool retires idle threads and re-spawns on demand is argued in cubrid-thread-worker-pool.md section “cubthread::worker_pool — partitioned pool with per-core queues”; temp workers serve the stored-procedure recursion case in cubrid-thread-manager.md section “Task worker pool sizing”. This chapter is the line-by-line HOW.

10.1 The worker lifecycle and the fields that drive it

A worker_impl is a reusable slot, not a thread; over its life it may own zero, one, or many OS threads in succession. Four fields encode its cycle.

Field	INACTIVE (no thread)	RUNNING a task	WAITING for task	RETIRING
m_has_thread	false	true	true	flips true→false in get_new_task
m_wrapped_task	nullopt	holds current task	nullopt until assigned	nullopt
m_context_p	nullptr	valid entry *	valid entry *	valid→nullptr in finish_run
m_stop	false	maybe true	flips to true on stop	true or timeout
m_is_temp	fixed at construction — a temp slot never enters WAITING; it runs one task then dies

INVARIANT — m_has_thread is the single-thread-per-slot latch. Written only under m_task_mutex, set false before the context is retired so a concurrent assign_task seeing false spawns a fresh thread rather than notifying the dying one; violate it and two threads race one m_context_p.

stateDiagram-v2
    [*] --> Inactive
    Inactive --> Starting : assign_task sets\nm_has_thread true\nstart_thread
    Starting --> InitRun : run entry\ninit_run creates context
    InitRun --> Executing : execute_current_task
    Executing --> GetNewTask : task retired\ncontext recycled
    GetNewTask --> Executing : queued task\nor cv woke with task
    GetNewTask --> Retiring : empty and timeout\nor m_stop
    Retiring --> Inactive : m_has_thread false\nfinish_run retires context
    InitRun --> TempExec : m_is_temp
    TempExec --> TempFinish : execute one task\nfinish_run
    TempFinish --> [*] : register_free_temp_list\nthread exits

Figure 10-1. worker_impl slot lifecycle. Pooled cycles Executing↔GetNewTask; temp takes the single-shot branch.

10.2 worker_impl::run — the thread entry point

run is passed to std::thread in start_thread; two shapes selected by m_is_temp.

// worker_impl::run -- src/thread/thread_worker_pool_impl.hpp
void worker_pool_impl<Stats>::core_impl::worker_impl::run (void)
{
  os::resources::cpu::clearaffinity ();     /* <- drop the spawner's CPU pin */
  pthread_setname_np (pthread_self (), ... get_name ().c_str ());  /* <- name for top/gdb */
  init_run ();                              /* <- create the entry context */
  if (m_is_temp)
    {
      execute_current_task ();              /* <- temp always arrives WITH a task */
      finish_run ();                        /* <- retire context, self-park */
      return;                               /* <- thread dies; no loop, no wait */
    }
  if (!m_wrapped_task.has_value ())
    {
      if (get_new_task ())                  /* <- prestart path: warmup / pool_threads */
        { assert (m_wrapped_task.has_value ()); }
    }
  if (m_wrapped_task.has_value ())
    {
      do { execute_current_task (); }
      while (get_new_task ());              /* <- get_new_task false is the ONLY loop exit */
    }
  // else: never got a task (prestarted, then instant timeout/stop)
}

clearaffinity frees OS scheduling; init_run always runs. The temp branch runs one task and returns (never waits). The pooled path may prestart taskless (warmup / pool_threads), calling get_new_task once; then the loop runs until get_new_task returns false, which has already called finish_run internally (Section 10.5), so run falls off the end and the thread exits.

10.3 init_run / finish_run — the context bracket

Each thread brackets its work with exactly one context create/retire pair.

// worker_impl::init_run -- src/thread/thread_worker_pool_impl.hpp
void ...::worker_impl::init_run (void)
{
  assert (m_has_thread);
  if (!m_is_temp)   /* <- debug: a running pooled slot must be off the available list */
    static_cast<core_impl *> (m_parent_core)->check_worker_not_available (*this);
  // ... stats ...
  m_context_p = &m_parent_core->get_entry_manager ().create_context ();  /* <- claim entry */
}

// worker_impl::finish_run -- src/thread/thread_worker_pool_impl.hpp
void ...::worker_impl::finish_run (void)
{
  assert (!m_wrapped_task.has_value ());   /* <- never retire context with a task pending */
  assert (m_context_p != nullptr);
  m_parent_core->get_entry_manager ().retire_context (*m_context_p);
  m_context_p = nullptr;
  if (m_is_temp)
    static_cast<core_impl *> (m_parent_core)->register_free_temp_list (this);  /* <- self-park */
}

create_context claims a pooled cubthread::entry and runs the manager’s on_create hook (for a transaction worker, assigns a transaction index).

INVARIANT — context lifetime equals thread lifetime. m_context_p is non-null for exactly the span between init_run and finish_run; between tasks it is recycled (Section 10.4), not retired — far cheaper. A temp slot, whose thread is about to exit, cannot delete itself, so it parks itself via register_free_temp_list (Section 10.6) for a later sweep.

10.4 execute_current_task and task retirement

// worker_impl::execute_current_task -- src/thread/thread_worker_pool_impl.hpp
void ...::worker_impl::execute_current_task (void)
{
  assert (m_wrapped_task.has_value ());
  m_wrapped_task->execute (*m_context_p);                 /* <- task->execute(entry&) */
  retire_current_task ();                                 /* <- task->retire(); reset optional */
  m_parent_core->get_entry_manager ().recycle_context (*m_context_p);  /* <- cheap reset */
  if (m_is_temp == false)
    static_cast<core_impl *> (m_parent_core)->free_all_temp_list ();   /* <- reclaim dead temps */
}

retire_current_task calls m_wrapped_task->retire() (default delete this) and resets the optional, satisfying finish_run’s pre-task assert — a task is always retired after execute, even on the temp path. The tail free_all_temp_list makes temp cleanup free of a dedicated reaper: every time a pooled worker finishes, it clears its core’s free-temp graveyard. Temp slots skip it (they are being reclaimed, not reclaiming).

10.5 get_new_task — the wait on the worker condition

The only place a pooled worker blocks, and where a slot decides to retire.

// worker_impl::get_new_task -- src/thread/thread_worker_pool_impl.hpp
bool ...::worker_impl::get_new_task (void)
{
  assert (!m_wrapped_task.has_value ());
  std::unique_lock<std::mutex> ulock (m_task_mutex, std::defer_lock);  /* <- not yet held */
  if (!m_stop)
    {
      std::optional<wrapped_task> queued_task =
          static_cast<core_impl *> (m_parent_core)->get_task_or_become_available (*this);
      if (queued_task.has_value ())
        {
          m_wrapped_task.emplace (std::move (*queued_task));
          return true;                       /* <- fast path: drained the queue */
        }
      ulock.lock ();
      if (!m_wrapped_task.has_value () && !m_stop)
        condvar_wait (m_task_cv, ulock, ... get_idle_timeout (),
                      [this] { return m_wrapped_task.has_value () || m_stop; });
    }
  else
    {
      static_cast<core_impl *> (m_parent_core)->become_available (*this);  /* <- re-advertise */
      ulock.lock ();
    }
  if (!m_wrapped_task.has_value ())
    {
      m_has_thread = false;   /* <- latch: next assign_task must spawn a new thread */
      finish_run ();          /* <- retire context BEFORE releasing the slot */
      return false;
    }
  ulock.unlock ();            /* <- else: a pusher populated m_wrapped_task */
  return true;                /* <- wakeup_with_task */
}

get_task_or_become_available (Ch 9) atomically returns a queued task or, finding the queue empty, pushes this slot onto m_available_workers and returns nullopt — so on the wait path the slot is already advertised free and a pusher may set m_wrapped_task at any moment, which is why the lambda re-checks under m_task_mutex.

stateDiagram-v2
    [*] --> CheckStop
    CheckStop --> GetOrAvail : not m_stop
    CheckStop --> BecomeAvail : m_stop, stopping
    GetOrAvail --> EmplaceReturn : queued task found
    GetOrAvail --> LockMutex : queue empty, now on available list
    LockMutex --> Wait : no task and not stop
    LockMutex --> CheckTask : skip wait
    BecomeAvail --> CheckTask : lock mutex
    Wait --> CheckTask : task, stop, or idle_timeout
    CheckTask --> Retire : m_wrapped_task empty
    CheckTask --> RunTask : m_wrapped_task set
    EmplaceReturn --> [*] : found_in_queue\nreturn true
    Retire --> [*] : m_has_thread false\nfinish_run\nreturn false
    RunTask --> [*] : wakeup_with_task\nreturn true

Figure 10-2. worker_impl::get_new_task. Both false routes (idle-timeout with empty queue, or stop) flow through the retire node; every true route leaves with m_wrapped_task populated.

Branches: (1) not stopping, queue non-empty — fast path, no blocking. (2) not stopping, queue empty — slot now on the available list; take the mutex and if still no task, condvar_wait on m_task_cv bounded by get_idle_timeout() (which collapses to wait for an infinite timeout, wait_for otherwise). (3) m_stop already set — no wait; re-advertise via become_available so teardown accounting sees it, then lock. (4) post-wait, no task — retire: m_has_thread = false, finish_run, return false. (5) post-wait, task present — unlock, return true.

INVARIANT — the retire decision is made under m_task_mutex. The retire-vs-run read of m_wrapped_task and the m_has_thread = false write share the same mutex assign_task takes, serializing “worker decides to die” against “pusher assigns work” so a task never lands on a slot already committed to retiring.

10.6 Temp workers: creation and the free-temp handoff

Temp workers exist for recursive execution (a running task, e.g. a stored procedure, itself needs a worker) where blocking on the normal queue could deadlock. Creation is in execute_task_as_temp (dispatch side, Ch 9).

// core_impl::execute_task_as_temp -- src/thread/thread_worker_pool_impl.hpp
void ...::core_impl::execute_task_as_temp (wrapped_task &&task_ref)
{
  auto w = allocate_worker (true);          /* <- is_temp = true */
  w->set_parent_core (*this);
  std::lock_guard<std::mutex> ulock (m_temp_workers_mutex);
  m_temp_workers.push_back (std::move (w));  /* <- core owns the temp slot */
  static_cast<worker_impl *> (m_temp_workers.back ().get ())->assign_task (std::move (task_ref));
}

// core_impl::register_free_temp_list -- src/thread/thread_worker_pool_impl.hpp
void ...::core_impl::register_free_temp_list (worker *w)
{
  std::unique_lock<std::mutex> ulock (m_temp_workers_mutex);
  auto it = std::find_if (m_temp_workers.begin (), m_temp_workers.end (),
                          [w] (const std::unique_ptr<worker> &p) { return p.get () == w; });
  assert (it != m_temp_workers.end ());
  m_free_temp_workers.push_back (std::move (*it));  /* <- move ownership, active -> free */
  m_temp_workers.erase (it);
}

// core_impl::free_all_temp_list -- src/thread/thread_worker_pool_impl.hpp
void ...::core_impl::free_all_temp_list ()
{
  std::unique_lock<std::mutex> ulock (m_temp_workers_mutex);
  m_free_temp_workers.clear ();   /* <- destroys the parked unique_ptrs */
}

assign_task on a temp slot sets m_has_thread = true and calls start_thread() unconditionally (no pre-existing thread to notify); the fresh thread runs, hits m_is_temp, executes once, and finish_run’s temp tail moves the unique_ptr from m_temp_workers to m_free_temp_workers — the object stays alive because its thread is still inside finish_run. A pooled worker’s free_all_temp_list later clear()s the free list to destroy them.

INVARIANT — a temp slot is destroyed only by a different thread. The self-park plus the pooled-only clear() (guard m_is_temp == false) guarantee no thread ever runs delete this out from under its own stack; both lists are guarded by m_temp_workers_mutex.

10.7 Teardown: stop_execution across pool, core, worker

Shutdown fans out top-down and confirms bottom-up.

// worker_pool_impl::stop_execution -- src/thread/thread_worker_pool_impl.hpp
void worker_pool_impl<Stats>::stop_execution (void)
{
  if (m_stopped.exchange (true))
    return;                 /* <- another caller already owns the stop */
  // ... 30s(release)/60s(debug) timeout, 10ms spin ...
  while (true)
    {
      bool has_running_workers = false;
      for (const auto &it : m_cores)
        has_running_workers |= it->stop_execution ();   /* <- notify every worker */
      if (!has_running_workers)
        break;                                          /* <- everyone quiesced */
      if (now () > timeout)
        { assert (false); break; }                      /* <- gave up; something is wedged */
      std::this_thread::sleep_for (time_spin_sleep);
    }
  for (const auto &it : m_cores)                         /* <- queues now safe to drain */
    static_cast<core_impl *> (it.get ())->retire_queued_tasks ();
}

INVARIANT — exactly one thread performs teardown. m_stopped.exchange(true) returns the prior value, so the first caller sees false and runs the whole spin/retire once; later callers return. is_running() reads !m_stopped, so execute_task (Ch 9) rejects new tasks the instant the flag flips. The loop breaks normally on an all-quiet sweep, abnormally on timeout with assert(false) (compiled out in release). Queues drain only after quiescence.

// worker_impl::stop_execution -- src/thread/thread_worker_pool_impl.hpp
bool ...::worker_impl::stop_execution (void)
{
  context_type *context_p = m_context_p;       /* <- plain read, deliberately unsynchronized */
  bool has_thread = false;
  if (context_p != nullptr)
    m_parent_core->get_entry_manager ().stop_execution (*context_p);  /* <- wake a long task */
  std::unique_lock<std::mutex> ulock (m_task_mutex);
  if (m_has_thread)
    has_thread = true;                          /* <- report still-running to the pool loop */
  m_stop = true;
  ulock.unlock ();
  if (!m_is_temp)
    m_task_cv.notify_one ();                    /* <- kick a waiting pooled thread */
  return has_thread;
}

Branches: (1) a live context gets the manager’s stop_execution hook to interrupt a task blocked deep in execution (e.g. a lock wait). (2) m_has_thread under the lock drives the pool’s retry return. (3) m_stop = true is what get_new_task checks. (4) notify_one is skipped for temp workers — a temp slot never waits on m_task_cv. The m_context_p read is intentionally lock-free; observation racing with retirement is documented and tolerated.

stateDiagram-v2
    [*] --> Exchange
    Exchange --> [*] : was true, someone else owns it
    Exchange --> Sweep : was false
    Sweep --> Notify : each core stop_execution
    Notify --> CheckRunning : set m_stop\nnotify_one if pooled
    CheckRunning --> CheckTimeout : any has_thread
    CheckRunning --> Drain : all quiesced
    CheckTimeout --> Sweep : not exceeded\nsleep 10ms
    CheckTimeout --> Drain : exceeded\nassert false
    Drain --> [*] : retire_queued_tasks per core

Figure 10-3. Teardown fan-out. The pool re-sweeps until every worker reports has_thread == false or the timeout fires; only then are queues drained.

10.8 core_impl::retire_queued_tasks — draining the backlog

After quiescence, tasks queued but never dispatched still hold heap-allocated task objects. They must be retired, not executed.

// core_impl::retire_queued_tasks -- src/thread/thread_worker_pool_impl.hpp
void ...::core_impl::retire_queued_tasks (void)
{
  std::unique_lock<std::mutex> ulock (m_workers_mutex);   /* <- same mutex the queue uses */
  while (!m_task_queue.empty ())
    {
      wrapped_task queued_task = std::move (m_task_queue.front ());
      m_task_queue.pop ();
      queued_task.retire ();   /* <- task->retire(); frees the task, no execute */
    }
}

Unconditional drain-to-empty. Holding m_workers_mutex (the push-side lock) plus m_stopped == true means no execute_task can race in a new task. The moved-from wrapped_task destructor asserts its inner pointer is null, catching a double-free at the boundary.

10.9 The admission cap — worker_pool_task_capper

The pool never blocks a pusher: execute_task queues without bound. worker_pool_task_capper is the opt-in front door that bounds in-flight work to the worker count, applying back-pressure via capped_task decorators.

// worker_pool_task_capper -- src/thread/thread_worker_pool_taskcap.cpp
worker_pool_task_capper::worker_pool_task_capper (worker_pool *worker_pool) /* ... */
{
  m_tasks_available = m_max_tasks = worker_pool->get_worker_count ();  /* <- cap = pool size */
}
bool worker_pool_task_capper::try_task (task<entry> *task)
{
  std::unique_lock<std::mutex> ulock (m_mutex);
  if (m_tasks_available == 0)
    return false;              /* <- non-blocking: caller keeps the task */
  execute (task);  return true;
}
void worker_pool_task_capper::push_task (task<entry> *task)
{
  std::unique_lock<std::mutex> ulock (m_mutex);
  m_cond_var.wait (ulock, [&] { return (m_tasks_available > 0); });  /* <- block for a permit */
  execute (task);
}
void worker_pool_task_capper::execute (task<entry> *task)   /* <- private, under m_mutex */
{
  m_tasks_available--;                                       /* <- consume a permit */
  m_worker_pool->execute (new capped_task (*this, task));    /* <- wrap and dispatch */
}
void worker_pool_task_capper::end_task ()
{
  std::unique_lock<std::mutex> ulock (m_mutex);
  m_tasks_available++;                                       /* <- return the permit */
  ulock.unlock ();
  m_cond_var.notify_all ();                                  /* <- wake all push_task waiters */
}
void worker_pool_task_capper::capped_task::execute (entry &ctx)
{
  m_nested_task->execute (ctx);   /* <- real work, on a real worker context */
  m_capper.end_task ();           /* <- permit returned exactly once, post-execute */
}
worker_pool_task_capper::capped_task::~capped_task ()
{ m_nested_task->retire (); }     /* <- free the wrapped task when the wrapper is retired */

try_task and push_task differ only in the empty-cap branch: try_task returns false and leaves ownership with the caller; push_task blocks on m_cond_var until end_task frees a permit. Both call the private execute.

INVARIANT — permit conservation bounds concurrency at m_max_tasks. m_tasks_available starts at worker_count, decrements at admission and increments at completion (both under m_mutex), staying in [0, m_max_tasks], so at most m_max_tasks capped_tasks are ever in flight and the pool’s queue never grows; the permit is returned after the nested task runs but before the wrapper is retired.

One subtlety: the cap is fixed from get_worker_count() at construction and is not re-read. For the Phase-2 elastic pool whose width changes, see cubrid-thread-manager.md section “Elastic worker pool”.

10.10 Chapter summary — key takeaways

1. A worker_impl is a reusable slot, not a thread. run is the thread body; m_has_thread (written only under m_task_mutex) is the latch that tells a pusher whether to notify an existing thread or spawn a new one.
2. init_run / finish_run bracket exactly one context per thread lifetime, while recycle_context cheaply resets it between tasks; the pooled loop is do execute_current_task while get_new_task.
3. get_new_task is the single blocking point and the retire decision. It returns a queued task, waits on m_task_cv bounded by the idle timeout, or — finding nothing — sets m_has_thread = false, finish_run, returns false.
4. Temp workers are single-shot and self-parking. Created by execute_task_as_temp, they run one task and register on m_free_temp_workers; a pooled worker later reclaims them via free_all_temp_list, so no thread ever delete this-es itself.
5. Teardown is idempotent and confirmed. m_stopped.exchange(true) elects one stopper; the pool re-sweeps cores until all report has_thread == false or a timeout asserts, then retire_queued_tasks drains the backlog.
6. worker_pool_task_capper adds the back-pressure the pool lacks. A permit counter bounded at worker_count gates admission; try_task fails fast, push_task blocks, and capped_task::execute returns the permit right after the real work completes.
7. Every branch retires a task or a context exactly once. Task retirement is after execute, in retire_queued_tasks, and in ~capped_task; context retirement is once per thread in finish_run. The asserts on m_wrapped_task and m_context_p are the guard rails.

Chapter 11: Atomic Free Statistics and Observability

The connection reactor never touches a perfmon atomic while moving bytes. Load is instead recorded in per-owner plain-integer arrays, snapshotted by copy onto a message at a slow timer cadence, and folded into an EWMA model by the single-threaded coordinator. This chapter traces that pipeline field by field: the cubconn::statistics value types, the coordinator::worker_statistics rows, the push (statistics_metrics_to_coordinator) with its consumer branch, and the two operator surfaces — the statistics_print overlay and the SQL SHOW JOB QUEUES view — ending at the one place the pipeline is a stub: css_get_task_stats.

Cross-link: Why the redesign abandons perfmon atomics on the reactor hot path — false sharing, cache-line contention, C10K fan-out — is argued in cubrid-thread-manager.md §“Observability without contention” (CBRD-26191). This chapter is the how. The back-end request pool it contrasts against is cubrid-thread-worker-pool.md §“Worker pool statistics”.

11.1 The value types — cubconn::statistics

connection_statistics.hpp defines two enum class key sets and one fixed-size counter array template metrics<T, VT>. No map, no hashing, no lock — a stat “write” is array[static_cast<size_t>(key)] += value.

// cubconn::statistics::worker -- src/connection/connection_statistics.hpp
enum class worker : std::uint8_t
{
  PACKET_COUNT, CLIENT_NUM,                       /* network */
  MQ_REQUESTED,                                   /* message queue */
  MQ_COMPLETED, MQ_NEW_CLIENT, MQ_HANDOFF_CLIENT,
  MQ_TAKEOVER_CLIENT, MQ_SHUTDOWN_CLIENT,
  MQ_SEND_PACKET, MQ_RELEASE_PACKET,
  BLOCKED_RMUTEX,                                 /* us */
  STATS_COUNT,                                    /* array size */
  NA                                              /* "no stat" sentinel */
};

context keys (per-connection): BYTES_IN_TOTAL, BYTES_OUT_TOTAL (bytes); OPEND_NS, LAST_ACTIVE_NS, LAST_MOVED_NS (ns wall-clock marks); MOVE_COUNT; RECV_BUDGET_HIT, SEND_BUDGET_HIT (Ch 5 IO-bound flags); STATS_COUNT sizes the array.

worker keys (per-worker-thread, shown above): monotone counters except BLOCKED_RMUTEX (microseconds blocked on css_conn_entry::rmutex) and CLIENT_NUM (a gauge — add on adopt, sub on close). MQ_* mirror the message-queue verbs of Ch 8.

The array template:

// metrics<T,VT> -- src/connection/connection_statistics.hpp
template <class T, typename VT = std::uint64_t>
class metrics {
    // ... condensed: copy ctor, move ctor, cross-VT converting ctor ...
    inline void add (T key, VT value);  /* array[key] += value  — plain, non-atomic */
    inline void sub (T key, VT value);
    inline VT   get (T key);
  private:
    VT m_values[static_cast<std::size_t> (T::STATS_COUNT)];  /* the whole state */
};

Field	Role	Why it exists
m_values[T::STATS_COUNT]	The complete counter state, one slot per enumerator	Fixed array → O(1) add, no allocation, one cache-friendly blob to memcpy

The template’s VT parameter is the numeric type: workers use the default std::uint64_t (raw counters); the coordinator instantiates metrics<T, double> for EWMA accumulators. The converting constructor and operator-/operator* (both returning metrics<T,double>) exist only to lift raw uint64 samples into the double EWMA math of §11.4.

INVARIANT — single-writer, therefore atomic-free. Each metrics instance has exactly one writer thread. A context::m_stats is touched only by the worker that currently owns that context (its receiver and transmitter hold a raw metrics<context>* back into it — receiver::m_stats, transmitter::m_stats); a worker’s own m_stats is touched only by that worker’s reactor loop. No two threads write the same array, so add/sub are plain += — no std::atomic, no fence, no lock. Share one metrics across threads and you silently reintroduce the lost-update / false-sharing cost the redesign removed.

INVARIANT — STATS_COUNT sizes, NA never indexes. The array is sized by the STATS_COUNT sentinel, which must remain the last allocated enumerator. worker adds one more name, NA, past STATS_COUNT; it is out of array bounds by construction and is used only as a “this message type has no counter” marker in the Ch 8 dispatch table ({ handler, statistics::worker::NA }). Passing NA to add writes one slot past the array. The dispatch loop guards it: if (handler[...].second != statistics::worker::NA) m_stats.add (...).

11.2 The hot-path writes — where counters actually move

Every increment sits inline in a path already doing work, so the cost is one add against an L1-resident array:

// worker + context hot-path adds -- connection_worker.cpp / receiver.cpp / transmitter.cpp
m_stats.add  (statistics::worker::PACKET_COUNT, ...->size ());  /* worker: after a drain */
m_stats.add  (statistics::worker::BLOCKED_RMUTEX, us);          /* worker: rmutex block time */
m_stats->add (statistics::context::BYTES_IN_TOTAL, bytes);      /* context: per recv */
m_stats->add (statistics::context::RECV_BUDGET_HIT, 1);         /* context: BudgetExhausted, Ch 5 */
m_stats->add (statistics::context::BYTES_OUT_TOTAL, bytes);     /* context: per send */
m_stats->add (statistics::context::SEND_BUDGET_HIT, 1);         /* context: send budget hit */

Context lifecycle marks are set, not add: on adopt, ctx->m_stats.set (OPEND_NS/LAST_ACTIVE_NS, m_timens); each serviced packet re-stamps LAST_ACTIVE_NS; hand-off bumps MOVE_COUNT and moves CLIENT_NUM (sub on the loser, add on the winner). None is synchronized.

11.3 The coordinator’s rows — coordinator::worker_statistics

The coordinator keeps std::vector<worker_statistics> m_statistics, one row per worker slot. A row is the model built from raw snapshots — it never shares memory with a worker.

// coordinator::worker_statistics -- src/connection/coordinator.hpp
struct worker_statistics {
  double   m_score;                                            /* placement scalar */
  double   m_core;                                             /* CPU utilization 0..1 */
  uint64_t m_last_cpu_time;
  uint32_t m_client_num;
  uint64_t m_last_updated;
  statistics::metrics<statistics::context, double> m_sum;      /* rollup of all contexts */
  std::pair<statistics::metrics<statistics::worker, double>,
            statistics::metrics<statistics::worker>> m_worker;  /* {accumulated EWMA, previous raw} */
  std::unordered_map<uint64_t,
    std::pair<statistics::metrics<statistics::context, double>,
              statistics::metrics<statistics::context>>> m_contexts;  /* per-connection {EWMA, prev} */
};

Field	Role	Why it exists
m_score	Single scalar the placement/rebalance/scale logic ranks workers by	Ch 6 needs one comparable number, not a vector
m_core	Fraction of a core the worker burned since last sample	(cpu_time_ns delta) / (wall delta); the CPU term
m_last_cpu_time	Previous CLOCK_THREAD_CPUTIME_ID reading	Baseline for the m_core delta
m_client_num	Live connection count on this worker	Immediate gauge, copied straight from the sample’s CLIENT_NUM
m_last_updated	Wall-clock ns of the last accepted sample	Doubles as the “have I seen this worker yet?” flag (§11.4) and the EWMA delta base
m_sum	Sum of every live context’s accumulated-EWMA metrics	Gives worker-level byte/budget totals without re-walking contexts on read
m_worker.first	Accumulated EWMA of the worker counters (double)	Smoothed load signal
m_worker.second	Previous raw worker sample (uint64)	The subtrahend for the next delta
m_contexts[id].first	Accumulated EWMA of one connection’s context counters	Rebalance picks a single connection to move (Ch 6)
m_contexts[id].second	Previous raw context sample	Per-connection delta base

INVARIANT — accumulator/previous pairing. For m_worker and each m_contexts entry, .first is a smoothed double and .second is the last raw uint64 sample. EWMA advances by diff = current - previous, so .second must be re-set to current on every accepted sample or the next delta double-counts. A .first present without a matching .second computes its delta against zero and spikes once. The consumer keeps the two in lock-step (§11.4).

Figure 11-1 shows the ownership crossing — the only place worker-owned memory becomes coordinator-owned memory is the message.

flowchart LR
  subgraph W["worker thread — single writer"]
    ws["worker m_stats\nmetrics<worker>"]
    cs["context m_stats\nmetrics<context> per conn"]
  end
  subgraph M["coordinator::message — moved"]
    mw["statistics.worker\n{index, metrics copy}"]
    mc["statistics.contexts\nvector of {id, metrics copy}"]
  end
  subgraph C["coordinator thread — single reader"]
    row["m_statistics[index]\nworker_statistics row"]
  end
  ws -->|copy| mw
  cs -->|copy| mc
  mw -->|std::move enqueue| row
  mc -->|std::move enqueue| row

Figure 11-1 — the metric data path. Copies cross the thread boundary; no shared mutable state.

11.4 The push — worker::statistics_metrics_to_coordinator

A MEDIUM_LATENCY (5 s) timer registered at worker start fires this; its whole job is snapshot-by-value, enqueue-by-move.

// worker::statistics_metrics_to_coordinator -- src/connection/connection_worker.cpp
bool worker::statistics_metrics_to_coordinator () {
  coordinator::message message;
  message.type = coordinator::message_type::STATISTICS;
  message.statistics.cpu_time_ns = get_time_ns (CLOCK_THREAD_CPUTIME_ID);   /* CPU burned */
  message.statistics.time_ns     = get_time_ns (CLOCK_MONOTONIC);           /* wall clock */
  message.statistics.worker.first  = m_index;                              /* which slot */
  message.statistics.worker.second = m_stats;                              /* copy of counters */
  message.statistics.contexts.reserve (m_context.size ());
  for (context *ctx : m_context)                                           /* only LIVE contexts */
    message.statistics.contexts.emplace_back (ctx->m_id, ctx->m_stats);    /* copy each */
  m_coordinator->enqueue (std::move (message));                            /* MPSC, non-blocking */
  return true;
}

The function has no error branch — it always returns true. The branch that matters is loop cardinality: m_context holds only currently-owned contexts, so a hibernating or drained worker copies an empty contexts vector, and that consequence lands in the consumer. enqueue pushes onto the coordinator’s tbb::concurrent_queue (multi-producer) and bumps m_queue_size; it never blocks, so a slow coordinator cannot stall a reactor.

The consumer, handle_message_queue_statistics, is where every branch lives:

// coordinator::handle_message_queue_statistics -- src/connection/coordinator.cpp
index = item.statistics.worker.first;
delta = item.statistics.time_ns - m_statistics[index].m_last_updated;
m_statistics[index].m_core =
    static_cast<double> (item.statistics.cpu_time_ns - m_statistics[index].m_last_cpu_time) / delta;
this->statistics_update_connection (delta, item.statistics.worker, item.statistics.contexts);
m_statistics[index].m_last_cpu_time = item.statistics.cpu_time_ns;
m_statistics[index].m_last_updated  = item.statistics.time_ns;
this->statistics_update_score (index);
if ((m_status == status::DRAINING && (int) index == m_scaling.draining_worker)
    && item.statistics.contexts.empty ())
  this->scale_down_finish ();

Branch map:

1. First sample (m_last_updated == 0, tested inside statistics_update_connection): the else arm seeds .first and .second from the raw sample, no EWMA; m_core’s delta is against a zero baseline this once and discarded next tick.
2. Subsequent sample: m_sum.reset(), then statistics_EWMA advances m_worker and each m_contexts[id] (each accumulator asserted present, keeping .first/.second paired).
3. Context rollup: m_sum is re-summed from every m_contexts[id].first — a fresh reduction each tick, so a departed connection drops out naturally.
4. DRAINING + empty contexts: the drained worker (Ch 6) reported zero contexts → scale_down_finish() retires the slot — the only place an empty-context push is load-bearing.
5. Otherwise: fall through, model updated, no scaling action.

flowchart TD
  A["timer 5s fires\nstatistics_metrics_to_coordinator"] --> B["snapshot cpu_time, wall time,\ncopy m_stats + each ctx m_stats"]
  B --> C["enqueue std::move, bump m_queue_size"]
  C --> D["coordinator drains queue\nhandle_message_queue_statistics"]
  D --> E{"m_last_updated == 0 ?"}
  E -->|yes| F["seed .first and .second\nno EWMA"]
  E -->|no| G["reset m_sum, EWMA worker + contexts"]
  F --> H["re-sum m_sum over contexts"]
  G --> H
  H --> I["statistics_update_score index"]
  I --> J{"DRAINING and draining_worker\nand contexts empty ?"}
  J -->|yes| K["scale_down_finish"]
  J -->|no| L["done"]

Figure 11-2 — push and its branch-complete consumer.

The smoothing kernel converts a raw counter delta into a per-time-unit rate, then blends it:

// coordinator::statistics_EWMA -- src/connection/coordinator.cpp
diff = (current > prev) ? static_cast<double> (current - prev) : 0;   /* guard counter reset */
acc  = acc * (1 - alpha) + diff * (alpha / (time_delta * 1e-6));      /* alpha = 0.06 */
prev = current;                                                       /* keep .second paired */

statistics_update_score then blends the client count, worker EWMAs (MQ_COMPLETED, BLOCKED_RMUTEX) and context EWMAs (bytes, budget hits) via the EVAL_WORKER/EVAL_CONTEXT macros into m_score — the single number Ch 6 ranks by.

11.5 The developer overlay — coordinator::statistics_print

Reached only through the controller debug socket (control_type::SHOW_STATS in handle_controller_request), this paints a full-screen terminal dashboard. It reads the already-built model with no locking — it runs on the coordinator thread itself.

// coordinator::statistics_print -- src/connection/coordinator.cpp
printf ("\033[2J\033[H");                                   /* clear + home */
for (i = 0; i < m_max_worker; i++) {
  if (m_statistics[i].m_contexts.size () == 0) continue;     /* skip idle/unallocated slots */
  printf ("------ worker %d (%d) ------\n", (int) i, (int) m_statistics[i].m_contexts.size ());
  core += m_statistics[i].m_core;
  printf ("SCORE: %lf\n", m_statistics[i].m_score);
  // ... condensed: per-worker detail block is #if-commented out ...
  mq_completed += m_statistics[i].m_worker.first.get (statistics::worker::MQ_COMPLETED);
  bytes_in     += m_statistics[i].m_sum.get (statistics::context::BYTES_IN_TOTAL);
  bytes_out    += m_statistics[i].m_sum.get (statistics::context::BYTES_OUT_TOTAL);
}

Branches: the per-worker loop skips any zero-context slot (continue) — idle, hibernating, or never-allocated workers neither print nor contribute to core/bytes_in/bytes_out. The summary block is unconditional. Two derived values have divide hazards worth knowing before editing: core / m_max_worker (safe, m_max_worker >= 1) and m_task_statistics.depth.first / m_task_statistics.workers * 100, whose denominator is PRM_ID_TASK_WORKER and is assumed non-zero. The status line is a nested ternary over STABLE / DRAINING / EXPANDING. This surface is a developer tool, not an operator API — ANSI-escaped stdout driven by a raw control socket. Production observability is §11.6.

11.6 The operator surfaces — SHOW JOB QUEUES and the task stub

The SQL-visible view is SHOW JOB QUEUES, columns declared in metadata_of_job_queues:

Column	Source	Meaning today
Jobq_index	core index	Back-end request-pool core, reinterpreted (job queues no longer exist)
Num_total_workers	wp_core.get_worker_count()	Workers in that core
Num_busy_workers	counted live via mapper	Workers with a non-null tran_index
Num_connection_workers	hard-coded 0	Reactor workers live in a separate pool

// css_wp_core_job_scan_mapper -- src/connection/server_support.c
(void) db_make_int (&vals[val_index++], (int) core_index);
(void) db_make_int (&vals[val_index++], (int) wp_core.get_worker_count ());
int busy_count = 0;
wp_core.map_running_contexts (stop_mapper, css_wp_worker_get_busy_count_mapper, busy_count);
(void) db_make_int (&vals[val_index++], (int) busy_count);
(void) db_make_int (&vals[val_index++], 0);   /* connection workers reported here as 0 */

Note the discipline: SHOW JOB QUEUES reports the back-end request pool (Ch 9–10), not the connection reactor — whose load model (§11.3) has no SQL surface yet. And even this legacy view avoids a hot-path counter: css_job_queues_start_scan derives busyness from tran_index != NULL_TRAN_INDEX by mapping live contexts at scan time, not by incrementing per request.

The task-statistics feed is currently a stub. Every LOW_LATENCY tick, statistics_update_task calls css_get_task_stats(stats) and folds the result into m_task_statistics (requested/started/completed/depth), feeding both the statistics_print summary and the auto-scale score (m_task_statistics.completed.first * 2 in statistics_scaling). But:

// css_get_task_stats -- src/connection/server_support.c
void css_get_task_stats (UINT64 *stats_out) {
//  css_Server_request_worker_pool->get_task_stats (stats_out);   /* <- commented out */
}

The real call is commented out, so stats_out is left untouched and statistics_update_task reads whatever was on the stack: the first tick seeds .second from garbage/zero and later ticks EWMA against it. So the requested/started/completed/depth overlay lines and the task term of the scaling score are not meaningful until this is wired — the byte/mq signals (§11.4) are live, the task signal is dead code.

For contrast, the live perfmon path survives for the back-end pool: css_get_thread_stats → worker_pool_impl::get_stats → core_impl::get_stats, which takes m_workers_mutex and accumulates each worker’s cubperf::stat_value array. That lock-and-perfmon pattern is exactly what §11.1 avoids on the reactor — acceptable here only because these stats are read for er_log_stats, not per packet.

11.7 Chapter summary — key takeaways

1. Atomic-free by single-writer ownership. metrics<T,VT> is a plain fixed array written by exactly one thread (a context’s owning worker, or a worker’s own loop), so add/sub are unsynchronized += — the whole “no perfmon on the hot path” discipline (§11.1).
2. The thread boundary is crossed only by copy-then-move. statistics_metrics_to_coordinator copies the counter arrays onto a move-only message at a 5 s cadence and enqueues it non-blocking; the coordinator is the single reader (Figure 11-1, §11.4).
3. The coordinator keeps a model, not the counters. worker_statistics pairs a double EWMA (.first) with the previous raw sample (.second) for the worker and each context, plus the m_sum rollup and derived m_score (§11.3).
4. Every consumer branch matters. First-sample seeding vs. EWMA advance, per-context rollup, and the DRAINING+empty-contexts arm that triggers scale_down_finish are the branches of handle_message_queue_statistics (§11.4, Figure 11-2).
5. Two operator surfaces, different pools. statistics_print is a coordinator-thread ANSI dashboard over the reactor model (skips zero-context slots); SHOW JOB QUEUES reports the back-end pool’s cores, deriving busyness by on-demand context mapping and reporting connection workers as 0 (§11.5–§11.6).
6. css_get_task_stats is a stub. Its real call is commented out, so m_task_statistics and the task term of the auto-scale score are inert today — wire it before trusting the task signal (§11.6).
7. The old perfmon path survives where it is cheap. worker_pool_impl::get_stats still locks and reads cubperf values for the back-end pool; the redesign removed atomics only from the per-packet reactor path, not everywhere (§11.6).

Chapter 12: Phase 2 Data Structures Elastic Pool and Concurrency Slots

Source scope: every symbol in this chapter comes from PR #7323 (feature/worker_pool_elastic), which is NOT yet merged to develop. Line numbers in the trailing POSHINTS block are relative to that PR worktree and will not match develop.

Phase 1 (Ch 9–11) bounds concurrency by the number of worker threads. That coupling breaks when a running task blocks on a logical wait (a lock, a critical section, an SP round-trip): the thread is alive but idle, yet still counts against concurrency. Phase 2 introduces a second currency — the concurrency slot. Runnable concurrency is now = number of held slots, and thread count may overcommit past the slot count so a thread parked on a logical wait can surrender its slot to a fresh thread.

This chapter is structural only — every field of every new object. The flows (acquire, release, slot daemon, logical-wait wiring) are Ch 13–14, and the why lives in the companion cubrid-thread-manager.md §“Elastic worker pool (CBRD-26684)”, §“Concurrency slots and the per-core slot pool (CBRD-26685 / 26689)”, and §“The slot daemon — soft/hard wait and inter-core transfer (CBRD-26686 / 26691)”. Two families appear: the templated pool family (worker_pool_elastic<Stats>, core_elastic, worker_elastic) — the Phase-1 engine re-skinned to carry a slot per task; and the non-templated slot family (concurrency_slot, concurrency_slot_pool, the publisher/subscriber pair, concurrency_slot_daemon) in concurrency_slot.hpp/.cpp, which knows nothing about Stats and bridges back through type-erased void *.

12.1 worker_pool_elastic<Stats> — the slot-aware pool

// worker_pool_elastic -- src/thread/thread_worker_pool_elastic.hpp
template <stats_t Stats>
class worker_pool_elastic final : public worker_pool_impl<Stats>
{
    friend class manager;
  public:
    using unique_slot = std::unique_ptr<concurrency_slot>;   /* <- ownership token */
    // ... condensed: worker, task_type, wrapped_task, stats aliases ...
    class core_elastic;
  private:
    std::atomic<std::size_t> m_current_worker;   /* placed in same cache line */
    std::atomic<std::size_t> m_max_concurrency;
    std::atomic<std::size_t> m_max_worker;
};

final, derived from Phase-1 worker_pool_impl<Stats> (Ch 9). The non-type parameter stats_t Stats (on/off) selects the atomic-free stats policy (Ch 11) at compile time — the slot machinery is duplicated for both, which is why the .cpp dynamic_casts always try both <stats_t::on> and <stats_t::off>. The alias unique_slot = std::unique_ptr<concurrency_slot> is the ownership token threaded through every hand-off in Ch 13.

Field	Role	Why it exists
m_current_worker	Atomic count of live worker threads across all cores	Global overcommit budget; reserve_available_worker CAS-increments it before spawning (Ch 13). Its address is shared by reference into every core_elastic
m_max_concurrency	Atomic target = total runnable slots = total slots minted	The concurrency ceiling; split across cores by adjust_runtime_parameter
m_max_worker	Atomic hard cap on total live threads	Lets threads overcommit past m_max_concurrency so a slot-holder that blocks can hand its slot to another thread

INVARIANT — m_max_worker >= m_max_concurrency. Enforced by assert (max_worker >= max_concurrency) in adjust_runtime_parameter. Slots cap runnable work; workers may exceed slots so a logically waiting worker’s slot is never idle. Violate it and a blocked worker’s slot cannot be reused, collapsing elastic behaviour back to thread-bounded concurrency.

INVARIANT — m_current_worker seeds to max_concurrency, not 0. The constructor initializes m_current_worker (max_concurrency) while initialize builds exactly m_max_concurrency workers, so at bring-up live-threads == slots and the atomic already reflects that census. Starting at 0 would mis-budget the first overcommits.

12.2 core_elastic — the per-core slot owner

// core_elastic -- src/thread/thread_worker_pool_elastic.hpp
template <stats_t Stats>
class worker_pool_elastic<Stats>::core_elastic final
  : public worker_pool_impl<Stats>::core_impl
{
    friend class worker_pool_elastic;
    class worker_elastic;
    // ... condensed: adjust_workers, execute_task, get_task_and_slot_or_become_available ...
    concurrency_slot_pool m_slots;
    std::size_t m_max_concurrency;                 // per core
    std::atomic<std::size_t> &m_current_worker;    // global, not hard cap
    std::atomic<std::size_t> &m_max_worker;
    stats_base m_retired_stats;
};

Each core embeds a concurrency_slot_pool (m_slots) built as m_slots (this, this->m_core_mutex) — the second argument means the slot pool does not own a mutex, it borrows the core’s m_core_mutex, so slot state and worker state share one lock.

Field	Role	Why it exists
m_slots	The per-core concurrency_slot_pool	Gates how many of this core’s tasks may run at once
m_max_concurrency	This core’s slice of the pool-wide max_concurrency (its target slot count)	adjust_runtime_parameter distributes the global target across cores (quotient + remainder)
m_current_worker	Reference to the pool-global live-thread atomic	Overcommit is bounded globally, not per-core, so one busy core can borrow the pool’s whole thread budget
m_max_worker	Reference to the pool-global thread cap atomic	Read in reserve_available_worker’s CAS loop
m_retired_stats	Accumulated stats_base of workers removed during downscale	Stats must survive worker deletion; get_retire_if_excess folds a dying worker’s counters here before erasing it

INVARIANT — the two atomics are held by reference, not by value. m_current_worker and m_max_worker alias the pool’s members (passed in allocate_core). A per-core copy would let each core spawn max_worker threads independently, blowing the global cap by core_countx.

12.3 worker_elastic — a worker that carries a slot

// worker_elastic -- src/thread/thread_worker_pool_elastic.hpp
template <stats_t Stats>
class worker_pool_elastic<Stats>::core_elastic::worker_elastic final
  : public worker_pool_impl<Stats>::core_impl::worker_impl
{
    friend class core_elastic;
    // ... condensed: assign_task, run, execute_current_task, get_new_task ...
    unique_slot m_slot;   // guarded by m_task_mutex
};

worker_elastic adds exactly one field to Phase-1 worker_impl: the slot bound to its current task. assign_task moves a wrapped_task and a unique_slot into the worker under m_task_mutex; execute_current_task moves m_slot onto the thread entry (m_context_p->m_slot) just before running.

Field	Role	Why it exists
m_slot	The concurrency_slot paired with the assigned task; guarded by m_task_mutex	A worker only runs when it owns a slot; keeping the slot on the worker (not the queue) means the slot travels with the task through wake-up

INVARIANT — task and slot are co-present or co-absent. Every wait predicate and post-condition in get_new_task asserts (!m_wrapped_task.has_value () && !m_slot) || (m_wrapped_task.has_value () && m_slot). A worker may never hold a task without a slot (it would run unbounded) or a slot without a task (it would leak concurrency). This is the single most load-bearing invariant of Phase 2 — it is what makes “concurrency == held slots” literally true.

12.4 concurrency_slot — the concurrency token

The slot is a tiny object; its value is in who holds it and for how long.

// concurrency_slot -- src/thread/concurrency_slot.hpp
class concurrency_slot
{
    friend class concurrency_slot_pool;
    // ... condensed: reset, return_to_pool, start_waiting, has_wait_expired ...
    concurrency_slot_pool *const m_owner_pool;   /* fixed at construction */
    concurrency_slot_pool *m_holder_pool;
    bool m_wait;                                 // soft wait
    std::chrono::time_point<std::chrono::steady_clock> m_wait_since;
    // guarded by the holder pool mutex (core mutex)
};

Field	Role	Why it exists
m_owner_pool	const — the pool that minted the slot	A borrowed slot must ultimately return to its owner; const makes the home address immutable for the slot’s life
m_holder_pool	The pool currently lending the slot out	Differs from owner after the daemon transfers a slot between cores; set_holder_pool stamps it on acquire
m_wait	Soft-wait flag; set by start_waiting when the holder enters a logical wait	Marks the slot as stealable by the daemon
m_wait_since	steady_clock timestamp the wait began	has_wait_expired compares against a 50 ms threshold to decide a hard steal

INVARIANT — owner is immutable, holder is mobile. With m_owner_pool const and m_holder_pool mutable, a slot lent across cores always knows its way home. return_to_pool offers the slot to m_owner_pool->return_slot(...), then m_holder_pool, then forces it back to the owner — a three-step drain that keeps each pool’s m_slot_count honest under transfer. Lose the split and cross-core borrowing leaks slots at teardown.

12.5 concurrency_slot_pool — the per-core slot bank

// concurrency_slot_pool -- src/thread/concurrency_slot.hpp
class concurrency_slot_pool : public concurrency_slot_subscriber
{
    // ... condensed: try_acquire_slot, acquire_slot, release_slot, borrow_surplus_slots, get_score ...
    static constexpr std::size_t SLOT_SURPLUS_THRESHOLD = 2;   // must be > 1
    const void *m_parent;
    std::queue<std::unique_ptr<concurrency_slot>> m_available_slots;
    bool m_surplus;
    std::chrono::time_point<std::chrono::steady_clock> m_surplus_since;
    std::size_t m_slot_count;
    std::size_t m_target_count;
    std::list<entry *> m_wait_queue;   // entries waiting to acquire a slot
    std::mutex *m_mutex;               // the owning core mutex
};

The pool banks free slots in a FIFO queue and parks slot-starved threads in a wait list. It is also a concurrency_slot_subscriber (§12.6): its constructor passes concurrency_slot_daemon::get_publisher () to the subscriber base, so every core’s pool auto-registers with the one global daemon.

Field	Role	Why it exists
SLOT_SURPLUS_THRESHOLD	static constexpr = 2	Min free slots before a pool is “surplus”; > 1 guarantees a pool always keeps one slot for itself when the daemon borrows
m_parent	const void * back-pointer to the owning core_elastic	Type-erased so this non-templated file needn’t see Stats; cast back via dynamic_cast to core_elastic<on/off> in wakeup_workers/has_queued_task
m_available_slots	FIFO queue of free unique_slot	The bank; try_acquire_slot pops the front, release_slot pushes the back
m_surplus	True while available >= SLOT_SURPLUS_THRESHOLD	Cheap flag the daemon reads before considering a borrow
m_surplus_since	When the surplus condition began	borrow_surplus_slots only steals after surplus has persisted > 2 s, damping churn
m_slot_count	Slots physically in this pool right now (owned + borrowed)	Downscale (adjust_concurrency) destroys owned slots until it reaches target
m_target_count	Desired slot count = the core’s m_max_concurrency	Divergence from m_slot_count drives create/destroy during reconfiguration
m_wait_queue	list<entry *> of thread entries blocked on acquire_slot	The logical-wait parking lot; release_slot hands a freed slot to the front waiter
m_mutex	std::mutex * pointing at the owning core’s m_core_mutex	Slot pool state shares the core lock — no separate mutex, no lock-order hazard

INVARIANT — slot-pool state is guarded by the core mutex. m_mutex is a pointer, not an owned mutex, and the ulock overloads all assert (ulock.owns_lock ()). Callers already holding the core lock (e.g. core_elastic::execute_task) pass it straight through, so a dispatch decision and its slot acquisition are one atomic critical section. A private mutex would force dispatch to take two locks in a fixed order.

INVARIANT — m_parent (core) and the subscriber’s m_identifier (pool) are different void *s. m_parent is the core pointer (reaches adjust_workers/has_queued_task); the inherited m_identifier (§12.6) is the worker-pool pointer (the daemon’s grouping key). Confusing them mis-routes wake-ups or mis-groups pools.

12.6 Publisher / subscriber — how the daemon finds the pools

The daemon must iterate every active slot pool without holding a pointer to each core. Publisher/subscriber inverts that dependency: a pool registers itself on init, so the daemon just walks a map.

// concurrency_slot_subscriber / publisher -- src/thread/concurrency_slot.hpp
class concurrency_slot_subscriber
{
    concurrency_slot_publisher *m_publisher;
    void *m_identifier;   // currently used only as a worker_pool type
};

class concurrency_slot_publisher
{
    friend class concurrency_slot_subscriber;
    std::map<void *, std::vector<concurrency_slot_subscriber *>> m_subscribers;
    std::mutex m_mutex;
};

Struct	Field	Role	Why it exists
concurrency_slot_subscriber	m_publisher	The daemon publisher to (un)subscribe with	Set once at construction; ~subscriber calls deactivate for RAII cleanup
	m_identifier	Grouping key = owning worker_pool pointer; nullptr until activate	Lets the publisher bucket all of one pool’s per-core subscribers together
concurrency_slot_publisher	m_subscribers	map from identifier (worker pool) to its vector of per-core slot pools	The daemon’s iteration target: for each pool, all its cores’ slot banks
	m_mutex	Guards the subscriber map	subscribe/unsubscribe/traverse all lock it

The registration handshake (in concurrency_slot.cpp): concurrency_slot_pool::initialize calls activate (identifier) with the worker-pool pointer; activate stores m_identifier and calls m_publisher->subscribe (identifier, this), appending the pool to m_subscribers[identifier]. So one key (a worker_pool_elastic) groups its N per-core pools. traverse (Func) locks m_mutex and invokes func (identifier, subscribers) per group — the entry point Ch 14’s daemon uses to rebalance a pool’s cores together.

INVARIANT — a subscriber is registered at most once. subscribe assert_release (false) if the pool is already present, and deactivate is idempotent (guards on non-null m_identifier). Double registration would make the daemon steal-and-redistribute the same core’s slots twice per tick.

12.7 concurrency_slot_daemon — the publisher singleton

// concurrency_slot_daemon -- src/thread/concurrency_slot.hpp
class concurrency_slot_daemon : public concurrency_slot_publisher
{
  public:
    static void initialize ();
    static concurrency_slot_publisher *get_publisher ();
  private:
    static concurrency_slot_daemon &get_instance ();   // Meyers singleton
    daemon *m_daemon;
};

The daemon is the publisher (it derives from concurrency_slot_publisher) and is a process-wide singleton reached via get_instance() (a function-local static). get_publisher() returns &get_instance() — exactly the pointer every slot pool subscribes to, so every per-core pool lands under one map.

Field	Role	Why it exists
m_daemon	The cubthread::daemon * running concurrency_slot_daemon_task on a 50 ms looper	The periodic engine that steals soft/hard-waited slots and rebalances them across a pool’s cores (Ch 14); nullptr until create_daemon

REGISTER_DAEMON (concurrency_slot_daemon) wires initialize/finalize into the manager’s daemon lifecycle. Pools subscribe in initialize and unsubscribe in ~concurrency_slot_subscriber; the daemon is the stable singleton at the center.

12.8 thread_entry additions — where a running task’s slot lives

While a task executes, its slot sits on the thread entry (cubthread::entry), so the logical-wait code (lock manager, critical sections) can find and release it without knowing the worker pool. These fields live in thread_entry.hpp — not thread_compat.hpp, which only type-erases the entry to void * for client modules.

// cubthread::entry additions -- src/thread/thread_entry.hpp
thread_resume_suspend_status resume_status;   /* resume status */
// ... condensed ...
#if defined (SERVER_MODE)
  /* concurrency slot held by the entry; only set for workers from an elastic worker pool */
  std::unique_ptr<cubthread::concurrency_slot> m_slot;
#endif

Field	Role	Why it exists
m_slot	The slot the running task holds; SERVER_MODE only, nullptr for non-elastic threads	execute_current_task moves the worker’s slot here before running; the lock/CSS wait path releases it here on a soft wait so another thread can run
resume_status	thread_resume_suspend_status — why the thread suspended/woke	The slot wait reuses the existing suspend/resume machinery; acquire_slot sets THREAD_CONCURRENCY_SLOT_SUSPENDED, saves the prior value, and restores it after
slot_waits (in event_stat)	struct timeval cumulative time this thread blocked on a slot	Slow-query tracing; sits beside lock_waits/latch_waits in the per-entry event_stats

The enum is append-only (persisted meaning must never shift), so Phase 2 tacks its states onto the end:

// thread_resume_suspend_status (tail) -- src/thread/thread_entry.hpp
  THREAD_CONCURRENCY_SLOT_SUSPENDED = 25,
  THREAD_CONCURRENCY_SLOT_RESUMED = 26,
  THREAD_SLEEP_FUNC_SUSPENDED = 27

Value	Meaning	Set by
THREAD_CONCURRENCY_SLOT_SUSPENDED = 25	Thread parked in a slot pool’s m_wait_queue waiting for a slot	acquire_slot before pthread_cond_wait
THREAD_CONCURRENCY_SLOT_RESUMED = 26	A freed slot was handed to this waiter; wake it	release_slot via thread_wakeup_already_had_mutex
THREAD_SLEEP_FUNC_SUSPENDED = 27	Suspended inside a sleep helper (adjacent Phase-2 wait)	logical-wait wiring (Ch 14)

INVARIANT — the slot wait is nested and must restore resume_status. acquire_slot saves the caller’s resume_status (e.g. THREAD_LOCK_RESUMED) before overwriting it with THREAD_CONCURRENCY_SLOT_SUSPENDED, and restores it on exit. The slot wait is auxiliary to an outer lock/CSS wait; leaking THREAD_RESUME_DUE_TO_INTERRUPT upward would make the outer wait believe its resource was never granted even when it was, corrupting lock cleanup ownership. The interrupt is not lost — it is tracked separately through the thread/transaction interrupt state (Ch 14).

12.9 Struct relationships

Figure 12-1 shows ownership within one pool; Figure 12-2, how the daemon reaches every pool through the publisher.

flowchart TB
  subgraph POOL["worker_pool_elastic<Stats>"]
    WPE["m_current_worker<br/>m_max_concurrency<br/>m_max_worker<br/>(atomics)"]
    subgraph CORE["core_elastic (per core)"]
      CE["m_max_concurrency<br/>&amp;m_current_worker<br/>&amp;m_max_worker<br/>m_retired_stats"]
      SLOTS["m_slots :<br/>concurrency_slot_pool"]
      WK["worker_elastic<br/>m_slot"]
    end
  end
  ENTRY["cubthread::entry<br/>m_slot / resume_status"]
  SLOT["concurrency_slot<br/>owner + holder pool"]

  WPE -->|by reference| CE
  CORE --> SLOTS
  CORE --> WK
  SLOTS -->|mints / banks| SLOT
  WK -->|carries on assign| SLOT
  WK -->|moves on execute| ENTRY
  ENTRY -->|holds while running| SLOT

Figure 12-1. A slot’s journey: minted by m_slots, carried by a worker_elastic, moved onto the entry for execution, returned to its owner pool.

flowchart TB
  DAEMON["concurrency_slot_daemon<br/>(singleton, is-a publisher)"]
  MAP["m_subscribers:<br/>worker_pool ptr to vector"]
  P1["core 0 slot pool<br/>(subscriber)"]
  P2["core 1 slot pool<br/>(subscriber)"]
  PN["core N slot pool<br/>(subscriber)"]

  DAEMON --> MAP
  P1 -->|activate/subscribe| MAP
  P2 -->|activate/subscribe| MAP
  PN -->|activate/subscribe| MAP
  DAEMON -->|traverse per pool| P1
  DAEMON -->|traverse per pool| P2
  DAEMON -->|traverse per pool| PN

Figure 12-2. Publish/subscribe: each concurrency_slot_pool subscribes itself under its owning worker pool’s pointer; the daemon traverses the map to rebalance a pool’s cores together.

12.10 Chapter summary — key takeaways

1. Concurrency is a slot count, not a thread count. worker_pool_elastic<Stats> adds three atomics (m_current_worker, m_max_concurrency, m_max_worker) with invariant m_max_worker >= m_max_concurrency so threads overcommit past slots.
2. core_elastic owns a per-core concurrency_slot_pool and holds the two thread-budget atomics by reference, bounding overcommit globally; m_retired_stats preserves counters of downscaled workers.
3. The task-slot pairing invariant is the heart of Phase 2: a worker_elastic (and the entry it runs on) holds a task and a slot together or neither — making “runnable == held slots” literally true.
4. A concurrency_slot has an immutable m_owner_pool, a mobile m_holder_pool, and a soft-wait flag (m_wait/m_wait_since), letting the daemon steal it from a waiting thread and lend it across cores while it still finds its way home.
5. The concurrency_slot_pool shares the owning core’s mutex via m_mutex, banks free slots in m_available_slots, parks starved threads in m_wait_queue, and reaches its core via m_parent.
6. Publisher/subscriber inverts the daemon-to-pool dependency: each pool subscribes itself (keyed by its worker_pool pointer) to the single concurrency_slot_daemon, which traverses the map every 50 ms.
7. A running task’s slot lives on cubthread::entry::m_slot, and the slot wait reuses resume_status with new enum values (THREAD_CONCURRENCY_SLOT_SUSPENDED/RESUMED), saving and restoring the outer wait’s status so nested lock/CSS waits stay correct.

Chapter 13: Phase 2 Flows Slot Acquire Release and the Elastic Run Loop

Chapter 12 laid out the Phase 2 structures. This chapter traces runtime: how a task acquires a slot to run, releases it, and how the elastic worker loop differs line-by-line from the base core_impl / worker_impl loop of Chapters 9-10. All code is PR #7323 (feature/worker_pool_elastic), not yet merged; line numbers are relative to that worktree. The why lives in the companion “Elastic worker pool: the redesign narrative”; here we trace HOW.

13.1 The slot’s two owners and the acquire surface

A concurrency_slot is a permission token: holding one means “you may run a request on this core now.” Its back-pointers drive every flow here.

Field	Role	Why it exists
m_owner_pool (const)	Pool that created the slot; never changes.	A pool destroys only its own slots on shrink, so every core’s m_slot_count stays accurate; a borrowed slot always knows its home.
m_holder_pool	Pool currently holding it; set by set_holder_pool at acquire, cleared by reset.	Routes a borrowed slot (owner != holder) back to whoever needs it on return.
m_wait / m_wait_since	Soft-wait flag + timestamp from start_waiting; has_wait_expired is true after 50 ms.	Lets the rebalance daemon (Ch 14) steal a slot from a thread parked in a logical wait.

Two acquire entry points, and the distinction is Phase 2: try_acquire_slot (non-blocking, returns nullptr if none free — the dispatcher path), and acquire_slot(entry *) (blocking, parks the caller until a slot is handed to it — the logical-wait path of Ch 14). Each has a thin lock-taking overload plus a real std::unique_lock & overload.

13.2 try_acquire_slot and acquire_slot

try_acquire_slot(ulock) is trivial: empty → nullptr; else pop m_available_slots.front(), set_holder_pool(this), check_surplus_slots, return. acquire_slot(thread_p, ulock) is the blocking path:

// concurrency_slot_pool::acquire_slot -- src/thread/concurrency_slot.cpp
    if (!m_available_slots.empty ())          /* FAST: hand slot immediately */
      {
        thread_p->m_slot = std::move (m_available_slots.front ());
        m_available_slots.pop ();
        thread_p->m_slot->set_holder_pool (this);
        check_surplus_slots (); return true;
      }
    m_wait_queue.push_back (thread_p);         /* SLOW: enqueue self, sleep */
    ulock.unlock ();
    saved_status = thread_p->resume_status;    /* <- remember OUTER wait's result */
    thread_p->resume_status = THREAD_CONCURRENCY_SLOT_SUSPENDED;
    while (thread_p->resume_status == THREAD_CONCURRENCY_SLOT_SUSPENDED)
      { pthread_cond_wait (&thread_p->wakeup_cond, &thread_p->th_entry_lock); }

After the wait, exactly one wake reason holds:

1. THREAD_CONCURRENCY_SLOT_RESUMED — a peer’s release_slot already planted a slot in thread_p->m_slot and popped the waiter. Restore resume_status = saved_status, return true. No dequeue needed.
2. THREAD_RESUME_DUE_TO_INTERRUPT — cancellation woke it. Restore saved_status, re-lock, then: if thread_p->m_slot is set (slot handed over then interrupted — a race) give it back via release_slot and return false; else std::find/erase self from m_wait_queue (absent means a concurrent release_slot already removed it as stale) and return false.

Invariant 13-C (outer wait result survives the nested slot wait). The slot wait is auxiliary, nested inside a LOCK/CSS wait. saved_status is captured before and restored on every return path. Leaking THREAD_RESUME_DUE_TO_INTERRUPT up would make LOCK/CSS believe the original resource was never granted even when it was, corrupting their wakeup ownership. The interrupt is not lost — it lives in the separate thread/transaction interrupt state.

stateDiagram-v2
  [*] --> CheckSlots : acquire_slot
  CheckSlots --> Return_true : slots free, pop, set holder
  CheckSlots --> Waiting : empty, enqueue self, save status, cond_wait
  Waiting --> Return_true : SLOT_RESUMED
  Waiting --> HandedBack : INTERRUPT, m_slot set
  Waiting --> EraseSelf : INTERRUPT, m_slot null
  HandedBack --> Return_false : release_slot
  EraseSelf --> Return_false : erase from wait_queue

Figure 13-1. Branch-complete concurrency_slot_pool::acquire_slot.

13.3 release_slot — shrink, store, or hand off

release_slot(slot, ulock) decides a freed slot’s fate. Three outcomes, in order:

// concurrency_slot_pool::release_slot -- src/thread/concurrency_slot.cpp
    slot->reset ();
    if (slot->m_owner_pool == this && m_target_count < m_slot_count)
      { slot.reset (); m_slot_count--; check_surplus_slots (); return; }  /* SHRINK */
    while (true)
      {
        if (m_wait_queue.empty ())                                        /* STORE */
          { m_available_slots.emplace (std::move (slot)); break; }
        waiter = m_wait_queue.front (); m_wait_queue.pop_front ();        /* HAND OFF */
        ulock.unlock ();  waiter->lock ();          /* core mutex dropped BEFORE entry lock */
        if (waiter->resume_status == THREAD_CONCURRENCY_SLOT_SUSPENDED)
          {
            waiter->m_slot = std::move (slot); waiter->m_slot->set_holder_pool (this);
            thread_wakeup_already_had_mutex (waiter, THREAD_CONCURRENCY_SLOT_RESUMED);
            waiter->unlock (); ulock.lock (); break;
          }
        waiter->unlock (); ulock.lock ();            /* waiter already INTERRUPTED: retry next */
      }
    check_surplus_slots ();

• Shrink fires only when the slot is owned by this and the pool is over target (adjust_concurrency lowered m_target_count): the slot is deleted, not pooled — the lazy half of a concurrency shrink (§13.5).
• Store — no waiters, push onto m_available_slots.
• Hand off — pop the front waiter, drop the core mutex, take its th_entry_lock, re-check resume_status. Still SLOT_SUSPENDED → plant slot + wake with THREAD_CONCURRENCY_SLOT_RESUMED (the branch §13.2 case 1 wakes into). Already interrupted → skip and retry the next waiter; the slot must find a home or be stored.

Invariant 13-F (lock order: core mutex before entry lock). release_slot never holds both: it ulock.unlock()s before waiter->lock(), re-locks after waiter->unlock(). Holding the core mutex while grabbing th_entry_lock reintroduces lock inversion against the acquire path.

The RTTI bridge. concurrency_slot_pool is template-agnostic but must call the templated core, so it stores m_parent as void * and downcasts at runtime. wakeup_workers(ulock) dynamic_casts m_parent to each core_elastic<stats_t::on/off> instantiation and calls that core’s adjust_workers(ulock) (see Ch 14); has_queued_task(ulock) uses the identical two-arm cast to ask the core !m_task_queue.empty (). Both have a third, fallthrough arm: if the downcast matches neither instantiation, wakeup_workers is a silent no-op and has_queued_task returns false. This is defensive — m_parent is always a core_elastic, so it is effectively unreachable — but it keeps the bridge total.

13.4 return_to_pool and return_slot — the owner/holder three-hop

When a task finishes, its slot is routed to whoever needs it most, not just pushed back:

// concurrency_slot::return_to_pool -- src/thread/concurrency_slot.cpp
    auto r = m_owner_pool->return_slot (std::move (slot), false);      /* 1. offer owner */
    if (r) { r = m_holder_pool->return_slot (std::move (r), false);    /* 2. offer holder */
             if (r) { m_owner_pool->return_slot (std::move (r), true); } } /* 3. force home */

return_slot(slot, force) is the accept/reject gate — it consumes the slot on any demand signal, else std::moves ownership back to the caller:

// concurrency_slot_pool::return_slot -- src/thread/concurrency_slot.cpp
    if (has_queued_task (ulock) ||                                    /* queued work here */
        (m_available_slots.empty () && !m_wait_queue.empty ()) ||     /* blocked acquirers */
        (slot->m_owner_pool == this && m_slot_count > m_target_count) || force)
      { release_slot (std::move (slot), ulock); wakeup_workers (ulock); return nullptr; }
    return slot;                                                       /* rejected */

Offer to the owner first (may reclaim/destroy if it shrank), then the holder (may have queued work), and if neither wants it force it home so it is never dropped — the force hop guarantees termination and conserves m_slot_count. On accept, wakeup_workers fires, so a slot returning into a core with queued tasks immediately triggers adjust_workers.

stateDiagram-v2
  [*] --> OfferOwner : return_to_pool
  OfferOwner --> Done : consumed
  OfferOwner --> OfferHolder : rejected
  OfferHolder --> Done : consumed
  OfferHolder --> ForceOwner : rejected
  ForceOwner --> Done : force always accepts
  Done --> [*]

Figure 13-2. Owner to holder to force return protocol of return_to_pool.

13.5 needs_slot, surplus, and adjust_concurrency

needs_slot(ulock) — the daemon’s demand probe: !m_wait_queue.empty () || (m_available_slots.empty () && has_queued_task (ulock)). available_slots() returns m_available_slots.size () with no lock of its own — every caller already holds the core mutex. check_surplus_slots stamps the surplus clock only on the empty→surplus transition, so m_surplus_since measures how long the core has been over-provisioned (the daemon requires ≥ 2 s before stealing):

// concurrency_slot_pool::check_surplus_slots -- src/thread/concurrency_slot.cpp
    if (m_available_slots.size () >= SLOT_SURPLUS_THRESHOLD)   /* == 2 */
      { if (!m_surplus) { m_surplus_since = steady_clock::now (); m_surplus = true; } }
    else { m_surplus = false; }

adjust_concurrency(concurrency, ulock) grows/shrinks the owned slot set to a new target (driven by PRM_ID_MAX_REQUEST_CONCURRENCY):

// concurrency_slot_pool::adjust_concurrency -- src/thread/concurrency_slot.cpp
    m_target_count = concurrency;
    if (m_slot_count < concurrency)           /* GROW */
      { i = m_slot_count; m_slot_count = concurrency;
        for (; i < concurrency; i++)
          { release_slot (std::unique_ptr<concurrency_slot> (new concurrency_slot (this)), ulock); } }
    else if (m_slot_count > concurrency)      /* SHRINK: available+owned only */
      { while (m_slot_count > concurrency && !m_available_slots.empty ())
          { auto s = std::move (m_available_slots.front ()); m_available_slots.pop ();
            if (s->m_owner_pool == this) { s.reset (); m_slot_count--; }  /* destroy owned */
            else { bucket.emplace_back (std::move (s)); } }               /* keep borrowed */
        for (auto &s : bucket) { m_available_slots.push (std::move (s)); }
        check_surplus_slots (); }

• Grow creates the shortfall as new owned slots and pushes each through release_slot, so blocked acquirers in m_wait_queue are woken immediately instead of leaving slots idle.
• Shrink is eager only for slots that are both available and owned; borrowed slots (owner != this) are bucketed and re-pooled (never destroyed here), and in-flight slots aren’t touched. Their reduction happens lazily via release_slot’s shrink branch (§13.3) as they return and see m_target_count < m_slot_count.

Invariant 13-D (a pool destroys only its own slots). Both shrink paths gate s.reset() on s->m_owner_pool == this, keeping every core’s m_slot_count an accurate count of slots it created.

13.6 core_elastic::execute_task — dispatch gated by a slot

Base execute_task (Ch 9) asks only “is a worker free?” The elastic override adds the slot gate first — no slot, no dispatch:

// worker_pool_elastic<Stats>::core_elastic::execute_task -- src/thread/thread_worker_pool_elastic.hpp
    if (!this->m_parent_pool->is_running ()) { task_p->retire (); return; }   /* rejected */
    wrapped_task task_ref (task_p);
    std::unique_lock<std::mutex> ulock (this->m_core_mutex);
    unique_slot slot = m_slots.try_acquire_slot (ulock);
    if (slot)
      { worker_p = get_or_make_available_worker ();
        if (worker_p)
          { if (!this->m_task_queue.empty ())        /* FIFO: run oldest queued, enqueue new */
              { this->m_task_queue.push_back (std::move (task_ref));
                wrapped_task q = std::move (this->m_task_queue.front ()); this->m_task_queue.pop_front ();
                ulock.unlock (); try_execute_task_with_slot (worker_p, std::move (q), std::move (slot)); }
            else { ulock.unlock (); try_execute_task_with_slot (worker_p, std::move (task_ref), std::move (slot)); }
            return; }
        release_slot (std::move (slot), ulock); }    /* slot but no worker (global cap) */
    this->m_task_queue.push_back (std::move (task_ref)); /* fall through: queue */

Terminal outcomes: (1) pool stopped → retire; (2) slot + worker + non-empty queue → run oldest, enqueue new (FIFO); (3) slot + worker + empty queue → run new directly; (4) slot but no worker (global m_max_worker cap) → release slot, enqueue; (5) no slot → enqueue. try_execute_task_with_slot handles thread-start failure: if assign_task returns the pair back, it re-locks, push_fronts the task (FIFO), release_slots the slot, pushes the worker back to m_available_workers, returns false.

13.7 The elastic run loop vs. the base loop

worker_elastic overrides three virtuals — get_new_task, execute_current_task, run — plus the available-worker machinery.

get_new_task — task and slot travel together, fast path first. Elastic calls get_task_and_slot_or_become_available (a try_acquire_slot under the core mutex that dequeues a task only if a slot is free). If it returns a populated optional — a queued task and a free slot were both obtained under the core mutex — get_new_task bumps stats::id::found_in_queue, emplaces both, and returns true immediately, skipping the CV wait entirely. Only when it returns nullopt (which also enqueues the worker into m_available_workers) does the worker take m_task_mutex and wait:

// worker_elastic::get_new_task -- src/thread/thread_worker_pool_elastic.hpp
    if (queued.has_value ()) { stats::time_and_increment (m_stats, stats::id::found_in_queue);
      this->m_wrapped_task.emplace (std::move (queued->first)); m_slot = std::move (queued->second);
      return true; }                                            /* FAST: no wait */
    ulock.lock ();
    assert ((!this->m_wrapped_task.has_value () && !m_slot)
            || (this->m_wrapped_task.has_value () && m_slot));   /* INV 13-A */
    if ((!this->m_wrapped_task.has_value () && !m_slot) && !this->m_stop)
      { condvar_wait (this->m_task_cv, ulock, /* persistent: infinite; else idle_timeout */,
          [this] { return (this->m_wrapped_task.has_value () && m_slot) || this->m_stop; }); }

Invariant 13-A (a worker holds a task and its slot together, or neither). The paired assertion (!task && !slot) || (task && slot) is checked on entry and after the wait, and the CV predicate is (task && slot) || stop. This makes the slot the true admission token: a worker is never woken with a task it lacks permission to run, nor a slot without a task. The producer side enforces the coupling by popping a task only once try_acquire_slot succeeds.

The m_stop branch also differs: elastic calls become_available(*this) to keep a stopping worker visible before falling through. On the no-task exit it clears m_has_thread, calls finish_run (retire context), returns false; on found-task it unlocks, bumps wakeup_with_task, check_worker_not_availables (debug), returns true.

execute_current_task — the slot lives on the thread entry during execution:

// worker_elastic::execute_current_task -- src/thread/thread_worker_pool_elastic.hpp
    this->m_context_p->m_slot = std::move (m_slot);        /* slot -> thread entry */
    this->m_wrapped_task->execute (*this->m_context_p);
    if (this->m_context_p->m_slot)                          /* task may have swapped/lost it */
      { this->m_context_p->m_slot->return_to_pool (std::move (this->m_context_p->m_slot));
        this->m_context_p->m_slot = nullptr; }

The if (m_context_p->m_slot) guard matters: a task whose slot was stolen mid logical-wait (§13.1) returns with a null slot and skips return_to_pool; otherwise the slot goes home via the three-hop of §13.4. Threading the slot through the thread_entry is what lets a running request itself wait on / return slots (Ch 14).

run — reap-on-exit. Elastic run calls the base Ch 10 loop, then get_retire_if_excess(this), which removes a now-surplus thread only under three gates, all under the core mutex:

// core_elastic::get_retire_if_excess -- src/thread/thread_worker_pool_elastic.hpp
    if (this->m_workers.size () > m_max_concurrency && !this->has_workers_snapshot_readers ())
      { auto available_it = std::find (m_available_workers.begin (), m_available_workers.end (), w);
        if (available_it != this->m_available_workers.end ())   /* still idle, not re-selected */
          { this->m_available_workers.erase (available_it);     /* erase from available FIRST */
            /* find w in m_workers */
            stats::accumulate (w->m_stats, m_retired_stats);    /* preserve counters */
            this->m_workers.erase (worker_it); release_available_worker (); } }

Gate 1: m_workers.size() > m_max_concurrency (this core is over its slot target). Gate 2: !has_workers_snapshot_readers() (Inv 13-E). Gate 3: w must still be present in m_available_workers. Between finish_run and this check, a concurrent execute_task may have re-claimed w for a queued task; then find returns end() (// not selected yet fails) and the reap is skipped — the worker keeps running. Only when all three hold does the code erase w from m_available_workers first, then locate and erase its unique_ptr<worker> from m_workers, folding stats into m_retired_stats before destruction.

Invariant 13-E (no reap while a snapshot reader is live). stop_execution, get_stats, and map_running_contexts build a snapshot_guard that copies raw worker * out of m_workers and bumps m_workers_readers. Erasing a unique_ptr<worker> under a live snapshot would dangle those pointers, so get_retire_if_excess skips reaping when has_workers_snapshot_readers (); the worker retries on its next idle exit. Its stats are folded into m_retired_stats first so observability (Ch 11) never loses a retired worker’s counters.

Available-worker rework — global overcommit ceiling. Base get_available_worker only pops from a fixed m_available_workers. Elastic wraps it in get_or_make_available_worker, which grows the vector up to a global cap:

// core_elastic::get_or_make_available_worker -- src/thread/thread_worker_pool_elastic.hpp
    worker_p = this->get_available_worker ();               /* base: reuse an idle worker */
    if (!worker_p && reserve_available_worker ())           /* none idle -> try to grow */
      { this->m_workers.push_back (this->allocate_worker ());
        worker_p = this->m_workers.back ().get (); worker_p->set_parent_core (*this); }
    return static_cast<worker_elastic *> (worker_p);        /* nullptr if at global cap */

reserve_available_worker is a lock-free CAS loop on the shared m_current_worker atomic, capped by m_max_worker; it returns false when expected >= m_max_worker. Both atomics are pool-wide, shared by reference into every core_elastic, so the cap on total threads is global even though each core grows its own m_workers. release_available_worker (fetch_sub(1)) returns a permit on reap. This is a soft cap on threads, distinct from the per-core m_max_concurrency cap on slots — the essence of the redesign: threads may overcommit past admitted concurrency so a blocked thread’s slot can be lent onward.

get_pool_size — unchanged, and that is the subtlety. worker_pool_elastic does not override it; it still returns m_pool_size, the entry-reservation count fixed at construction. In the base pool that equals the live worker count; in the elastic pool the live thread count is dynamic (Σ core->m_workers.size (), bounded by m_max_worker) and decoupled from get_pool_size. Callers that read get_pool_size as “threads that exist” now get the reservation ceiling — use get_runtime_stats / get_max_worker for the live picture.

stateDiagram-v2
  [*] --> BaseRun : base worker_impl run loop
  BaseRun --> GetNewTask
  GetNewTask --> Execute : task+slot
  Execute --> GetNewTask
  GetNewTask --> FinishRun : idle or stop
  FinishRun --> ReapCheck : clear m_has_thread
  ReapCheck --> Reap : over max_concurrency, no readers, still available
  ReapCheck --> Keep : any gate fails
  Reap --> [*]
  Keep --> [*]

Figure 13-3. Elastic worker lifecycle: base loop plus slot-coupled fetch and three-gate reap-on-exit.

13.8 Chapter summary — key takeaways

1. The slot is the admission token. try_acquire_slot (non-blocking dispatcher) and acquire_slot (blocking logical-wait) are the only ways to become runnable; get_new_task couples task and slot so a worker never runs without permission (Inv 13-A).
2. Ownership vs. holding drives every return. m_owner_pool is immutable; return_to_pool offers owner→holder→force (Fig 13-2), conserving m_slot_count — a core only destroys slots it created (Inv 13-D).
3. release_slot has three fates — shrink an over-target owned slot, store when no waiter, or hand off to a blocked acquirer — always dropping the core mutex before a waiter’s th_entry_lock (Inv 13-F).
4. The interrupt race is handled explicitly. acquire_slot restores the outer wait’s resume_status on every path (Inv 13-C) and reclaims a slot handed over just before an interrupt.
5. adjust_concurrency shrinks eagerly for available+owned slots, lazily (via release_slot) for in-flight ones, so reducing concurrency never blocks on running tasks; check_surplus_slots times the over-provisioned window for the daemon.
6. Threads overcommit past admitted concurrency. reserve_available_worker’s CAS loop enforces a global m_max_worker thread cap, independent of the per-core m_max_concurrency slot cap; get_new_task’s fast path skips the CV wait when a queued task and a free slot are found together (found_in_queue); get_retire_if_excess reaps a surplus thread only when it is over target, no snapshot_guard reader is live (Inv 13-E), and the worker is still in m_available_workers (else it was re-selected and keeps running).
7. get_pool_size is now a reservation ceiling, not a live count — the elastic pool leaves it unoverridden; read live thread state via get_runtime_stats / get_max_worker.

Chapter 14: Phase 2 Flows the Slot Daemon and Logical Wait Wiring

Chapters 12–13 mapped the elastic pool’s slot structures and the synchronous acquire/release/return paths a running worker takes. This chapter closes the loop with the two asynchronous mechanisms that keep the slot budget liquid: the concurrency_slot_daemon (~50 ms tick redistributing idle and stuck slots) and the suspension wiring that surrenders a slot when a thread blocks on a lock, a CSS job queue, or a PL call, and reacquires one on wake. Reader question: how is a freed slot redistributed, and how does a lock / CSS / PL wait surrender and reacquire a slot?

Worktree note. All code is from the PR #7323 feature/worker_pool_elastic worktree — not merged to develop; POSHINT line numbers are relative to it.

Cross-links. Elastic-pool motivation is cubrid-thread-manager.md; the per-core concurrency_slot_pool fields, SLOT_SURPLUS_THRESHOLD, the pub/sub subscribe/activate plumbing, and try_acquire_slot/release_slot/ return_slot/borrow_surplus_slots are Chapters 12–13. This chapter reads those as given.

14.1 The daemon shell and the tick contract

REGISTER_DAEMON (concurrency_slot_daemon) registers a singleton that is the publisher (concurrency_slot_daemon : concurrency_slot_publisher), so the object owning the m_subscribers map is the one whose task walks it. create_daemon wraps a concurrency_slot_daemon_task in a fixed 50 ms looper:

// concurrency_slot_daemon::create_daemon -- src/thread/concurrency_slot.cpp
looper loop = looper (std::chrono::milliseconds (50));
m_daemon = cubthread::get_manager ()->create_daemon (loop, daemon_task, "concurrency");

execute runs a five-step contract: (1) reconcile parameters, then per registered worker-pool identifier (2) steal 50 ms-expired slots off blocked entries, (3) steal surplus off over-provisioned cores iff demand exists, (4) redistribute, (5) wake fed cores. Steps 2–5 run inside the traverse_subscribers callback.

INVARIANT — the tick is the only unforced slot mover. Synchronous release_slot/return_slot move a slot only when a task ends or a waiter arrives; idle surplus and slots pinned in blocked entries are reclaimed only by the daemon. If it stalls, the pool degrades to static sizing but never corrupts — every slot still routes home via return_to_pool.

14.2 Step 1 — parameter reconciliation and cgroup-aware core count

check_and_propagate_parameters re-reads the two dynamic tunables and acts only on drift from the task’s cached copies:

// concurrency_slot_daemon_task::check_and_propagate_parameters -- src/thread/concurrency_slot.cpp
if (max_request_concurrency != m_max_request_concurrency || max_request_worker != m_max_request_worker)
  {
    tune_parameters (max_request_concurrency, max_request_worker);          /* clamp to sane band */
    css_set_max_concurrency_and_workers (max_request_concurrency, max_request_worker); /* push to live pool */
    prm_set_integer_value (PRM_ID_MAX_REQUEST_CONCURRENCY, max_request_concurrency);   /* write back clamped */
    /* ...set worker, update m_ caches... */
  }

tune_parameters clamps both into [core_count, CSS_MAX_CLIENT_COUNT] and forces worker >= concurrency:

// concurrency_slot_daemon_task::tune_parameters -- src/thread/concurrency_slot.cpp
std::size_t core_count = system_core_count ();
max_request_concurrency = std::min (std::max (max_request_concurrency, core_count), max_client_count);
max_request_worker      = std::min (std::max (max_request_worker, core_count), max_client_count);
if (max_request_worker < max_request_concurrency) { max_request_worker = max_request_concurrency; }

system_core_count returns os::resources::cpu::effective ().adjusted_max, not hardware_concurrency. effective() (lazy static in resources.cpp) intersects process affinity, kernel online, and the cgroup CPU limit (cpu.cfs_quota_us/cpu.max, read by cgroup.cpp), then floors and clamps >= 1:

// cpu::effective (correction & adjustment) -- src/base/resources.cpp
ctx.adjusted_max = static_cast<std::size_t> (std::floor (ctx.max));   /* min of affinity, online, cgroup quota */
if (ctx.adjusted_max <= 0) { ctx.adjusted_max = 1; }

So a container pinned to cpu.max=2 on a 64-core host yields core_count == 2. Defaults: max_request_concurrency = system_core_count()*3 (source note “adjusted based on CBRD-26636”), ceiling CSS_MAX_CLIENT_COUNT; max_request_worker = CSS_MAX_CLIENT_COUNT — the tunables from CBRD-26688 / CBRD-26690 / CBRD-26977. css_set_max_concurrency_and_workers propagates a live change through adjust_runtime_parameter (Ch 12).

14.3 traverse_subscribers and the per-identifier body

traverse_subscribers forwards to base traverse, which holds the publisher mutex for the whole walk and calls the callback once per map entry: key = worker-pool identifier, value = that pool’s per-core concurrency_slot_pool subscribers.

// concurrency_slot_publisher::traverse -- src/thread/concurrency_slot.hpp
std::lock_guard<std::mutex> lock (m_mutex);
for (auto &subset : m_subscribers) { func (subset.first, subset.second); }  /* (identifier, subs) */

INVARIANT — the subscriber set is frozen for one identifier’s tick. The publisher mutex is held across traverse, so no core can subscribe/unsubscribe mid-harvest and distribute_slots can never feed a pool being torn down. This is a different lock from each pool’s core mutex (retaken per-pool below). execute’s lambda takes subs by value (auto subs) — a private copy it can clear and rebuild freely.

flowchart TD
  A["execute tick"] --> B["check_and_propagate_parameters"]
  B --> C["traverse_subscribers per identifier"]
  C --> D["steal_from_entries_if_excess<br/>harvest 50ms-expired slots off blocked entries"]
  D --> E{"has_slot_demand?"}
  E -->|yes| F["steal_from_cores_if_excess<br/>borrow_surplus_slots per core"]
  E -->|no| G["distribute_slots by score"]
  F --> G
  G --> H["wakeup_workers on fed cores"]
  H --> C

Figure 14-1. One execute tick; D–H repeat per identifier.

has_slot_demand gates step 3, returning true if any pool’s needs_slot holds (non-empty wait queue, or empty available list with a queued task — Ch 13).

14.4 Soft-wait promotion and stealing from entries

Step 2 reclaims a slot from an entry blocked long enough that its slot is wasted, via a two-stage soft-to-hard promotion carried on the slot. A lock suspend does not release its slot — it marks it soft-waiting:

// concurrency_slot::start_waiting / has_wait_expired -- src/thread/concurrency_slot.cpp
void concurrency_slot::start_waiting () { m_wait = true; m_wait_since = std::chrono::steady_clock::now (); }
void concurrency_slot::stop_waiting () { m_wait = false; }
bool concurrency_slot::has_wait_expired (const std::chrono::time_point<std::chrono::steady_clock> &now)
{ constexpr auto threshold = std::chrono::milliseconds (50); return m_wait && (now - m_wait_since >= threshold); }

The daemon promotes soft to hard by mapping over every running context and stealing any expired slot:

// concurrency_slot_daemon_task::steal_from_entries_if_excess -- src/thread/concurrency_slot.cpp
auto func = [&slots, &now] (entry &thread_ref, bool &stop) {
  thread_ref.lock ();
  if (thread_ref.m_status == cubthread::entry::status::TS_WAIT &&
      thread_ref.m_slot && thread_ref.m_slot->has_wait_expired (now))
    { slots.emplace_back (std::move (thread_ref.m_slot)); thread_ref.m_slot = nullptr; }  /* steal out */
  thread_ref.unlock ();
};
pool->map_running_contexts (func);   /* stats on/off both dispatched */

INVARIANT — a suspended entry’s m_slot is guarded by the thread entry lock. start_waiting/stop_waiting run under th_entry_lock (the hooks below); the daemon takes thread_ref.lock () before reading m_status/m_slot and moving the slot out. Same lock both sides → daemon and waking thread can never claim the same slot. The TS_WAIT check is the second guard: a TS_RUN entry holding a slot is skipped. The 50 ms threshold is the design — a lock wait shorter than one daemon period keeps its slot; a longer one loses it.

stateDiagram-v2
  [*] --> Held : acquire_slot on task start
  Held --> SoftWait : LOCK suspend start_waiting
  SoftWait --> Held : wake before 50ms, stop_waiting
  SoftWait --> Stolen : daemon has_wait_expired at 50ms
  Stolen --> Reacquired : wake finds m_slot null, acquire_slot blocks
  Held --> Returned : CSS suspend returns slot now
  Returned --> Reacquired : wake calls acquire_slot
  Reacquired --> [*]

Figure 14-2. Slot lifecycle across a logical wait.

14.5 Inter-core surplus transfer and score-based distribution

Step 3 harvests idle surplus only under demand: each core sheds via borrow_surplus_slots (Ch 13: down to SLOT_SURPLUS_THRESHOLD, only after 2000 ms in surplus). Step 4, distribute_slots, scores every core, sorts descending, and hands slots round-robin over that order — score is deficit-weighted:

// concurrency_slot_pool::get_score -- src/thread/concurrency_slot.cpp
return -static_cast<float> (m_available_slots.size ()) + m_wait_queue.size () * 2.0f;  /* neediest = highest */

// concurrency_slot_daemon_task::distribute_slots -- src/thread/concurrency_slot.cpp
subs.clear ();
std::sort (scores.begin (), scores.end (), [] (const auto &a, const auto &b) { return a.second > b.second; });
while (!slots.empty ()) {
  auto slot = std::move (slots.back ()); slots.pop_back ();
  auto pool = static_cast<concurrency_slot_pool *> (scores[i++ % scores.size ()].first);
  pool->release_slot (std::move (slot));                 /* Ch 13: store or hand to a waiter */
  if (std::find (subs.begin (), subs.end (), ...) == subs.end ()) { subs.push_back (pool); }  /* record fed pool */
}

distribute_slots rebuilds subs to only the pools it fed, so step 5’s wakeup_workers touches exactly those cores. release_slot is the Ch 13 routine (store, or hand to a queued waiter, honoring owner-vs-holder shrink).

INVARIANT — no slot is dropped in a tick. Everything stolen in steps 2–3 is placed in step 4 (execute asserts slots.empty ()), or destroyed inside release_slot only when the owning core exceeds m_target_count. A slot may change holder here (owner is fixed for life) — that is how a slot minted on core A runs work on core B, and return_to_pool’s owner-then-holder routing still sends it home.

14.6 Suspension wiring — CSS hard-release vs LOCK soft-wait

The legacy-C blocking primitives bracket every logical wait with thread_prepare_suspension before the pthread_cond_wait loop and thread_prepare_resumption after. Suspension flips the entry to TS_WAIT and, for a slot-managed entry, branches on why it blocks:

// thread_prepare_suspension -- src/thread/thread_entry.cpp
if (thread_p->m_slot) switch (suspended_reason) {
  case THREAD_CSS_QUEUE_SUSPENDED:                               /* long idle wait: give slot back now */
    holder = thread_p->m_slot->get_holder_pool ();
    { auto slot = std::move (thread_p->m_slot); thread_p->m_slot = nullptr; slot->return_to_pool (std::move (slot)); }
    break;
  case THREAD_LOCK_SUSPENDED:                                   /* short-ish: keep slot, mark soft-wait */
    holder = thread_p->m_slot->get_holder_pool ();
    thread_p->start_waiting ();
    break;
  default: break;
}

Suspend reason	Slot action	Rationale
THREAD_CSS_QUEUE_SUSPENDED	return_to_pool now (hard release)	Handler parked on the CSS queue is idle indefinitely; a held slot would starve real work
THREAD_LOCK_SUSPENDED	start_waiting (retain slot)	Lock waits are usually brief; keep the slot for a fast grant, let the daemon steal it after 50 ms

entry::start_waiting/stop_waiting are one-line SERVER_MODE-only forwards to m_slot. In both cases holder is captured before the slot moves — resumption needs the pool pointer even when m_slot is now null.

14.7 Resumption wiring and the wakeup loop

After the outer wait loop exits (predicate: resume_status no longer equals the suspend reason), thread_prepare_resumption reconciles the slot, running only when holder is set, and branches three ways:

// thread_prepare_resumption -- src/thread/thread_entry.cpp
if (holder) {
  if (timedout || thread_p->resume_status == THREAD_RESUME_DUE_TO_INTERRUPT
               || thread_p->resume_status == THREAD_RESUME_DUE_TO_SHUTDOWN) {   /* branch 0: aborting */
    if (thread_p->m_slot) { holder->release_slot (std::move (thread_p->m_slot)); thread_p->m_slot = nullptr; }
  } else {
    if (thread_p->m_slot) { thread_p->stop_waiting (); }              /* branch 1: soft-wait survived */
    else { holder->acquire_slot (thread_p); }                        /* branch 2: reacquire, may block */
  }
}
thread_p->m_status = status;   /* restore pre-wait TS_RUN/TS_CHECK */

• Branch 0 (timeout / interrupt / shutdown). Wait is unwinding. If the slot is still held (unexpired LOCK soft-wait), give it back via release_slot; never block to reacquire. If already stolen or CSS-returned, nothing to do.
• Branch 1 (normal resume, slot retained). LOCK only, wait < 50 ms so the daemon never stole it. stop_waiting clears m_wait; resume at zero cost.
• Branch 2 (normal resume, slot gone). CSS (returned at suspend) and the LOCK case whose slot the daemon stole; acquire_slot blocks until a slot is handed back (Ch 13 wait-queue path).

INVARIANT — the nested slot wait must not clobber the original resume result. acquire_slot runs a second pthread_cond_wait keyed on THREAD_CONCURRENCY_SLOT_SUSPENDED but saves/restores resume_status around it (Ch 13). Otherwise a THREAD_LOCK_RESUMED the lock manager already recorded would be overwritten and lock_suspend would wrongly conclude the lock was not granted. The interrupt is not lost — it persists in the transaction interrupt state for the upper cancellation path.

thread_suspend_wakeup_and_unlock_entry is the no-timeout driver (passes timedout = false, so branch 0 needs interrupt/shutdown); the timeout variant passes error == ER_CSS_PTHREAD_COND_TIMEDOUT and can hit branch 0 on a plain timeout.

// thread_suspend_wakeup_and_unlock_entry -- src/thread/thread_entry.cpp
thread_prepare_suspension (thread_p, status, suspended_reason, start_time, holder);
while (thread_p->resume_status == suspended_reason)
  { pthread_cond_wait (&thread_p->wakeup_cond, &thread_p->th_entry_lock); }   /* outer wait */
thread_prepare_resumption (thread_p, status, suspended_reason, false, start_time, holder);
thread_p->unlock ();

14.8 The three call sites

Lock manager (logical wait). lock_suspend registers lock-wait bookkeeping then calls thread_suspend_wakeup_and_unlock_entry (..., THREAD_LOCK_SUSPENDED) — that one call is the entire slot handoff, driving the soft-wait path (14.6– 14.7). On wake it reads resume_status: THREAD_LOCK_RESUMED = granted (slot back in hand per branch 1/2), THREAD_RESUME_DUE_TO_INTERRUPT = torn down (branch 0 already dropped any retained slot). The lock code never touches the slot.

Network wake-eligible switch. net_server_wakeup_workers interrupts a transaction’s suspended worker, keyed on resume_status:

// net_server_wakeup_workers -- src/communication/network_sr.c
switch (suspended_p->resume_status) {
  case THREAD_CSECT_READER_SUSPENDED: case THREAD_CSECT_WRITER_SUSPENDED:
  case THREAD_CSECT_PROMOTER_SUSPENDED: case THREAD_LOCK_SUSPENDED:
  case THREAD_JOB_QUEUE_SUSPENDED:
    wakeup_now = false; break;                              /* CS / lock waits NOT interrupt-woken here */
  case THREAD_CSS_QUEUE_SUSPENDED: case THREAD_CONCURRENCY_SLOT_SUSPENDED:  /* Phase-2 slot wait: eligible */
  /* ...other idle waits... */
    wakeup_now = true; break;
  /* ...resumed states -> false... */
}
if (wakeup_now) { thread_wakeup_already_had_mutex (suspended_p, THREAD_RESUME_DUE_TO_INTERRUPT); }

The Phase-2 addition is THREAD_CONCURRENCY_SLOT_SUSPENDED in the “wake now” group: a thread parked inside acquire_slot (resumption branch 2) must be interruptible, or a client cancel could hang behind slot starvation. THREAD_LOCK_SUSPENDED stays “do not wake” — its slot is a soft-wait the daemon reclaims.

PL out-of-process call. A stored procedure blocked on the Java PL server would otherwise pin a slot for the whole round trip, so read_data_from_java releases around the blocking receive and reacquires after:

// read_data_from_java -- src/sp/pl_execution_stack_context.hpp
#if defined (SERVER_MODE)
  auto *holder = thread_concurrency_slot_release (m_thread_p);   /* give slot back before blocking recv */
#endif
  int error = conn->receive_buffer (b, &interrupt_func, 500);
#if defined (SERVER_MODE)
  thread_concurrency_slot_acquire (m_thread_p, holder);          /* reacquire (may block) after recv */
#endif

thread_concurrency_slot_release (in thread_manager.hpp) locks the entry, moves m_slot out, return_to_pools it, and returns the holder pool; thread_concurrency_slot_acquire re-blocks on acquire_slot under TS_WAIT if holder is set. Unlike LOCK, PL uses an explicit release/reacquire pair because a PL call is expected to be long — no point retaining the slot for 50 ms.

14.9 Chapter summary — key takeaways

1. The daemon is the only unforced slot mover. A 50 ms looper reconciles parameters, steals expired slots off blocked entries, steals idle surplus off over-provisioned cores (under demand), redistributes by score, wakes fed cores.
2. Soft-to-hard promotion is the LOCK policy. start_waiting stamps m_wait_since; has_wait_expired promotes at 50 ms; steal_from_entries_if_excess yanks the slot only from a TS_WAIT entry whose wait expired — serialized against the waker by the entry lock.
3. CSS/PL release eagerly, LOCK retains optimistically. Suspension return_to_pools for THREAD_CSS_QUEUE_SUSPENDED, only marks soft-wait for THREAD_LOCK_SUSPENDED; PL uses an explicit thread_concurrency_slot_release/_acquire pair.
4. Resumption has three branches. Abort drops any retained slot without reacquiring; normal-resume-with-slot calls stop_waiting (zero cost); normal-resume-without-slot blocks in acquire_slot, which saves/restores resume_status so a granted THREAD_LOCK_RESUMED is not clobbered.
5. Distribution places by deficit and drops nothing. Cores sort by get_score (-available + 2*waiters), slots go round-robin to the neediest, slots.empty () is asserted; a slot’s holder may cross cores while its owner stays fixed for routing home.
6. Core count is cgroup-aware. tune_parameters clamps the tunables into [system_core_count(), CSS_MAX_CLIENT_COUNT] with worker >= concurrency; system_core_count is adjusted_max — the floored intersection of affinity, online, and cgroup CPU quota — so container limits are honored.

Position hints as of this revision

The following are line numbers as observed on 2026-07-08; symbols are the canonical anchor and line numbers are hints that decay.

Symbol	File	Line
cpu::effective	PR:src/base/resources.cpp	192
PRM_ID_MAX_REQUEST_WORKER	PR:src/base/system_parameter.c	5212
PRM_ID_MAX_REQUEST_CONCURRENCY	PR:src/base/system_parameter.c	5231
net_server_wakeup_workers	PR:src/communication/network_sr.c	928
css_set_max_concurrency_and_workers	PR:src/connection/server_support.c	2913
read_data_from_java	PR:src/sp/pl_execution_stack_context.hpp	154
concurrency_slot_subscriber::activate	PR:src/thread/concurrency_slot.cpp	53
concurrency_slot_publisher::subscribe	PR:src/thread/concurrency_slot.cpp	82
concurrency_slot::return_to_pool	PR:src/thread/concurrency_slot.cpp	162
concurrency_slot::return_to_pool	PR:src/thread/concurrency_slot.cpp	163
concurrency_slot::start_waiting	PR:src/thread/concurrency_slot.cpp	179
concurrency_slot::stop_waiting	PR:src/thread/concurrency_slot.cpp	186
concurrency_slot::has_wait_expired	PR:src/thread/concurrency_slot.cpp	192
concurrency_slot::has_wait_expired	PR:src/thread/concurrency_slot.cpp	193
concurrency_slot_pool::concurrency_slot_pool	PR:src/thread/concurrency_slot.cpp	204
concurrency_slot_pool::adjust_concurrency	PR:src/thread/concurrency_slot.cpp	240
concurrency_slot_pool::adjust_concurrency(ulock)	PR:src/thread/concurrency_slot.cpp	248
concurrency_slot_pool::try_acquire_slot	PR:src/thread/concurrency_slot.cpp	291
concurrency_slot_pool::try_acquire_slot(ulock)	PR:src/thread/concurrency_slot.cpp	299
concurrency_slot_pool::acquire_slot	PR:src/thread/concurrency_slot.cpp	321
concurrency_slot_pool::acquire_slot	PR:src/thread/concurrency_slot.cpp	328
concurrency_slot_pool::acquire_slot(ulock)	PR:src/thread/concurrency_slot.cpp	329
concurrency_slot_pool::release_slot	PR:src/thread/concurrency_slot.cpp	413
concurrency_slot_pool::release_slot(ulock)	PR:src/thread/concurrency_slot.cpp	421
concurrency_slot_pool::return_slot	PR:src/thread/concurrency_slot.cpp	481
concurrency_slot_pool::borrow_surplus_slots	PR:src/thread/concurrency_slot.cpp	500
concurrency_slot_pool::borrow_surplus_slots	PR:src/thread/concurrency_slot.cpp	501
concurrency_slot_pool::needs_slot	PR:src/thread/concurrency_slot.cpp	525
concurrency_slot_pool::needs_slot(ulock)	PR:src/thread/concurrency_slot.cpp	533
concurrency_slot_pool::available_slots	PR:src/thread/concurrency_slot.cpp	541
concurrency_slot_pool::get_score	PR:src/thread/concurrency_slot.cpp	552
concurrency_slot_pool::check_surplus_slots	PR:src/thread/concurrency_slot.cpp	568
concurrency_slot_pool::wakeup_workers	PR:src/thread/concurrency_slot.cpp	585
concurrency_slot_pool::has_queued_task	PR:src/thread/concurrency_slot.cpp	600
concurrency_slot_daemon_task::execute	PR:src/thread/concurrency_slot.cpp	661
concurrency_slot_daemon_task::has_slot_demand	PR:src/thread/concurrency_slot.cpp	697
concurrency_slot_daemon_task::steal_from_entries_if_excess	PR:src/thread/concurrency_slot.cpp	711
concurrency_slot_daemon_task::steal_from_cores_if_excess	PR:src/thread/concurrency_slot.cpp	745
concurrency_slot_daemon_task::distribute_slots	PR:src/thread/concurrency_slot.cpp	762
concurrency_slot_daemon_task::tune_parameters	PR:src/thread/concurrency_slot.cpp	822
concurrency_slot_daemon_task::check_and_propagate_parameters	PR:src/thread/concurrency_slot.cpp	837
concurrency_slot_daemon::create_daemon	PR:src/thread/concurrency_slot.cpp	900
concurrency_slot_subscriber	PR:src/thread/concurrency_slot.hpp	52
concurrency_slot_subscriber::m_publisher	PR:src/thread/concurrency_slot.hpp	62
concurrency_slot_subscriber::m_identifier	PR:src/thread/concurrency_slot.hpp	63
concurrency_slot_publisher	PR:src/thread/concurrency_slot.hpp	66
concurrency_slot_publisher::m_subscribers	PR:src/thread/concurrency_slot.hpp	81
concurrency_slot_publisher::m_mutex	PR:src/thread/concurrency_slot.hpp	82
concurrency_slot	PR:src/thread/concurrency_slot.hpp	92
concurrency_slot::m_owner_pool	PR:src/thread/concurrency_slot.hpp	116
concurrency_slot::m_holder_pool	PR:src/thread/concurrency_slot.hpp	117
concurrency_slot::m_wait	PR:src/thread/concurrency_slot.hpp	119
concurrency_slot::m_wait_since	PR:src/thread/concurrency_slot.hpp	120
concurrency_slot_pool	PR:src/thread/concurrency_slot.hpp	125
concurrency_slot_pool::SLOT_SURPLUS_THRESHOLD	PR:src/thread/concurrency_slot.hpp	163
SLOT_SURPLUS_THRESHOLD	PR:src/thread/concurrency_slot.hpp	163
concurrency_slot_pool::m_parent	PR:src/thread/concurrency_slot.hpp	165
concurrency_slot_pool::m_available_slots	PR:src/thread/concurrency_slot.hpp	167
concurrency_slot_pool::m_surplus	PR:src/thread/concurrency_slot.hpp	169
concurrency_slot_pool::m_surplus_since	PR:src/thread/concurrency_slot.hpp	170
concurrency_slot_pool::m_slot_count	PR:src/thread/concurrency_slot.hpp	172
concurrency_slot_pool::m_target_count	PR:src/thread/concurrency_slot.hpp	173
concurrency_slot_pool::m_wait_queue	PR:src/thread/concurrency_slot.hpp	175
concurrency_slot_pool::m_mutex	PR:src/thread/concurrency_slot.hpp	178
concurrency_slot_daemon	PR:src/thread/concurrency_slot.hpp	190
concurrency_slot_daemon::m_daemon	PR:src/thread/concurrency_slot.hpp	210
concurrency_slot_publisher::traverse	PR:src/thread/concurrency_slot.hpp	217
concurrency_slot_publisher::traverse	PR:src/thread/concurrency_slot.hpp	218
concurrency_slot_daemon::traverse_subscribers	PR:src/thread/concurrency_slot.hpp	229
entry::start_waiting	PR:src/thread/thread_entry.cpp	472
entry::stop_waiting	PR:src/thread/thread_entry.cpp	478
thread_suspend_wakeup_and_unlock_entry	PR:src/thread/thread_entry.cpp	525
thread_suspend_timeout_wakeup_and_unlock_entry	PR:src/thread/thread_entry.cpp	553
thread_prepare_suspension	PR:src/thread/thread_entry.cpp	598
thread_prepare_resumption	PR:src/thread/thread_entry.cpp	660
event_stat::slot_waits	PR:src/thread/thread_entry.hpp	115
thread_resume_suspend_status	PR:src/thread/thread_entry.hpp	146
THREAD_CONCURRENCY_SLOT_SUSPENDED	PR:src/thread/thread_entry.hpp	173
THREAD_CONCURRENCY_SLOT_RESUMED	PR:src/thread/thread_entry.hpp	174
THREAD_SLEEP_FUNC_SUSPENDED	PR:src/thread/thread_entry.hpp	175
cubthread::entry::resume_status	PR:src/thread/thread_entry.hpp	258
cubthread::entry::m_slot	PR:src/thread/thread_entry.hpp	333
entry::m_slot	PR:src/thread/thread_entry.hpp	333
thread_concurrency_slot_release	PR:src/thread/thread_manager.hpp	630
thread_concurrency_slot_acquire	PR:src/thread/thread_manager.hpp	657
worker_pool_elastic	PR:src/thread/thread_worker_pool_elastic.hpp	54
worker_pool_elastic::unique_slot	PR:src/thread/thread_worker_pool_elastic.hpp	65
worker_pool_elastic::m_current_worker	PR:src/thread/thread_worker_pool_elastic.hpp	95
worker_pool_elastic::m_max_concurrency	PR:src/thread/thread_worker_pool_elastic.hpp	97
worker_pool_elastic::m_max_worker	PR:src/thread/thread_worker_pool_elastic.hpp	98
core_elastic	PR:src/thread/thread_worker_pool_elastic.hpp	107
core_elastic::m_slots	PR:src/thread/thread_worker_pool_elastic.hpp	155
core_elastic::m_max_concurrency	PR:src/thread/thread_worker_pool_elastic.hpp	158
core_elastic::m_current_worker	PR:src/thread/thread_worker_pool_elastic.hpp	161
core_elastic::m_max_worker	PR:src/thread/thread_worker_pool_elastic.hpp	162
core_elastic::m_retired_stats	PR:src/thread/thread_worker_pool_elastic.hpp	164
worker_elastic	PR:src/thread/thread_worker_pool_elastic.hpp	173
worker_elastic::m_slot	PR:src/thread/thread_worker_pool_elastic.hpp	197
worker_pool_elastic::worker_pool_elastic	PR:src/thread/thread_worker_pool_elastic.hpp	209
worker_pool_elastic::adjust_runtime_parameter	PR:src/thread/thread_worker_pool_elastic.hpp	239
core_elastic::core_elastic	PR:src/thread/thread_worker_pool_elastic.hpp	319
worker_pool_elastic::core_elastic::execute_task	PR:src/thread/thread_worker_pool_elastic.hpp	403
worker_pool_elastic::core_elastic::get_task_and_slot_or_become_available	PR:src/thread/thread_worker_pool_elastic.hpp	484
worker_pool_elastic::core_elastic::get_retire_if_excess	PR:src/thread/thread_worker_pool_elastic.hpp	509
core_elastic::reserve_available_worker	PR:src/thread/thread_worker_pool_elastic.hpp	578
worker_pool_elastic::core_elastic::reserve_available_worker	PR:src/thread/thread_worker_pool_elastic.hpp	579
worker_pool_elastic::core_elastic::release_available_worker	PR:src/thread/thread_worker_pool_elastic.hpp	599
worker_pool_elastic::core_elastic::get_or_make_available_worker	PR:src/thread/thread_worker_pool_elastic.hpp	607
worker_pool_elastic::core_elastic::try_execute_task_with_slot	PR:src/thread/thread_worker_pool_elastic.hpp	624
worker_elastic::get_new_task	PR:src/thread/thread_worker_pool_elastic.hpp	703
worker_elastic::run	PR:src/thread/thread_worker_pool_elastic.hpp	789
worker_elastic::execute_current_task	PR:src/thread/thread_worker_pool_elastic.hpp	799
system_core_count	PR:src/thread/thread_worker_pool_impl.cpp	39
worker_pool_impl::get_pool_size	PR:src/thread/thread_worker_pool_impl.hpp	691
core_impl::has_workers_snapshot_readers	PR:src/thread/thread_worker_pool_impl.hpp	1236
core_impl::get_available_worker	PR:src/thread/thread_worker_pool_impl.hpp	1244
worker_impl::run	PR:src/thread/thread_worker_pool_impl.hpp	1523
lock_suspend	PR:src/transaction/lock_manager.c	2270
epoll::epoll	src/base/epoll.cpp	37
epoll::wait	src/base/epoll.cpp	54
epoll::add_descriptor	src/base/epoll.cpp	59
TIMEOUT_INFINITE	src/base/epoll.hpp	33
TIMEOUT_NOWAIT	src/base/epoll.hpp	37
cubsocket::epoll	src/base/epoll.hpp	42
os::resources::cpu::context	src/base/resources.hpp	45
PRM_NAME_CSS_RECV_BUDGET_PER_CONNECTION	src/base/system_parameter.c	791
PRM_ID_CSS_AUTO_SCALING_WINDOW_SIZE	src/base/system_parameter.c	5243
PRM_ID_CSS_RECV_BUDGET_PER_CONNECTION	src/base/system_parameter.c	5259
PRM_ID_CSS_SEND_BUDGET_PER_CONNECTION	src/base/system_parameter.c	5271
net_server_start	src/communication/network_sr.c	1058
css_initialize_server_interfaces	src/communication/network_sr.c	1137
result	src/connection/buffer.hpp	43
context::context	src/connection/connection_context.cpp	71
context::context(capacity)	src/connection/connection_context.cpp	73
context::context()	src/connection/connection_context.cpp	95
context::reset	src/connection/connection_context.cpp	119
context::prepare	src/connection/connection_context.cpp	121
context::reset	src/connection/connection_context.cpp	126
thread_watcher	src/connection/connection_context.hpp	38
message_blocker	src/connection/connection_context.hpp	45
master::context	src/connection/connection_context.hpp	80
connection::state	src/connection/connection_context.hpp	128
connection::state	src/connection/connection_context.hpp	129
ignore_level	src/connection/connection_context.hpp	135
connection::ignore_level	src/connection/connection_context.hpp	136
connection::context	src/connection/connection_context.hpp	141
connection::context	src/connection/connection_context.hpp	142
context::m_id	src/connection/connection_context.hpp	152
context::m_recv	src/connection/connection_context.hpp	161
context::m_send	src/connection/connection_context.hpp	177
context::m_stats	src/connection/connection_context.hpp	189
context capacity ctor	src/connection/connection_context.hpp	191
DEFAULT_HEADER_DATA	src/connection/connection_defs.h	379
NET_HEADER	src/connection/connection_defs.h	391
in_method	src/connection/connection_defs.h	444
idx	src/connection/connection_defs.h	451
pending_request_count	src/connection/connection_defs.h	515
pool::initialize	src/connection/connection_pool.cpp	62
pool::finalize	src/connection/connection_pool.cpp	89
pool::claim_context	src/connection/connection_pool.cpp	140
pool::claim_context	src/connection/connection_pool.cpp	151
pool::retire_context	src/connection/connection_pool.cpp	160
pool::retire_context	src/connection/connection_pool.cpp	177
pool::try_to_lock_resource	src/connection/connection_pool.cpp	187
pool::initialize_freelist	src/connection/connection_pool.cpp	213
pool::finalize_freelist	src/connection/connection_pool.cpp	231
pool::initialize_freelist	src/connection/connection_pool.cpp	246
pool::initialize_topology	src/connection/connection_pool.cpp	249
pool::initialize_workers	src/connection/connection_pool.cpp	269
pool::finalize_freelist	src/connection/connection_pool.cpp	272
pool::finalize_workers	src/connection/connection_pool.cpp	314
pool::initialize_coordinator	src/connection/connection_pool.cpp	353
pool::start_coordinator	src/connection/connection_pool.cpp	376
pool::finalize_coordinator	src/connection/connection_pool.cpp	388
pool	src/connection/connection_pool.hpp	39
pool::freelist	src/connection/connection_pool.hpp	42
pool::claim_context	src/connection/connection_pool.hpp	70
pool::m_mutex	src/connection/connection_pool.hpp	78
pool::m_freelist	src/connection/connection_pool.hpp	96
pool::m_freelist	src/connection/connection_pool.hpp	101
css_wakeup_handler	src/connection/connection_sr.c	3061
context	src/connection/connection_statistics.hpp	32
statistics::context	src/connection/connection_statistics.hpp	32
context (enum)	src/connection/connection_statistics.hpp	32
worker	src/connection/connection_statistics.hpp	60
worker (enum)	src/connection/connection_statistics.hpp	60
statistics::worker::MQ_COMPLETED	src/connection/connection_statistics.hpp	73
statistics::worker::BLOCKED_RMUTEX	src/connection/connection_statistics.hpp	84
worker_to_string	src/connection/connection_statistics.hpp	93
metrics	src/connection/connection_statistics.hpp	111
metrics	src/connection/connection_statistics.hpp	112
metrics::operator-	src/connection/connection_statistics.hpp	230
metrics::add	src/connection/connection_statistics.hpp	264
css_conn_entry::add_pending_request	src/connection/connection_support.cpp	2803
css_conn_entry::start_request	src/connection/connection_support.cpp	2809
css_conn_entry::add_working_task	src/connection/connection_support.cpp	2827
css_conn_entry::end_working_task	src/connection/connection_support.cpp	2833
REGISTER_CONNECTION connection_worker	src/connection/connection_worker.cpp	70
worker::worker	src/connection/connection_worker.cpp	75
m_recv_budget	src/connection/connection_worker.cpp	92
m_send_budget	src/connection/connection_worker.cpp	93
statistics timer register	src/connection/connection_worker.cpp	126
worker::enqueue	src/connection/connection_worker.cpp	160
worker::notify	src/connection/connection_worker.cpp	182
worker::enqueue_and_notify	src/connection/connection_worker.cpp	218
worker::push_task_into_worker_pool	src/connection/connection_worker.cpp	288
worker::purge_stale_contexts	src/connection/connection_worker.cpp	294
worker::wakeup_blocked_worker	src/connection/connection_worker.cpp	320
worker::is_wait_required	src/connection/connection_worker.cpp	330
worker::requires_client_info	src/connection/connection_worker.cpp	330
worker::has_remaining_tasks	src/connection/connection_worker.cpp	340
worker::is_registering_client	src/connection/connection_worker.cpp	340
worker::has_remaining_tasks	src/connection/connection_worker.cpp	345
worker::start_connection_close	src/connection/connection_worker.cpp	360
worker::end_connection_close	src/connection/connection_worker.cpp	380
worker::handle_connection_close	src/connection/connection_worker.cpp	386
worker::retry_connection_close	src/connection/connection_worker.cpp	391
worker::handle_connection_close	src/connection/connection_worker.cpp	419
m_stats.sub CLIENT_NUM	src/connection/connection_worker.cpp	533
statistics_metrics_to_coordinator	src/connection/connection_worker.cpp	562
worker::eventfd_register	src/connection/connection_worker.cpp	632
worker::eventfd_register	src/connection/connection_worker.cpp	650
worker::eventfd_clear	src/connection/connection_worker.cpp	658
worker::eventfd_starttimer	src/connection/connection_worker.cpp	724
worker::eventfd_addtimer	src/connection/connection_worker.cpp	753
worker::eventfd_removetimer	src/connection/connection_worker.cpp	785
worker::eventfd_handler	src/connection/connection_worker.cpp	817
worker::handle_message_queue_send_packet	src/connection/connection_worker.cpp	880
worker::handle_message_queue_send_packet	src/connection/connection_worker.cpp	929
worker::handle_message_queue_release_packet	src/connection/connection_worker.cpp	980
worker::handle_message_queue_new_client	src/connection/connection_worker.cpp	1016
worker::handle_message_queue_release_packet	src/connection/connection_worker.cpp	1048
ctx->m_stats.set OPEND_NS	src/connection/connection_worker.cpp	1055
worker::handle_message_queue_new_client	src/connection/connection_worker.cpp	1101
worker::handle_message_queue_by_index	src/connection/connection_worker.cpp	1297
worker::handle_message_queue_shutdown_client	src/connection/connection_worker.cpp	1312
worker::handle_message_queue	src/connection/connection_worker.cpp	1356
worker::handle_message_queue	src/connection/connection_worker.cpp	1462
worker::handle_data_packet	src/connection/connection_worker.cpp	1535
worker::handle_command_header_packet	src/connection/connection_worker.cpp	1545
worker::handle_reception	src/connection/connection_worker.cpp	1694
m_stats.add PACKET_COUNT	src/connection/connection_worker.cpp	1731
m_stats.add BLOCKED_RMUTEX	src/connection/connection_worker.cpp	1743
worker::handle_transmission	src/connection/connection_worker.cpp	1782
worker::handle_exhausted_add_context	src/connection/connection_worker.cpp	1837
worker::handle_exhausted	src/connection/connection_worker.cpp	1854
worker::initialize	src/connection/connection_worker.cpp	1943
worker::finalize	src/connection/connection_worker.cpp	1975
worker::run	src/connection/connection_worker.cpp	2007
worker::attach	src/connection/connection_worker.cpp	2107
worker	src/connection/connection_worker.hpp	52
worker::status	src/connection/connection_worker.hpp	55
worker::timer_type	src/connection/connection_worker.hpp	63
worker::timer_latency	src/connection/connection_worker.hpp	74
worker::timer_handle	src/connection/connection_worker.hpp	82
worker::exhausted_context	src/connection/connection_worker.hpp	90
exhausted_context	src/connection/connection_worker.hpp	90
worker::queue_type	src/connection/connection_worker.hpp	98
queue_type	src/connection/connection_worker.hpp	98
worker::message_type	src/connection/connection_worker.hpp	106
message_type	src/connection/connection_worker.hpp	106
worker::message	src/connection/connection_worker.hpp	130
worker::m_parent	src/connection/connection_worker.hpp	207
m_context	src/connection/connection_worker.hpp	221
worker::m_timer_handler	src/connection/connection_worker.hpp	230
worker::m_queue	src/connection/connection_worker.hpp	237
m_queue	src/connection/connection_worker.hpp	237
worker::m_queue_size	src/connection/connection_worker.hpp	241
m_queue_size	src/connection/connection_worker.hpp	241
m_removed_context	src/connection/connection_worker.hpp	246
worker::m_exhausted	src/connection/connection_worker.hpp	248
m_exhausted	src/connection/connection_worker.hpp	248
m_exhausted	src/connection/connection_worker.hpp	251
worker::m_stats	src/connection/connection_worker.hpp	251
controller	src/connection/controller.hpp	43
controller::recv	src/connection/controller.hpp	170
controller::send	src/connection/controller.hpp	193
EWMA_ALPHA	src/connection/coordinator.cpp	41
VAL_TO_SCORE	src/connection/coordinator.cpp	43
EVAL_WORKER	src/connection/coordinator.cpp	44
EVAL_CONTEXT	src/connection/coordinator.cpp	45
REGISTER_CONNECTION coordinator	src/connection/coordinator.cpp	55
coordinator::coordinator	src/connection/coordinator.cpp	57
coordinator::random_bit	src/connection/coordinator.cpp	229
coordinator::transfer_connection	src/connection/coordinator.cpp	237
coordinator::scale_up	src/connection/coordinator.cpp	281
coordinator::scale_down_finish	src/connection/coordinator.cpp	317
coordinator::scale_down	src/connection/coordinator.cpp	348
coordinator::scale_trial	src/connection/coordinator.cpp	378
coordinator::scale_selection	src/connection/coordinator.cpp	415
coordinator::statistics_EWMA	src/connection/coordinator.cpp	447
statistics_EWMA	src/connection/coordinator.cpp	447
coordinator::statistics_find_score_extremes	src/connection/coordinator.cpp	460
coordinator::statistics_update_score	src/connection/coordinator.cpp	482
statistics_update_score	src/connection/coordinator.cpp	482
coordinator::statistics_update_connection	src/connection/coordinator.cpp	502
statistics_update_connection	src/connection/coordinator.cpp	502
coordinator::statistics_update_task	src/connection/coordinator.cpp	545
statistics_update_task	src/connection/coordinator.cpp	545
coordinator::statistics_update	src/connection/coordinator.cpp	578
coordinator::statistics_rebalancing	src/connection/coordinator.cpp	586
coordinator::statistics_scaling	src/connection/coordinator.cpp	629
statistics_scaling	src/connection/coordinator.cpp	629
statistics_print	src/connection/coordinator.cpp	697
coordinator::eventfd_register	src/connection/coordinator.cpp	758
coordinator::eventfd_addtimer	src/connection/coordinator.cpp	865
coordinator::handle_message_queue_start	src/connection/coordinator.cpp	929
coordinator::handle_message_queue_new_client	src/connection/coordinator.cpp	934
coordinator::handle_message_queue_new_client	src/connection/coordinator.cpp	944
coordinator::handle_message_queue_return_to_pool	src/connection/coordinator.cpp	991
coordinator::handle_message_queue_statistics	src/connection/coordinator.cpp	1032
handle_message_queue_statistics	src/connection/coordinator.cpp	1032
handle_controller_request SHOW_STATS	src/connection/coordinator.cpp	1133
coordinator::initialize	src/connection/coordinator.cpp	1192
coordinator::finalize	src/connection/coordinator.cpp	1225
coordinator::run	src/connection/coordinator.cpp	1240
coordinator::run (m_timens)	src/connection/coordinator.cpp	1260
coordinator::run (timerfd dispatch)	src/connection/coordinator.cpp	1279
coordinator::attach	src/connection/coordinator.cpp	1314
coordinator	src/connection/coordinator.hpp	43
status	src/connection/coordinator.hpp	46
worker_statistics	src/connection/coordinator.hpp	52
worker_statistics	src/connection/coordinator.hpp	54
scaling_status	src/connection/coordinator.hpp	75
scaling_status	src/connection/coordinator.hpp	77
scaling_direction	src/connection/coordinator.hpp	81
scaling_direction	src/connection/coordinator.hpp	83
scaling_statistics	src/connection/coordinator.hpp	89
timer_latency	src/connection/coordinator.hpp	95
timer_type	src/connection/coordinator.hpp	103
timer_handle	src/connection/coordinator.hpp	113
control_type	src/connection/coordinator.hpp	123
control_recv	src/connection/coordinator.hpp	140
control_send	src/connection/coordinator.hpp	148
message_type	src/connection/coordinator.hpp	155
coordinator::message	src/connection/coordinator.hpp	167
message	src/connection/coordinator.hpp	171
m_timer_handler	src/connection/coordinator.hpp	245
m_controller	src/connection/coordinator.hpp	249
m_queue	src/connection/coordinator.hpp	256
m_queue_size	src/connection/coordinator.hpp	260
m_scaling	src/connection/coordinator.hpp	261
m_current_worker	src/connection/coordinator.hpp	265
m_migrating	src/connection/coordinator.hpp	268
m_scaling_statistics	src/connection/coordinator.hpp	269
m_scaling	src/connection/coordinator.hpp	271
m_scaling_statistics	src/connection/coordinator.hpp	279
m_task_statistics	src/connection/coordinator.hpp	284
m_task_statistics	src/connection/coordinator.hpp	294
m_statistics	src/connection/coordinator.hpp	294
m_statistics	src/connection/coordinator.hpp	304
statistics_print (decl)	src/connection/coordinator.hpp	335
statistics_EWMA	src/connection/coordinator.hpp	386
receiver::receive_in_allocated	src/connection/receiver.cpp	311
receiver BYTES_IN_TOTAL add	src/connection/receiver.cpp	330
receiver::drain	src/connection/receiver.cpp	430
RECV_BUDGET_HIT	src/connection/receiver.cpp	475
receiver RECV_BUDGET_HIT add	src/connection/receiver.cpp	475
receiver	src/connection/receiver.hpp	38
receiver::m_stats	src/connection/receiver.hpp	64
css_server_task	src/connection/server_support.c	144
css_job_queues_start_scan	src/connection/server_support.c	251
REGISTER_WORKERPOOL	src/connection/server_support.c	550
css_init	src/connection/server_support.c	554
thread_create_stats_worker_pool	src/connection/server_support.c	581
connections.initialize	src/connection/server_support.c	595
connections.finalize	src/connection/server_support.c	611
css_push_server_task	src/connection/server_support.c	2354
css_server_task::execute	src/connection/server_support.c	2376
css_get_thread_stats	src/connection/server_support.c	2636
css_get_task_stats	src/connection/server_support.c	2647
css_wp_worker_get_busy_count_mapper	src/connection/server_support.c	2669
css_wp_core_job_scan_mapper	src/connection/server_support.c	2695
transmitter::fill	src/connection/transmitter.cpp	66
SEND_BUDGET_HIT	src/connection/transmitter.cpp	78
transmitter SEND_BUDGET_HIT add	src/connection/transmitter.cpp	78
transmitter BYTES_OUT_TOTAL add	src/connection/transmitter.cpp	86
transmitter	src/connection/transmitter.hpp	38
main	src/executables/server.c	276
metadata_of_job_queues	src/parser/show_meta.c	528
entry_task	src/thread/thread_entry_task.hpp	56
entry_manager::create_context	src/thread/thread_entry_task.hpp	100
manager::alloc_entries	src/thread/thread_manager.cpp	82
manager::init_lockfree_system	src/thread/thread_manager.cpp	115
manager::push_task_on_core	src/thread/thread_manager.cpp	180
manager::claim_entry	src/thread/thread_manager.cpp	234
manager::set_max_thread_count_from_config	src/thread/thread_manager.cpp	266
cubthread::initialize	src/thread/thread_manager.cpp	315
initialize_thread_entries	src/thread/thread_manager.cpp	378
manager::create_worker_pool	src/thread/thread_manager.hpp	367
manager::create_and_track_resource	src/thread/thread_manager.hpp	424
manager::destroy_and_untrack_resource	src/thread/thread_manager.hpp	445
task::retire	src/thread/thread_task.hpp	75
condvar_wait	src/thread/thread_waiter.hpp	164
worker_pool	src/thread/thread_worker_pool.hpp	54
get_idle_timeout	src/thread/thread_worker_pool.hpp	91
worker_pool::core	src/thread/thread_worker_pool.hpp	123
worker_pool::core::worker	src/thread/thread_worker_pool.hpp	178
system_core_count	src/thread/thread_worker_pool_impl.cpp	39
worker_pool_impl	src/thread/thread_worker_pool_impl.hpp	106
m_cores	src/thread/thread_worker_pool_impl.hpp	202
m_max_workers	src/thread/thread_worker_pool_impl.hpp	205
m_stopped	src/thread/thread_worker_pool_impl.hpp	208
m_round_robin_counter	src/thread/thread_worker_pool_impl.hpp	211
stats_base	src/thread/thread_worker_pool_impl.hpp	220
stats_base_true	src/thread/thread_worker_pool_impl.hpp	234
stats_base::statset	src/thread/thread_worker_pool_impl.hpp	248
stats_base::time	src/thread/thread_worker_pool_impl.hpp	250
core_impl	src/thread/thread_worker_pool_impl.hpp	263
m_workers	src/thread/thread_worker_pool_impl.hpp	319
m_available_workers	src/thread/thread_worker_pool_impl.hpp	320
m_task_queue	src/thread/thread_worker_pool_impl.hpp	321
m_workers_mutex	src/thread/thread_worker_pool_impl.hpp	323
m_temp_workers	src/thread/thread_worker_pool_impl.hpp	326
m_free_temp_workers	src/thread/thread_worker_pool_impl.hpp	327
m_temp_workers_mutex	src/thread/thread_worker_pool_impl.hpp	329
worker_impl	src/thread/thread_worker_pool_impl.hpp	339
worker_impl::assign_task	src/thread/thread_worker_pool_impl.hpp	354
worker_impl::assign_task_void	src/thread/thread_worker_pool_impl.hpp	356
m_context_p	src/thread/thread_worker_pool_impl.hpp	395
worker_impl (fields)	src/thread/thread_worker_pool_impl.hpp	395
m_wrapped_task	src/thread/thread_worker_pool_impl.hpp	396
m_task_cv	src/thread/thread_worker_pool_impl.hpp	399
m_task_mutex	src/thread/thread_worker_pool_impl.hpp	401
m_stop	src/thread/thread_worker_pool_impl.hpp	403
m_has_thread	src/thread/thread_worker_pool_impl.hpp	404
m_is_temp	src/thread/thread_worker_pool_impl.hpp	406
m_stats	src/thread/thread_worker_pool_impl.hpp	408
wrapped_task	src/thread/thread_worker_pool_impl.hpp	417
wrapped_task::task_only	src/thread/thread_worker_pool_impl.hpp	420
wrapped_task::task_with_stats	src/thread/thread_worker_pool_impl.hpp	425
wrapped_task::inner_type	src/thread/thread_worker_pool_impl.hpp	434
wrapped_task::m_inner	src/thread/thread_worker_pool_impl.hpp	453
stats	src/thread/thread_worker_pool_impl.hpp	461
stats::id	src/thread/thread_worker_pool_impl.hpp	464
stats::statdef	src/thread/thread_worker_pool_impl.hpp	479
worker_pool_impl::worker_pool_impl	src/thread/thread_worker_pool_impl.hpp	525
worker_pool_impl::initialize	src/thread/thread_worker_pool_impl.hpp	552
worker_pool_impl::execute	src/thread/thread_worker_pool_impl.hpp	558
worker_pool_impl::execute_on_core	src/thread/thread_worker_pool_impl.hpp	565
worker_pool_impl::execute_on_core	src/thread/thread_worker_pool_impl.hpp	567
worker_pool_impl::is_running	src/thread/thread_worker_pool_impl.hpp	577
worker_pool_impl::stop_execution	src/thread/thread_worker_pool_impl.hpp	592
worker_pool_impl::stop_execution	src/thread/thread_worker_pool_impl.hpp	594
worker_pool_impl::get_stats	src/thread/thread_worker_pool_impl.hpp	670
worker_pool_impl::assign_workers_to_cores	src/thread/thread_worker_pool_impl.hpp	758
worker_pool_impl::get_next_core	src/thread/thread_worker_pool_impl.hpp	780
worker_pool_impl::get_round_robin_core_hash	src/thread/thread_worker_pool_impl.hpp	787
core_impl::execute_task	src/thread/thread_worker_pool_impl.hpp	894
worker_pool_impl::core_impl::execute_task	src/thread/thread_worker_pool_impl.hpp	896
core_impl::warmup	src/thread/thread_worker_pool_impl.hpp	946
core_impl::stop_execution	src/thread/thread_worker_pool_impl.hpp	969
core_impl::retire_queued_tasks	src/thread/thread_worker_pool_impl.hpp	998
core_impl::get_task_or_become_available	src/thread/thread_worker_pool_impl.hpp	1010
core_impl::become_available	src/thread/thread_worker_pool_impl.hpp	1030
core_impl::become_available	src/thread/thread_worker_pool_impl.hpp	1032
core_impl::check_worker_not_available	src/thread/thread_worker_pool_impl.hpp	1040
core_impl::check_worker_not_available	src/thread/thread_worker_pool_impl.hpp	1042
core_impl::register_free_temp_list	src/thread/thread_worker_pool_impl.hpp	1063
core_impl::free_all_temp_list	src/thread/thread_worker_pool_impl.hpp	1080
core_impl::get_stats	src/thread/thread_worker_pool_impl.hpp	1089
core_impl::initialize_workers	src/thread/thread_worker_pool_impl.hpp	1170
core_impl::execute_task_as_temp	src/thread/thread_worker_pool_impl.hpp	1194
worker_pool_impl::core_impl::worker_impl::assign_task	src/thread/thread_worker_pool_impl.hpp	1267
worker_impl::assign_task	src/thread/thread_worker_pool_impl.hpp	1267
worker_impl::stop_execution	src/thread/thread_worker_pool_impl.hpp	1326
worker_impl::run	src/thread/thread_worker_pool_impl.hpp	1387
worker_impl::init_run	src/thread/thread_worker_pool_impl.hpp	1430
worker_impl::finish_run	src/thread/thread_worker_pool_impl.hpp	1457
worker_impl::execute_current_task	src/thread/thread_worker_pool_impl.hpp	1479
worker_impl::retire_current_task	src/thread/thread_worker_pool_impl.hpp	1507
worker_impl::get_new_task	src/thread/thread_worker_pool_impl.hpp	1521
stats::statdef	src/thread/thread_worker_pool_impl.hpp	1673
worker_pool_task_capper::worker_pool_task_capper	src/thread/thread_worker_pool_taskcap.cpp	34
worker_pool_task_capper::try_task	src/thread/thread_worker_pool_taskcap.cpp	44
worker_pool_task_capper::push_task	src/thread/thread_worker_pool_taskcap.cpp	60
worker_pool_task_capper::execute	src/thread/thread_worker_pool_taskcap.cpp	74
worker_pool_task_capper::end_task	src/thread/thread_worker_pool_taskcap.cpp	83
capped_task::~capped_task	src/thread/thread_worker_pool_taskcap.cpp	106
capped_task::execute	src/thread/thread_worker_pool_taskcap.cpp	112
worker_pool_task_capper	src/thread/thread_worker_pool_taskcap.hpp	30
worker_pool_task_capper::capped_task	src/thread/thread_worker_pool_taskcap.hpp	55
boot_restart_server initialize_thread_entries	src/transaction/boot_sr.c	1638

Sources

• cubrid-thread-manager.md — the high-level companion (motivation, the two-phase redesign narrative, JIRA/ticket map). See also cubrid-thread-worker-pool.md (the base cubthread engine) and cubrid-lock-manager-detail.md (the logical waits Phase 2 hooks into).
• Code (merged develop, Chapters 1–11): src/connection/{connection_worker,connection_pool,connection_context,coordinator,connection_statistics,controller,connection_support,server_support}.{cpp,hpp,c}, src/base/epoll.{cpp,hpp}, src/thread/{thread_worker_pool_impl,thread_worker_pool,thread_worker_pool_taskcap,thread_manager,thread_task,thread_entry_task}.{cpp,hpp}.
• Code (PR #7323 feature/worker_pool_elastic, unmerged, Chapters 12–14): src/thread/{concurrency_slot,thread_worker_pool_elastic,thread_entry}.{cpp,hpp}, src/communication/network_sr.c, src/transaction/lock_manager.c, src/sp/pl_execution_stack_context.hpp, src/base/cgroup.cpp.
• Methodology: knowledge/methodology/code-analysis-detail-doc.md.