use rt mpsc channel

Author: lherman-cs

pulsebeam-runtime/src/sync/mpsc.rs

@@ -25,6 +25,9 @@ struct Slot<T> {
 
 #[derive(Debug)]
 struct Ring<T> {
+    // This is mspc, but we expect very low contention on the producers.
+    // Mutex is generally cheaper than RWLock. So, no reason to pay
+    // RWLock overhead.
     slots: Vec<Mutex<Slot<T>>>, // per-slot mutex
     mask: usize,
     head: AtomicU64, // next free sequence number

pulsebeam/src/gateway/actor.rs

@@ -1,13 +1,16 @@
 use crate::gateway::demux::Demuxer;
 use pulsebeam_runtime::actor::{ActorKind, ActorStatus, RunnerConfig};
 use pulsebeam_runtime::prelude::*;
-use pulsebeam_runtime::{actor, mailbox, net};
+use pulsebeam_runtime::{actor, net};
 use std::{io, sync::Arc};
 use tokio::task::JoinSet;
 
 #[derive(Clone)]
 pub enum GatewayControlMessage {
-    AddParticipant(String, mailbox::Sender<net::RecvPacketBatch>),
+    AddParticipant(
+        String,
+        pulsebeam_runtime::sync::mpsc::Sender<net::RecvPacketBatch>,
+    ),
     RemoveParticipant(String),
 }
 
@@ -154,7 +157,8 @@ impl GatewayWorkerActor {
     }
 
     async fn read_socket(&mut self) -> io::Result<()> {
-        const COOP_BUDGET: usize = 128;
+        // ~1,000 yields per second = ~99% CPU for other work
+        const COOP_BUDGET: usize = 256;
         let mut spent_budget: usize = 0;
 
         loop {
@@ -171,7 +175,7 @@ impl GatewayWorkerActor {
                         };
 
                         let src = batch.src;
-                        if !self.demuxer.demux(&mut self.socket, batch).await {
+                        if !self.demuxer.demux(&mut self.socket, batch) {
                             // In case there's a malicious actor, close immediately as there's no
                             // associated participant.
                             self.socket.close_peer(&src);

pulsebeam/src/gateway/demux.rs

@@ -1,11 +1,11 @@
+use pulsebeam_runtime::net;
 use pulsebeam_runtime::net::UnifiedSocketReader;
-use pulsebeam_runtime::{mailbox, net};
 
 use crate::gateway::ice;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 
-pub type ParticipantHandle = mailbox::Sender<net::RecvPacketBatch>;
+pub type ParticipantHandle = pulsebeam_runtime::sync::mpsc::Sender<net::RecvPacketBatch>;
 
 /// A UDP demuxer that maps packets to participants based on source address and STUN ufrag.
 ///
@@ -57,11 +57,7 @@ impl Demuxer {
 
     /// Routes a packet to the correct participant.
     /// Returns `true` if sent, `false` if dropped
-    pub async fn demux(
-        &mut self,
-        socket: &mut UnifiedSocketReader,
-        batch: net::RecvPacketBatch,
-    ) -> bool {
+    pub fn demux(&mut self, socket: &mut UnifiedSocketReader, batch: net::RecvPacketBatch) -> bool {
         let src = batch.src;
 
         let handle = if let Some(h) = self.addr_map.get_mut(&src) {
@@ -83,7 +79,7 @@ impl Demuxer {
             return false;
         };
 
-        if let Err(_) = handle.send(batch).await {
+        if let Err(_) = handle.try_send(batch) {
             // Handle is closed! Clean up everything related to this participant.
             if let Some(ufrag) = self.addr_to_ufrag.get(&src).cloned() {
                 tracing::info!("Participant handle closed, cleaning up ufrag: {:?}", ufrag);

pulsebeam/src/participant/actor.rs

@@ -70,7 +70,7 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
         ctx: &mut actor::ActorContext<ParticipantMessageSet>,
     ) -> Result<(), actor::ActorError> {
         let ufrag = self.core.rtc.direct_api().local_ice_credentials().ufrag;
-        let (gateway_tx, mut gateway_rx) = mailbox::new(64);
+        let (gateway_tx, mut gateway_rx) = pulsebeam_runtime::sync::mpsc::channel(64);
         let room_handle = self.room_handle.clone();
 
         let _ = self
@@ -159,7 +159,7 @@ impl actor::Actor<ParticipantMessageSet> for ParticipantActor {
                 },
 
                 // Priority 3: Ingress Work
-                Some(batch) = gateway_rx.recv() => {
+                Ok(batch) = gateway_rx.recv() => {
                     maybe_deadline = self.core.handle_udp_packet_batch(batch);
                 },