perf: to_channel shares one Arc<str> across all broadcast recipients instead of cloning the line per member (server-time members share a single time-tagged variant); single-recipient sends unchanged
This commit is contained in:
parent
fc58113db9
commit
cf2b157842
3 changed files with 103 additions and 19 deletions
|
|
@ -129,7 +129,7 @@ impl Server {
|
|||
out.send(format!(
|
||||
"SERVER {} {} {} :{}",
|
||||
self.name, pass, self.sid, self.server_desc
|
||||
));
|
||||
).into());
|
||||
sent_server = true;
|
||||
}
|
||||
}
|
||||
|
|
@ -150,7 +150,7 @@ impl Server {
|
|||
|
||||
fn link_out(&self, uid: Uid, line: String) {
|
||||
if let Some(l) = self.links.get(&uid) {
|
||||
l.out.send(line);
|
||||
l.out.send(line.into());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ use crate::link::{Link, RemoteServer, RemoteUser};
|
|||
use crate::module::Hook;
|
||||
use crate::modules::dnsbl;
|
||||
use crate::resolver;
|
||||
use crate::socketengine::OutSink;
|
||||
use crate::socketengine::{LineBuf, OutSink};
|
||||
use crate::users::{Caps, User, UserFlags};
|
||||
use crate::xline::XLine;
|
||||
use crate::Uid;
|
||||
|
|
@ -713,17 +713,17 @@ impl Server {
|
|||
} else {
|
||||
line
|
||||
};
|
||||
self.emit_to(uid, line);
|
||||
self.emit_to(uid, line.into());
|
||||
}
|
||||
}
|
||||
|
||||
/// Final hop for one line to a client: diverted into the labeled-response
|
||||
/// capture buffer when one is active for `uid`, otherwise written to the wire.
|
||||
fn emit_to(&self, uid: Uid, line: String) {
|
||||
fn emit_to(&self, uid: Uid, line: LineBuf) {
|
||||
if let Ok(mut cap) = self.label_capture.try_borrow_mut() {
|
||||
if let Some((cuid, buf)) = cap.as_mut() {
|
||||
if *cuid == uid {
|
||||
buf.push(line);
|
||||
buf.push(line.into_string());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
|
@ -861,7 +861,7 @@ impl Server {
|
|||
} else {
|
||||
format!("@{} {base}", tags.join(";"))
|
||||
};
|
||||
self.emit_to(uid, line);
|
||||
self.emit_to(uid, line.into());
|
||||
}
|
||||
|
||||
/// The escaped json-log value for `msg`, or `""` if none of `targets` want it
|
||||
|
|
@ -960,14 +960,36 @@ impl Server {
|
|||
}
|
||||
}
|
||||
|
||||
/// Send a line to every member of a channel, optionally skipping one uid.
|
||||
/// Send a line to every member of a channel, optionally skipping one uid. The
|
||||
/// line is allocated once and shared (`Arc`) across all recipients — a big
|
||||
/// channel broadcast no longer clones the string per member. A `server-time`
|
||||
/// member gets a time-tagged variant, itself built once and shared.
|
||||
pub fn to_channel(&self, key: &str, line: &str, except: Option<Uid>) {
|
||||
if let Some(ch) = self.channels.get(key) {
|
||||
for &uid in ch.members.keys() {
|
||||
if Some(uid) != except {
|
||||
self.send(uid, line.to_string());
|
||||
}
|
||||
let Some(ch) = self.channels.get(key) else {
|
||||
return;
|
||||
};
|
||||
let plain: std::sync::Arc<str> = std::sync::Arc::from(line);
|
||||
let sourced = line.starts_with(':'); // only `:prefix …` lines carry server-time
|
||||
let mut tagged: Option<std::sync::Arc<str>> = None;
|
||||
for &uid in ch.members.keys() {
|
||||
if Some(uid) == except {
|
||||
continue;
|
||||
}
|
||||
let want_time = sourced
|
||||
&& self
|
||||
.users
|
||||
.get(&uid)
|
||||
.map(|u| u.caps.server_time)
|
||||
.unwrap_or(false);
|
||||
let buf = if want_time {
|
||||
let t = tagged.get_or_insert_with(|| {
|
||||
std::sync::Arc::from(format!("@time={} {line}", iso_time(now())).as_str())
|
||||
});
|
||||
LineBuf::Shared(t.clone())
|
||||
} else {
|
||||
LineBuf::Shared(plain.clone())
|
||||
};
|
||||
self.emit_to(uid, buf);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1004,7 +1026,7 @@ impl Server {
|
|||
} else {
|
||||
format!("@{} {body}", tags.join(";"))
|
||||
};
|
||||
self.emit_to(uid, line);
|
||||
self.emit_to(uid, line.into());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1396,6 +1418,30 @@ mod tests {
|
|||
assert!(!ann.iter().any(|l| l.contains("PART")), "no fallback on case-only: {ann:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn to_channel_shares_line_and_tags_server_time_members() {
|
||||
let mut s = srv();
|
||||
let arx = add_user(&mut s, 1, "ann"); // plain (no server-time)
|
||||
let brx = add_user(&mut s, 2, "bob");
|
||||
s.users.get_mut(&2).unwrap().caps.server_time = true;
|
||||
s.join(1, "#c", None);
|
||||
s.join(2, "#c", None);
|
||||
let _ = arx.try_iter().count();
|
||||
let _ = brx.try_iter().count();
|
||||
s.to_channel("#c", ":x!u@h TOPIC #c :hi", None);
|
||||
let ann: Vec<String> = arx.try_iter().collect();
|
||||
let bob: Vec<String> = brx.try_iter().collect();
|
||||
assert!(
|
||||
ann.iter().any(|l| l == ":x!u@h TOPIC #c :hi"),
|
||||
"plain member gets the untagged line: {ann:?}"
|
||||
);
|
||||
assert!(
|
||||
bob.iter()
|
||||
.any(|l| l.starts_with("@time=") && l.ends_with(":x!u@h TOPIC #c :hi")),
|
||||
"server-time member gets the @time= variant: {bob:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn isupport_advertises_bot_and_account_extban() {
|
||||
let s = srv();
|
||||
|
|
|
|||
|
|
@ -57,7 +57,7 @@ fn normalize_addr(a: SocketAddr) -> SocketAddr {
|
|||
/// [`OutSink`], e.g. on quit), or a per-connection queue-limit override (from the
|
||||
/// assigned connection class).
|
||||
pub enum Out {
|
||||
Line(usize, String),
|
||||
Line(usize, LineBuf),
|
||||
Close(usize),
|
||||
Limits {
|
||||
token: usize,
|
||||
|
|
@ -71,6 +71,43 @@ pub enum Out {
|
|||
/// server links) get a plain channel to their writer thread; reactor connections
|
||||
/// (plaintext clients) get a token plus the shared reactor channel and its waker.
|
||||
/// Either way the core just calls [`OutSink::send`].
|
||||
/// A line queued for delivery: either uniquely owned, or an `Arc` shared by every
|
||||
/// recipient of a channel broadcast — so fanning one line out to N members allocates
|
||||
/// it once, not N times. Both forms write the identical bytes to the wire.
|
||||
pub enum LineBuf {
|
||||
Owned(String),
|
||||
Shared(Arc<str>),
|
||||
}
|
||||
|
||||
impl LineBuf {
|
||||
fn bytes(&self) -> &[u8] {
|
||||
match self {
|
||||
LineBuf::Owned(s) => s.as_bytes(),
|
||||
LineBuf::Shared(a) => a.as_bytes(),
|
||||
}
|
||||
}
|
||||
fn len(&self) -> usize {
|
||||
match self {
|
||||
LineBuf::Owned(s) => s.len(),
|
||||
LineBuf::Shared(a) => a.len(),
|
||||
}
|
||||
}
|
||||
/// Materialise an owned `String` (a move for `Owned`, one copy for `Shared`) —
|
||||
/// for the thread-model sinks and the labeled-response capture buffer.
|
||||
pub fn into_string(self) -> String {
|
||||
match self {
|
||||
LineBuf::Owned(s) => s,
|
||||
LineBuf::Shared(a) => a.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<String> for LineBuf {
|
||||
fn from(s: String) -> Self {
|
||||
LineBuf::Owned(s)
|
||||
}
|
||||
}
|
||||
|
||||
pub enum OutSink {
|
||||
Thread(Sender<String>),
|
||||
Reactor {
|
||||
|
|
@ -81,11 +118,12 @@ pub enum OutSink {
|
|||
}
|
||||
|
||||
impl OutSink {
|
||||
/// Queue one line for delivery (the writer appends CRLF).
|
||||
pub fn send(&self, line: String) {
|
||||
/// Queue one line for delivery (the writer appends CRLF). The reactor sink keeps
|
||||
/// a shared line shared (no copy); the thread sink materialises a `String`.
|
||||
pub fn send(&self, line: LineBuf) {
|
||||
match self {
|
||||
OutSink::Thread(s) => {
|
||||
let _ = s.send(line);
|
||||
let _ = s.send(line.into_string());
|
||||
}
|
||||
OutSink::Reactor { token, tx, waker } => {
|
||||
if tx.send(Out::Line(*token, line)).is_ok() {
|
||||
|
|
@ -621,7 +659,7 @@ fn reactor_loop(
|
|||
c.wbuf.drain(..c.wpos); // reclaim written prefix
|
||||
c.wpos = 0;
|
||||
}
|
||||
c.wbuf.extend_from_slice(line.as_bytes());
|
||||
c.wbuf.extend_from_slice(line.bytes());
|
||||
c.wbuf.extend_from_slice(b"\r\n");
|
||||
// softsendq: over the soft cap, stop reading
|
||||
// their commands until the backlog drains
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue