From cf2b1578423d63a71a317912eb3305568ffde66e Mon Sep 17 00:00:00 2001 From: reverse Date: Tue, 18 Aug 2026 20:30:07 +0000 Subject: [PATCH] perf: to_channel shares one Arc across all broadcast recipients instead of cloning the line per member (server-time members share a single time-tagged variant); single-recipient sends unchanged --- src/link.rs | 4 +-- src/server.rs | 70 +++++++++++++++++++++++++++++++++++++-------- src/socketengine.rs | 48 +++++++++++++++++++++++++++---- 3 files changed, 103 insertions(+), 19 deletions(-) diff --git a/src/link.rs b/src/link.rs index a747b34..860473c 100644 --- a/src/link.rs +++ b/src/link.rs @@ -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()); } } diff --git a/src/server.rs b/src/server.rs index 7c45d02..d52fb4a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -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) { - 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 = std::sync::Arc::from(line); + let sourced = line.starts_with(':'); // only `:prefix …` lines carry server-time + let mut tagged: Option> = 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 = arx.try_iter().collect(); + let bob: Vec = 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(); diff --git a/src/socketengine.rs b/src/socketengine.rs index 525b88f..09c2988 100644 --- a/src/socketengine.rs +++ b/src/socketengine.rs @@ -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), +} + +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 for LineBuf { + fn from(s: String) -> Self { + LineBuf::Owned(s) + } +} + pub enum OutSink { Thread(Sender), 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