perf(message): batch channel fanout under one registry lock · xeiaso.net/ircd.rs@961fdf9

+96 -20

2 changed files

Expand all

leveva

src

command

message.rs

registry.rs

+67 -20

leveva/src/command/message.rs

View file

··· 272 272 ctx.registry.deliver_uid(&client.uid, render(client.caps)); 273 273 } 274 274 }; 275 + // Batched channel fan-out. Pre-filter to LOCAL, non-sender members (locality is a 276 + // `ctx` decision the registry can't make), then deliver to all of them under a SINGLE 277 + // registry lock via `deliver_batch` — snapshotting each member's modes+caps in one pass 278 + // instead of the old `modes_of` + `caps_of` + `deliver_uid` trio (three lock cycles) per 279 + // member. The wire is rendered **once per distinct capability-combination** and reused 280 + // across members sharing it: `render`'s output depends only on the four caps keyed below 281 + // (`message-tags`/`server-time`/`oper-tag`/`account-tag`), so at most 16 wires are built 282 + // for any fan-out regardless of member count. The per-member `+D` deaf skip (slice 247) 283 + // is preserved via the modes snapshotted inside the batched pass. 284 + let fanout = |members: Vec<crate::ident::Uid>| { 285 + let locals: Vec<crate::ident::Uid> = members 286 + .into_iter() 287 + .filter(|m| *m != client.uid && ctx.is_local_uid(m)) 288 + .collect(); 289 + if locals.is_empty() { 290 + return; 291 + } 292 + let mut memo: std::collections::HashMap<u8, Vec<u8>> = 293 + std::collections::HashMap::new(); 294 + ctx.registry.deliver_batch(&locals, |modes, caps| { 295 + if crate::mode::is_deaf(modes) { 296 + return None; 297 + } 298 + let key = (caps.message_tags as u8) 299 + | ((caps.server_time as u8) << 1) 300 + | ((caps.oper_tag as u8) << 2) 301 + | ((caps.account_tag as u8) << 3); 302 + Some(memo.entry(key).or_insert_with(|| render(caps)).clone()) 303 + }); 304 + }; 275 305 // $$<servermask> / $#<hostmask>: an operator-only broadcast to every user on a 276 306 // matching server / with a matching host. A non-oper's `$$`/`$#` target is not a 277 307 // mask here — it falls through to nick resolution (→ 401 / silent), as in the oracle. ··· 360 390 // Deliver to LOCAL members only; remote members are reached once by the 361 391 // S2S relay below, not by a per-member client-form mailbox push. 362 392 if let Some(members) = ctx.channels.members(target) { 363 - for m in &members { 364 - // `+D` deaf members receive no channel message (slice 247); the 365 - // filter runs on their home server's local fan-out. 366 - if *m != client.uid 367 - && ctx.is_local_uid(m) 368 - && !crate::mode::is_deaf(ctx.registry.modes_of(m)) 369 - { 370 - ctx.registry.deliver_uid(m, render(ctx.registry.caps_of(m))); 371 - } 372 - } 393 + fanout(members); 373 394 } 374 395 // message-ids (slice 130): carry the *same* per-event `msgid` local 375 396 // consumers got onto the S2S wire, so a remote co-member sees the network- ··· 418 439 // sender, no `+C`/`+c` re-check — ops see the raw blocked content (the point 419 440 // of `+z`). (TAGMSG/REDACT S2S op-redirect remains the sole `+z` follow-on.) 420 441 if ctx.channels.op_moderated(target) { 421 - for op in ctx.channels.op_members(target) { 422 - // A deaf (`+D`) op gets no channel message either (slice 247). 423 - if op != client.uid 424 - && ctx.is_local_uid(&op) 425 - && !crate::mode::is_deaf(ctx.registry.modes_of(&op)) 426 - { 427 - ctx.registry 428 - .deliver_uid(&op, render(ctx.registry.caps_of(&op))); 429 - } 430 - } 442 + // Local ops only, one lock pass, wire memoized by cap-combo — same 443 + // batched fan-out as the `CanSend::Ok` arm (deaf ops still skipped). 444 + fanout(ctx.channels.op_members(target)); 431 445 // Carry the same network-stable msgid/server-time the `CanSend::Ok` path 432 446 // mints (only when a peer is linked), so a remote op sees the identical 433 447 // id/time a local op would. ··· 726 740 ctx.registry.set_modes(&uid_for("bob"), 0); 727 741 assert!(run(&ctx, "PRIVMSG #rust :two").is_empty()); 728 742 assert_eq!(delivered(&mut bob).trailing(), Some("two")); 743 + } 744 + 745 + /// Batched channel fan-out with **mixed caps + a deaf member**: one server-time consumer, 746 + /// two plain recipients (sharing a capability-combo, so the memoized wire is reused across 747 + /// them), and one `+D` deaf member. Each non-deaf member gets the text with exactly its own 748 + /// per-cap treatment (`@time` only for the consumer); the deaf member gets nothing. Exercises 749 + /// the single-lock `deliver_batch` pass and the render-once-per-cap-combo memoization while 750 + /// preserving the per-member deaf skip. 751 + #[test] 752 + fn channel_fanout_mixed_caps_and_deaf_member() { 753 + let ctx = ctx(); 754 + let mut consumer = claim_time_consumer(&ctx, "bob"); // negotiated server-time 755 + let mut plain1 = claim_observed(&ctx, "carol"); // shares the plain combo… 756 + let mut plain2 = claim_observed(&ctx, "dave"); // …with this one (memoized wire reused) 757 + let mut deaf = claim_observed(&ctx, "erin"); 758 + ctx.registry 759 + .set_modes(&uid_for("erin"), crate::mode::UserMode::Deaf.bit()); 760 + for nick in ["alice", "bob", "carol", "dave", "erin"] { 761 + ctx.channels.join("#c", &uid_for(nick), 0); 762 + } 763 + assert!(run(&ctx, "PRIVMSG #c :hi all").is_empty()); 764 + // The server-time consumer gets the text plus a well-formed @time tag. 765 + let mc = delivered(&mut consumer); 766 + assert_eq!(mc.trailing(), Some("hi all")); 767 + assert!(looks_like_server_time(mc.tag("time").unwrap())); 768 + // Both plain recipients get the identical (memoized) untagged wire. 769 + for rx in [&mut plain1, &mut plain2] { 770 + let m = delivered(rx); 771 + assert_eq!(m.trailing(), Some("hi all")); 772 + assert_eq!(m.tag("time"), None); 773 + } 774 + // The +D member receives nothing (deaf skip inside the batched pass). 775 + assert!(deaf.try_recv().is_err(), "a +D member receives no channel message"); 729 776 } 730 777 731 778 // ---- +R registered-only messages user mode (slice 248) ---------------------------

