Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 22 additions & 22 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ categories = ["network-programming", "web-programming::http-server"]

# 二进制走 CalVer,crate 走 SemVer —— 两者是两回事,卖给不同的人。
# 这里是 crate 的版本。
version = "0.83.1"
version = "0.83.2"

[workspace.dependencies]
# ── 内部 crate ───────────────────────────────────────────
Expand Down
40 changes: 29 additions & 11 deletions crates/tw-control/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2109,6 +2109,18 @@ impl Accept for tokio::net::UnixListener {
}
}

/// accept 出错之后歇多久再接。对方在接之前就重置、中止的只是那一个连接,马上接下一个;别的错
/// (打开的文件数到顶这类)立刻再 accept 还是同样的错,会原地空转、刷满日志 —— 和网关端口
/// (axum)一样歇 1 秒
pub(crate) fn accept_backoff(e: &std::io::Error) -> Option<std::time::Duration> {
use std::io::ErrorKind::{ConnectionAborted, ConnectionRefused, ConnectionReset};
(!matches!(
e.kind(),
ConnectionRefused | ConnectionAborted | ConnectionReset
))
.then_some(std::time::Duration::from_secs(1))
}

/// accept 循环。**交接时停下**([`handover::ControlListener::pause`]):停的时候不去接,
/// 监听器本身留着,恢复了接着接 —— 见 [`handover`] 的模块文档。
async fn accept_loop<L: Accept>(
Expand All @@ -2120,18 +2132,24 @@ async fn accept_loop<L: Accept>(
let mut accepting = handover.control().accepting();
loop {
handover::until_accepting(&mut accepting).await;
tokio::select! {
r = listener.accept_one() => match r {
Ok(stream) => hand_off(
stream,
app.clone(),
gate.clone(),
Local(Some(handover.control().closing())),
),
Err(e) => tracing::warn!("the control plane could not accept a connection: {e}"),
},
let r = tokio::select! {
r = listener.accept_one() => r,
// 停下的那一刻放下手里这次 accept:不然它还会再接一个
_ = accepting.wait_for(|on| !*on) => {}
_ = accepting.wait_for(|on| !*on) => continue,
};
match r {
Ok(stream) => hand_off(
stream,
app.clone(),
gate.clone(),
Local(Some(handover.control().closing())),
),
Err(e) => {
tracing::warn!("the control plane could not accept a connection: {e}");
if let Some(d) = accept_backoff(&e) {
tokio::time::sleep(d).await;
}
}
}
}
}
Expand Down
20 changes: 20 additions & 0 deletions crates/tw-control/src/remote.rs
Original file line number Diff line number Diff line change
Expand Up @@ -519,6 +519,9 @@ async fn accept_loop(
Ok(x) => x,
Err(e) => {
tracing::warn!("the remote control port could not accept a connection: {e}");
if let Some(d) = crate::accept_backoff(&e) {
tokio::time::sleep(d).await;
}
continue;
}
},
Expand Down Expand Up @@ -582,6 +585,23 @@ mod tests {
assert!(!allowed(&[], ip("10.1.2.3")), "空名单谁都不放");
}

#[test]
fn accept_errors_pause_unless_only_that_connection_went_away() {
use std::io::{Error, ErrorKind};
for kind in [
ErrorKind::ConnectionRefused,
ErrorKind::ConnectionAborted,
ErrorKind::ConnectionReset,
] {
assert_eq!(crate::accept_backoff(&Error::from(kind)), None, "{kind:?}");
}
// 打开的文件数到顶(EMFILE)报不出专门的 ErrorKind
assert_eq!(
crate::accept_backoff(&Error::other("too many open files")),
Some(Duration::from_secs(1))
);
}

#[test]
fn five_failures_in_a_minute_bench_a_source_for_a_minute() {
let r = Remote::default();
Expand Down
4 changes: 2 additions & 2 deletions crates/tw-control/src/sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -742,7 +742,7 @@ impl Sync {
/// 同步一轮(顺带处理几处冲突),更新状态、发事件
pub async fn round(&self, decisions: &[Decision]) -> Result<tw_api::SyncView, Fail> {
let mut guard = self.run.lock().await;
// Consume this trigger before any early return, or an expired deadline spins the scheduler.
// 一开头就把到点的改动认领下来:下面几处提前返回要是把它留着,过了点的时刻会让调度原地空转
if let Ok(mut d) = self.dirty.lock() {
*d = None;
}
Expand Down Expand Up @@ -1777,7 +1777,7 @@ mod tests {
sync.paused
.store(expected == tw_api::SyncState::Paused, Ordering::Relaxed);

// An expired dirty deadline must not override the next periodic poll forever.
// 过了点的改动时刻不能一直压过下一次定时检查
for _ in 0..2 {
sync.touch();
assert!(sync.dirty.lock().unwrap().is_some());
Expand Down
11 changes: 11 additions & 0 deletions release-notes/0.83.2.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
This release fixes core using a full CPU core while idle after a configuration change, and makes the control plane pause instead of spinning when it cannot accept connections.

**Upgrade notes**

- The control-plane protocol version stays at 54; ThinkWatch Lite releases that include 0.83.0 or 0.83.1 connect to 0.83.2 and the other way round.
- The request store's schema and the configuration are unchanged; nothing is cleared.
- No new message codes.

**No more busy loop after a configuration change.** Since 0.78.0, any local configuration change (renaming a client, adding an upstream, editing the file) scheduled a sync round 10 seconds later. When sync was off, could not be set up from its settings, or was paused in safe mode, that round ended without consuming the scheduled time, so the sync scheduler woke immediately again and kept one CPU core busy until core restarted. The scheduled time is now consumed at the start of every round. Periodic checks and retries are unchanged.

**The control plane pauses after an accept error.** When the local control socket or the remote control port could not accept a connection for a reason other than the connection itself going away (for example when core has run out of file descriptors), it retried at once, used a full CPU core and wrote a warning on every attempt. It now waits one second before accepting again, as the gateway ports already do.
Loading