Copilot commented on code in PR #13898:
URL: https://github.com/apache/trafficserver/pull/13898#discussion_r4447055747
##########
src/iocore/net/UnixNetAccept.cc:
##########
@@ -319,15 +442,31 @@ NetAccept::init_accept_per_thread()
NetAccept *a = (i < n - 1) ? clone() : this;
EThread *t = eventProcessor.thread_group[ET_NET]._thread[i];
a->mutex = get_NetHandler(t)->mutex;
+
+ a->uses_fast_accept = true;
t->schedule_imm(a);
}
+
+ return true;
}
void
NetAccept::stop_accept()
{
action_->cancel();
server.close();
+
+ // The retry timer dispatches into this acceptor, which the caller is about
to delete. Cancel it under
+ // the mutex that serializes acceptFastEvent. Per-thread clones are not
reachable from here; they notice the
+ // action is no longer listening and stand down on their own. Acceptors on a
dedicated
+ // accept thread never arm the timer, and their mutex is held by the accept
thread forever, so never wait on it.
+ if (uses_fast_accept && mutex != nullptr) {
+ SCOPED_MUTEX_LOCK(lock, mutex, this_ethread());
Review Comment:
This blocking NetHandler mutex acquisition can hang graceful shutdown when
its owning ET_NET thread is stuck in a plugin.
`SignalContinuation::state_running()` calls `stop_HttpProxyServer()` before
scheduling `AutoStopCont`, so the configured shutdown timeout is never
scheduled. Defer timer cancellation and acceptor reclamation to the owning
thread without blocking the signal continuation.
##########
src/iocore/net/UnixNetAccept.cc:
##########
@@ -375,9 +520,28 @@ NetAccept::do_blocking_accept(EThread *t)
int count = 0;
int additional_accepts = NetHandler::get_additional_accepts();
+ watchdog_apply_backlog(this);
+
+ if (watchdog_should_refuse(this)) {
+ // Dedicated accept thread - acceptLoopEvent calls straight back in, so
sleep rather than spin
+ // while the connections wait in the kernel accept queue.
+ Metrics::Counter::increment(net_rsb.watchdog_accepts_refused);
+ safe_delay(net_throttle_delay);
+ return 0;
Review Comment:
During SIGTERM, `UnixNetProcessor::stop_accept()` deletes the dedicated
acceptor even if its thread is sleeping here. On waking, returning 0 makes
`acceptLoopEvent()` call `do_blocking_accept()` again on the freed object. Keep
the acceptor alive until the worker observes cancellation and exits, with a
single owner responsible for deletion.
##########
include/iocore/eventsystem/WatchdogPolicy.h:
##########
@@ -0,0 +1,237 @@
+/** @file
+
+ The decision arithmetic of the watchdog's accept backpressure, free of
metrics, records and threads.
+
+ `Watchdog::Monitor` feeds each scan's peak awake span through advance() and
applies the result. Keeping
+ the arithmetic here lets it be exercised directly, without a running monitor.
+
+ @section license License
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <chrono>
+#include <cstdint>
+
+namespace Watchdog
+{
+
+/// Most aggressive release deadband the retain knob will produce. See
release_threshold().
+constexpr int32_t min_release_permille = 500;
+
+/// The configuration one scan is decided against.
+struct PolicyConfig {
+ /// Whether proxy.config.exec_thread.watchdog.mode is backpressure. Nothing
engages otherwise.
+ bool backpressure = false;
+ /// proxy.config.exec_thread.watchdog.timeout_shrink_ms
+ std::chrono::milliseconds shrink_timeout{0};
+ /// proxy.config.exec_thread.watchdog.timeout_refuse_ms; zero disables
refuse.
+ std::chrono::milliseconds refuse_timeout{0};
+ /// proxy.config.exec_thread.watchdog.backlog_decrease_permille
+ int decrease_permille = 500;
+ /// proxy.config.exec_thread.watchdog.backlog_increase_permille
+ int increase_permille = 100;
+ /// proxy.config.exec_thread.watchdog.backlog_min_permille
+ int min_permille = 50;
+};
+
+/// What carries from one scan to the next.
+struct PolicyState {
+ /// Whether the accept queue shrink is engaged. Latched through the release
deadband.
+ bool shrink_engaged = false;
+ /// Whether the accept paths are refusing. Latched through the release
deadband.
+ bool refuse = false;
+ /// Listen backlog scale in permille. 1000 is unthrottled.
+ int scale = 1000;
+ /** Backlog scale in force when the current shrink episode began, in
permille; 0 when none.
+
+ The mark that headroom-attenuated recovery decelerates into. Cleared
once the scale is back to 1000.
+ */
+ int trouble_scale = 0;
+
+ bool operator==(PolicyState const &) const = default;
+};
+
+/// The outcome of one scan.
+struct PolicyResult {
+ PolicyState state;
+ /// The peak reached the shrink threshold, whatever the mode.
+ bool saturated = false;
+ /// The peak reached the refuse threshold, whatever the mode. Never set
while that threshold is zero.
+ bool refuse_exceeded = false;
+};
+
+/** Peak awake span at or below which a scan counts as recovered, given an
action @a threshold.
+
+ Every action engages at its own threshold but releases only here, so the
two differ and the
+ band between them holds the current state - a Schmitt trigger rather than
a bare comparison.
+ Without it a peak one millisecond under the threshold reads as full
recovery when it is really
+ the boundary, which is exactly where the state chatters and where
releasing lets the queue climb
+ back into the next step of load.
+
+ Derived from the decrease knob rather than from one of its own because it
is the same statement:
+ retain 500 means halve the queue when saturated *and* wait for the peak to
halve before believing
+ the recovery. Aggressive shedding and late release belong together.
+
+ Floored at @c min_release_permille so the coupling cannot make
backpressure effectively
+ unreleasable - at retain 1 the threshold would be two milliseconds on a
two second timeout,
+ which a merely busy server may never reach. Half the threshold is already
far outside
+ measurement noise, so nothing is lost by refusing to go deeper; below the
floor the knob keeps
+ sharpening the shed and stops affecting the release.
+ */
+inline std::chrono::nanoseconds
+release_threshold(std::chrono::milliseconds threshold, int decrease_permille)
+{
+ int32_t const retain = std::max<int32_t>(min_release_permille,
decrease_permille);
+ // Scaled in nanoseconds so a small timeout keeps its fractional-millisecond
boundary.
+ return std::chrono::duration_cast<std::chrono::nanoseconds>(threshold) *
retain / 1000;
+}
+
+/** @return Permille points to add to the backlog scale on a recovered scan.
+
+ @a increase_permille of the scale currently in force, attenuated by the
headroom remaining to
+ @c state.trouble_scale, or unattenuated when there is no mark to measure
against or the scale is
+ already above it.
+ */
+inline int
+recovery_step_permille(PolicyState const &state, int increase_permille)
+{
+ int const rate = increase_permille;
+
+ // Proportional to the depth currently in force, so the step is a small
number of connections while
+ // the queue is deep in the hole and grows as it climbs out - the absolute
count admitted per scan
+ // is what re-saturates the threads, and a flat step in points of the
baseline is the same count at
+ // every depth. The cap restates the invariant rather than binding: the
scale is at most 1000, so
+ // the proportional value is at most the configured points, which it reaches
only at full depth.
+ // The floor keeps the step positive against integer truncation - below a
scale of 1000/rate the
+ // product truncates to zero, and a zero step would leave the queue stranded
at that depth, because
+ // the caller only republishes when the value actually changes.
+ int const step = std::max(1, std::min(state.scale * rate / 1000, rate));
+
+ // No mark yet - the first recovery of the process, or one that already
climbed back to full depth.
+ // Nothing to attenuate against, so the proportional step stands.
+ if (state.trouble_scale <= 0) {
+ return step;
+ }
+
+ // Past the mark already - it has been proven survivable since, so it
carries no information about
+ // the rest of the climb. Full rate the rest of the way to 1000.
+ if (state.scale >= state.trouble_scale) {
+ return step;
+ }
+
+ // Within one full step of the mark: land exactly on it rather than tapering
into it. The attenuated
+ // step goes to zero as the gap closes, and the one point floor above would
otherwise turn the last
+ // stretch into a crawl of single-point scans - dozens of them at a large
backlog - for no benefit,
+ // since the mark is a depth to sit at and reassess, not one to creep up on.
Never returns more than
+ // @a step, because the branch condition is exactly that the remaining gap
fits inside one, and @a
+ // step is itself bounded by the configured rate - so making the step
proportional cannot turn this
+ // into an unbounded leap onto the mark.
+ if (state.scale + step >= state.trouble_scale) {
+ return state.trouble_scale - state.scale;
+ }
+
+ // Attenuate by the fraction of the way still to go, so the climb
decelerates as it approaches the
+ // depth that last caused trouble instead of arriving there at full speed.
Far below the mark this is
+ // ~1 and costs nothing. The floor keeps the step positive against integer
truncation; the snap above
+ // is what stops it from being reached repeatedly.
+ int const headroom = state.trouble_scale - state.scale;
+ return std::max(1, step * headroom / state.trouble_scale);
+}
+
+/** Decide one scan.
+
+ Each action engages on the first scan at or above its own threshold and
releases only once the
+ peak has fallen to the deadband below it. Between the two thresholds the
current state holds. Shrink
+ and refuse each have their own threshold on the same peak and neither
depends on the other.
+
+ @param[in] config The configuration in force for this scan.
+ @param[in] state The state the previous scan left.
+ @param[in] peak The largest awake span across all watched threads since
the previous scan.
+ */
+inline PolicyResult
+advance(PolicyConfig const &config, PolicyState state,
std::chrono::steady_clock::duration peak)
+{
+ PolicyResult result;
+
+ result.saturated = peak >= config.shrink_timeout;
+ bool const recovered = peak <= release_threshold(config.shrink_timeout,
config.decrease_permille);
+
+ // Shrink stage state. Mode off releases now rather than waiting out the
deadband - the point of
+ // turning it off is to stop throttling immediately.
+ bool const was_shrinking = state.shrink_engaged;
+ if (!config.backpressure) {
+ state.shrink_engaged = false;
+ } else if (result.saturated) {
+ state.shrink_engaged = true;
+ } else if (recovered) {
+ state.shrink_engaged = false;
+ }
+
+ // The depth still in force as this episode begins is, by construction, a
depth that let in more
+ // than the threads could serve. Remember it as the mark that
headroom-attenuated recovery slows
+ // down against. Captured before the decrease below touches the scale.
+ if (state.shrink_engaged && !was_shrinking) {
+ state.trouble_scale = state.scale;
+ }
+
+ // Accept queue sizing. Multiplicative decrease on a saturated scan,
proportional increase once
+ // released, and held steady inside the release deadband - a peak that has
come down but not far
+ // enough is not evidence for shedding harder, and without the saturated
term a long stay in the
+ // deadband would squeeze the queue toward the floor while it was recovering.
+ if (config.backpressure) {
+ int want = state.scale;
+
+ if (state.shrink_engaged && result.saturated) {
+ // Integer truncation would stall the decrease entirely at small scales
- 3 permille retained
+ // at 900 rounds back to 3 - so always give up at least one point while
shrinking.
+ int const next = std::min(state.scale * config.decrease_permille / 1000,
state.scale - 1);
+ want = std::max(config.min_permille, next);
+ } else if (!state.shrink_engaged) {
+ want = std::min(1000, state.scale + recovery_step_permille(state,
config.increase_permille));
+ }
+ state.scale = want;
Review Comment:
The reloadable backlog floor is enforced only on a saturated shrink scan.
Raising `backlog_min_permille` from 50 to 500 while the peak remains in the
release deadband leaves the scale at 50 indefinitely, and recovery scans can
also publish values below the new floor. Clamp the resulting scale to the
current minimum on every backpressure scan.
##########
src/proxy/http/HttpProxyServerMain.cc:
##########
@@ -345,12 +345,95 @@ init_HttpProxyServer()
#endif
}
+namespace
+{
+
+/** Open a loopback-bound companion listener for @a port, exempt from watchdog
backpressure.
+ *
+ * Local connections arrive on the same port as all other traffic, so there is
no port to exempt -
+ * only a source address. But watchdog backpressure is a decision about a
+ * listen socket, taken before @c accept4, and a connection's peer address is
not knowable until after
+ * it. The way out is to let the kernel do the discrimination: a socket bound
explicitly to the
+ * loopback address takes precedence over the wildcard socket on the same
port, so local connections
+ * land on this listener and everything from off box lands on the original
one. Marking only this
+ * listener exempt then protects local connections without giving up the
backpressure for the flood.
+ *
+ * Opened whenever the watchdog is enabled at all, rather than when an active
mode is selected:
+ * `proxy.config.exec_thread.watchdog.mode` is reloadable, and binding or
unbinding sockets in
+ * response to a config change would be a great deal of risk for no benefit.
An exempt loopback
+ * listener behaves identically to the wildcard one when no backpressure is
engaged.
+ *
+ * @param[in] acceptor The acceptor whose continuation and options this
listener shares.
+ * @param[in] port The proxy port being duplicated on loopback.
+ */
+void
+start_loopback_companion_acceptor(HttpProxyAcceptor &acceptor, HttpProxyPort
&port)
+{
+ NetProcessor::AcceptOptions net_opt = acceptor._net_opt;
+
+ // localhost_only takes precedence over local_ip in accept_internal(), which
is what binds this to
+ // 127.0.0.1 or ::1 according to the port's address family.
+ net_opt.localhost_only = true;
+ net_opt.f_watchdog_exempt = true;
+
+ // Overlapping binds - this loopback socket and the port's wildcard socket -
require SO_REUSEPORT on
+ // both. The caller has already set it on the wildcard side.
+ net_opt.f_force_reuseport = true;
+
+ // Never take the proxy down over a supplementary listener. Without it the
exemption is simply not
+ // in effect, which the warning from Server::listen() reports.
+ net_opt.f_non_fatal_listen = true;
+
+ // Don't double the process's dedicated accept threads for a listener that
will see only a low
+ // volume of local connections. Left alone when accept threads are not in
use, so that the companion follows
+ // the same per-thread or SO_REUSEPORT model as the port it shadows.
+ int accept_threads = net_opt.accept_threads;
+ if (accept_threads < 0) {
+ accept_threads =
RecGetRecordInt("proxy.config.accept_threads").value_or(0);
+ }
+ if (accept_threads > 0) {
+ net_opt.accept_threads = 1;
+ }
Review Comment:
With dedicated accept threads, the companion dispatches local requests
through `eventProcessor.assign_thread(ET_NET)`, which selects workers
round-robin without checking whether they are stalled. A localhost health check
can therefore wait for the full stall while other event threads are idle. Use
the shared ET_NET accept path for companions even when the wildcard listener
uses dedicated accept threads.
##########
src/iocore/net/UnixNetAccept.cc:
##########
@@ -375,9 +520,28 @@ NetAccept::do_blocking_accept(EThread *t)
int count = 0;
int additional_accepts = NetHandler::get_additional_accepts();
+ watchdog_apply_backlog(this);
+
+ if (watchdog_should_refuse(this)) {
+ // Dedicated accept thread - acceptLoopEvent calls straight back in, so
sleep rather than spin
+ // while the connections wait in the kernel accept queue.
+ Metrics::Counter::increment(net_rsb.watchdog_accepts_refused);
+ safe_delay(net_throttle_delay);
Review Comment:
With dedicated accept threads, refusal cannot interrupt a blocking
`server.accept()`, so the next connection is still accepted and dispatched if
refusal engages while that call is waiting. This also breaks the new test's
unchanged `total_accepts` assertion for `accept_threads=1`. Use a nonblocking
listener with a bounded or explicitly wakeable readiness wait, and recheck
refusal before accepting.
##########
src/iocore/net/UnixNetAccept.cc:
##########
@@ -529,7 +693,52 @@ NetAccept::acceptFastEvent(int event, void *ep)
NetHandler *h = get_NetHandler(t);
int additional_accepts =
NetHandler::get_additional_accepts();
+ if (e == refuse_retry) {
+ refuse_retry = nullptr;
+ }
+
+ // Clones share action_ with the acceptor that stop_accept() closed, so this
is how they learn the listen fd is gone.
+ // It may already belong to another connection, so neither touch it nor arm
another retry.
+ if (!action_->is_listening()) {
+ if (refuse_retry != nullptr) {
+ refuse_retry->cancel(this);
+ refuse_retry = nullptr;
+ }
+ return EVENT_DONE;
+ }
+
+ watchdog_apply_backlog(this);
+
+ // See the equivalent comment in net_accept(). The listen fd here is
registered edge triggered, so
+ // refusing must come back on a timer: returning without draining would not
produce another
+ // readiness event for the connections already queued. New arrivals still
wake us while refusing,
+ // so only arm the timer if one is not already pending.
+ if (watchdog_should_refuse(this)) {
+ Metrics::Counter::increment(net_rsb.watchdog_accepts_refused);
+ if (refuse_retry == nullptr && action_->is_listening()) {
+ refuse_retry = t->schedule_in(this, HRTIME_MSECONDS(net_throttle_delay));
+ }
+ return EVENT_CONT;
+ }
+
+ // An arrival beat the retry timer to it, so this pass does the drain the
timer was armed for.
+ if (refuse_retry != nullptr) {
+ refuse_retry->cancel(this);
+ refuse_retry = nullptr;
+ }
+
do {
+ if (count > 0) {
+ watchdog_apply_backlog(this);
+ if (watchdog_should_refuse(this)) {
+ Metrics::Counter::increment(net_rsb.watchdog_accepts_refused);
+ if (refuse_retry == nullptr && action_->is_listening()) {
+ refuse_retry = t->schedule_in(this,
HRTIME_MSECONDS(net_throttle_delay));
+ }
+ goto Ldone;
+ }
+ }
Review Comment:
Connections discarded by the connection limits do not increment `count`, so
a continuous stream of rejected connections can keep this loop running without
ever checking newly engaged watchdog backpressure. Check on every iteration so
shedding switches from accepting and closing connections to leaving them queued
in the kernel. Apply the same change in `net_accept()` at line 207 and
`do_blocking_accept()` at line 536.
##########
src/proxy/http/HttpProxyServerMain.cc:
##########
@@ -361,6 +444,14 @@ start_HttpProxyServer()
for (int i = 0, n = proxy_ports.size(); i < n; ++i) {
HttpProxyAcceptor &acceptor = HttpProxyAcceptors[i];
HttpProxyPort &port = proxy_ports[i];
+
+ // Decided before the port is opened, because an overlapping bind needs
SO_REUSEPORT on *both*
+ // sockets and it can only be set before bind().
+ bool const companion = watchdog_shrink_timeout_ms > 0 &&
wants_loopback_companion(port);
+ if (companion) {
+ acceptor._net_opt.f_force_reuseport = true;
Review Comment:
An explicitly loopback-bound proxy port, such as `8080:ip-in=127.0.0.1`,
gets neither a companion nor `f_watchdog_exempt`. Backpressure therefore
shrinks and refuses that localhost listener, leaving local health checks
unprotected. Exempt existing loopback-bound listeners without opening another
socket, while handling the flag's effect on per-thread listener topology
explicitly.
##########
src/iocore/net/UnixNetAccept.cc:
##########
@@ -529,7 +693,52 @@ NetAccept::acceptFastEvent(int event, void *ep)
NetHandler *h = get_NetHandler(t);
int additional_accepts =
NetHandler::get_additional_accepts();
+ if (e == refuse_retry) {
+ refuse_retry = nullptr;
+ }
+
+ // Clones share action_ with the acceptor that stop_accept() closed, so this
is how they learn the listen fd is gone.
+ // It may already belong to another connection, so neither touch it nor arm
another retry.
+ if (!action_->is_listening()) {
+ if (refuse_retry != nullptr) {
+ refuse_retry->cancel(this);
+ refuse_retry = nullptr;
+ }
+ return EVENT_DONE;
Review Comment:
With `proxy.config.exec_thread.listen=1`, clones own independent listening
sockets, but cancelling their shared action closes only the original acceptor's
socket. Returning here leaves the clones' sockets and EventIO registrations
open during the drain, so clients can complete TCP handshakes but are never
accepted. Arrange owner-thread cleanup of each independently owned listener
before standing down.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