+29

leveva/src/registry.rs

View file

··· 478 478 } 479 479 } 480 480 481 + /// Batched fan-out (the performance path for [`crate::command::message`]'s channel 482 + /// delivery): deliver a wire line to each UID in `members` under a **single** lock 483 + /// acquisition, replacing the per-member `modes_of` + `caps_of` + `deliver_uid` trio (three 484 + /// independent lock cycles each) with one pass over the table. 485 + /// 486 + /// For every member still registered, `wire_for` is called with that member's mirrored 487 + /// user-modes and negotiated caps — snapshotted from the record while the lock is held — and 488 + /// decides delivery: `None` skips the member (e.g. a `+D` deaf receiver), `Some(bytes)` 489 + /// enqueues `bytes` on its mailbox with the usual SendQ accounting (identical to 490 + /// [`deliver_uid`](Self::deliver_uid)). The closure is expected to **memoize** its rendering 491 + /// by capability-combination so the wire is serialized once per distinct combo rather than 492 + /// once per member; it must not call back into the registry (the lock is held across the whole 493 + /// pass). A member not in the table is silently skipped, exactly as `deliver_uid` returns 494 + /// `false` for an unknown UID. Members are visited in slice order, so delivery order is 495 + /// preserved. 496 + pub fn deliver_batch<F>(&self, members: &[Uid], mut wire_for: F) 497 + where 498 + F: FnMut(u32, crate::cap::ClientCaps) -> Option<Vec<u8>>, 499 + { 500 + let g = self.inner.lock().expect("registry mutex poisoned"); 501 + for uid in members { 502 + if let Some(entry) = g.by_uid.get(uid) { 503 + if let Some(bytes) = wire_for(entry.record.modes, entry.record.caps) { 504 + entry.enqueue(Envelope::Wire(bytes)); 505 + } 506 + } 507 + } 508 + } 509 + 481 510 /// Forcibly disconnect the connection owning `uid` (a `KILL`): write `wire` to it, 482 511 /// then close it, relaying `QUIT :reason` to its channel co-members. Returns `false` 483 512 /// if no such client is registered. Delivered in-band on the mailbox so it lands

Configure Feed

Configure Feed