perf(browser): fail dead dials in 3 s and use the whole read budget

ant-core/src/browser/wasm_transport.rs

@@ -100,6 +100,13 @@ const DEFAULT_BROWSER_QUOTE_CONCURRENCY: usize = 4;
 const MAX_READ_RESPONSE_MEMORY: usize = 128 * 1024 * 1024;
 const READ_RESPONSE_RESERVATION: usize =
     3 * (MAX_BROWSER_RESPONSE_BYTES + PQ_ENCRYPTED_OVERHEAD_BYTES);
+/// How long a read's first GET waits for its response, once dispatched to its
+/// peer, before the read also asks another holder. A healthy peer takes 1-3 s to
+/// carry a 4 MiB record over a WebRTC DataChannel, so the native 1 s raced nearly
+/// every read and fetched a second copy. The delay also covers the GET's wait for
+/// its peer: a session slot, and replies its session carries ahead of it. A peer
+/// busy with several of this client's GETs may therefore be raced.
+const BROWSER_READ_HEDGE_DELAY: Duration = Duration::from_secs(3);
 const MAX_BROWSER_RANGE_BYTES: usize = 4 * 1024 * 1024;
 const MAX_RANGE_CACHE_BYTES: usize = 32 * 1024 * 1024;
 const MAX_UPLOAD_CHECKPOINT_BYTES: usize = 128 * 1024 * 1024;
@@ -238,7 +245,19 @@ impl Deref for BrowserClientLease {
 
 impl BrowserClientLease {
     async fn authenticated(&self) -> Result<LockedBrowserClient<'_>, RpcError> {
-        self.client.authenticated_before(&self.admission).await
+        self.client
+            .authenticated_before(&self.admission, None)
+            .await
+    }
+
+    /// Authenticate for a GET, whose session slot follows its read priority.
+    async fn authenticated_read(
+        &self,
+        read: &multiplex::ReadPriority<'_>,
+    ) -> Result<LockedBrowserClient<'_>, RpcError> {
+        self.client
+            .authenticated_before(&self.admission, Some(read))
+            .await
     }
 
     async fn hello(&self) -> Result<BrowserHello, String> {
@@ -281,7 +300,7 @@ impl BrowserClientLease {
                     client.authenticate().await?;
                     drop(client);
                     let client = match select(
-                        Box::pin(self.client.authenticated_before(&self.admission)),
+                        Box::pin(self.client.authenticated_before(&self.admission, None)),
                         Box::pin(sender.cancellation()),
                     )
                     .await
@@ -605,11 +624,14 @@ impl BrowserClientPool {
         }
     }
 
+    /// Physical GETs admitted at once: the adaptive fetch cap within the read
+    /// budget, or the budget's capacity before a client installs its limiter.
     fn read_limit(&self) -> usize {
-        let desired = self.fetch_limiter.borrow().as_ref().map_or_else(
-            || crate::data::client::adaptive::ChannelStart::default().fetch,
-            |limiter| limiter.current(),
-        );
+        let desired = self
+            .fetch_limiter
+            .borrow()
+            .as_ref()
+            .map_or_else(|| self.read_budget.capacity(), |limiter| limiter.current());
         self.read_budget.limit(desired)
     }
 
@@ -1137,6 +1159,7 @@ impl BrowserNodeClientCore {
     async fn authenticated_before(
         &self,
         admission: &TransferDeadline,
+        read: Option<&multiplex::ReadPriority<'_>>,
     ) -> Result<LockedBrowserClient<'_>, RpcError> {
         loop {
             let mut client = self.lock_before(admission).await?;
@@ -1146,7 +1169,11 @@ impl BrowserNodeClientCore {
                 .current_rpc()
                 .ok_or_else(|| "WebRTC session unavailable".to_string())?;
             let generation = self.generation.get();
-            match rpc.admit(admission.remaining()).await {
+            let admitted = match read {
+                Some(read) => rpc.admit_read(admission.remaining(), read).await,
+                None => rpc.admit(admission.remaining()).await,
+            };
+            match admitted {
                 Ok(_) if generation != self.generation.get() => continue,
                 Ok(slot) if !rpc.is_closed() => {
                     client.slot.replace(Some(slot));
@@ -1173,7 +1200,7 @@ impl BrowserNodeClientCore {
 
     #[cfg(feature = "test-utils")]
     async fn authenticated(&self) -> Result<LockedBrowserClient<'_>, RpcError> {
-        self.authenticated_before(&TransferDeadline::new(RPC_ADMISSION_TIMEOUT))
+        self.authenticated_before(&TransferDeadline::new(RPC_ADMISSION_TIMEOUT), None)
             .await
     }
 
@@ -1214,7 +1241,7 @@ impl Drop for BrowserNodeClientCore {
 struct LockedBrowserClient<'a> {
     client: &'a BrowserNodeClientCore,
     _guard: Option<MutexGuard<'a, ()>>,
-    slot: RefCell<Option<tokio::sync::OwnedSemaphorePermit>>,
+    slot: RefCell<Option<multiplex::SessionSlot>>,
     authenticated_generation: Option<u64>,
 }
 
@@ -2357,11 +2384,24 @@ impl BrowserNetworkClient {
         let inner = Rc::new(inner);
         let adapter = Rc::new(SharedNetworkAdapter::new(Rc::clone(&inner)));
         let network = crate::data::Network::from_browser(adapter.clone());
+        // Fetches beyond the GETs the read budget admits at once would only
+        // queue, so the fetch channel starts there and never probes past it. A
+        // WebRTC DataChannel carries a record more slowly than QUIC, so the
+        // native cold start would leave much of the budget idle. The adaptive
+        // controller still backs off from here. A hedge or a GET after
+        // discovery may then wait for a permit another fetch's GET frees, and
+        // holds nothing of its peer's meanwhile.
+        let read_capacity = inner.pool.read_budget.capacity();
+        let mut config = crate::data::ClientConfig::default();
+        config.adaptive.max.fetch = read_capacity;
+        let start = crate::data::client::adaptive::ChannelStart {
+            fetch: read_capacity,
+            ..Default::default()
+        };
         let shared = Rc::new(
-            crate::data::Client::from_network(network, crate::data::ClientConfig::default())
-                .with_chunk_cache(crate::data::ChunkCache::new(
-                    MAX_RANGE_CACHE_BYTES / MAX_BROWSER_RECORD_BYTES,
-                )),
+            crate::data::Client::from_network_with_start(network, config, start).with_chunk_cache(
+                crate::data::ChunkCache::new(MAX_RANGE_CACHE_BYTES / MAX_BROWSER_RECORD_BYTES),
+            ),
         );
         *inner.pool.fetch_limiter.borrow_mut() = Some(shared.controller().fetch.clone());
         let read_ahead = read_ahead::ReadAheadPool::new(&inner, &shared);

ant-core/src/browser/wasm_transport/multiplex.rs

@@ -3,19 +3,68 @@
 //! Cancellation drops the caller, not an admitted exchange. Its slot and read
 //! reservation remain owned until the authenticated reply is drained or the
 //! ordinary transport deadline closes the session.
+//!
+//! A speculative GET, one no read waits for yet such as read-ahead, is admitted
+//! only while speculative GETs hold less than a share of a session's slots, and
+//! never takes a slot another request is queued for. A GET a read waits for, such
+//! as a seek's, therefore never waits behind read-ahead that comes after it.
+//!
+//! Like the read budget's permits, a GET is classed once: when it queues as a
+//! read's, or when it is admitted. A GET whose read stops waiting for it, such as
+//! one cancelled by a seek, keeps its slot as a read's until its exchange
+//! completes, so read-ahead may hold more than the share for that long.
 use super::*;
-use crate::client_engine::read_budget::ReadPermit;
-use tokio::sync::Semaphore;
+use crate::client_engine::read_budget::{ReadPermit, SPECULATIVE_SHARE_DIVISOR};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, TryAcquireError};
 
 pub(super) const CAPABILITY: &str = "rpc-multiplex-4";
+/// Requests one multiplexed session carries at once.
 const MAX_REQUESTS: usize = 4;
 type Reply = Result<(BrowserResponseFrame, Duration), RpcError>;
 
+/// How a GET competes for a session's slots. It is speculative while no read
+/// waits for its record, and is asked again whenever `changed` changes.
+pub(super) struct ReadPriority<'a> {
+    pub(super) speculative: &'a dyn Fn() -> bool,
+    pub(super) changed: watch::Receiver<()>,
+}
+
+/// A session slot an admitted request holds until its exchange completes.
+pub(super) struct SessionSlot {
+    permit: Option<OwnedSemaphorePermit>,
+    session: Rc<RpcSession>,
+    speculative: bool,
+}
+
+impl Drop for SessionSlot {
+    fn drop(&mut self) {
+        // Release the slot before waking those waiting for one.
+        drop(self.permit.take());
+        if self.speculative {
+            self.session
+                .speculative
+                .set(self.session.speculative.get() - 1);
+        }
+        self.session.released.send_replace(());
+    }
+}
+
+/// Resolves on the next change. A closed channel never changes again.
+async fn next_change(receiver: &mut watch::Receiver<()>) {
+    if receiver.changed().await.is_err() {
+        futures::future::pending::<()>().await;
+    }
+}
+
 pub(super) struct RpcSession {
     channel: RtcDataChannel,
     inbox: Rc<ResponseInbox>,
     pq: Rc<RefCell<Option<PqSession>>>,
     slots: std::sync::Arc<Semaphore>,
+    /// Slots held by speculative GETs.
+    speculative: Cell<usize>,
+    /// Changes when a slot is released or added, or the session closes.
+    released: watch::Sender<()>,
     send: Mutex<()>,
     pending: RefCell<HashMap<u64, oneshot::Sender<Reply>>>,
     changed: watch::Sender<()>,
@@ -31,11 +80,14 @@ impl RpcSession {
         pq: Rc<RefCell<Option<PqSession>>>,
     ) -> Rc<Self> {
         let (changed, _) = watch::channel(());
+        let (released, _) = watch::channel(());
         let session = Rc::new(Self {
             channel,
             inbox,
             pq,
             slots: std::sync::Arc::new(Semaphore::new(1)),
+            speculative: Cell::new(0),
+            released,
             send: Mutex::new(()),
             pending: RefCell::new(HashMap::new()),
             changed,
@@ -70,6 +122,7 @@ impl RpcSession {
         if !self.multiplex.replace(true) {
             self.inbox.enable_multiplex();
             self.slots.add_permits(MAX_REQUESTS - 1);
+            self.released.send_replace(());
         }
     }
 
@@ -78,6 +131,7 @@ impl RpcSession {
             return;
         }
         self.slots.close();
+        self.released.send_replace(());
         self.inbox.fail(error.clone());
         self.channel.close();
         for (_, sender) in self.pending.take() {
@@ -91,24 +145,88 @@ impl RpcSession {
         self.closed.get()
     }
 
-    pub(super) async fn admit(
-        &self,
-        timeout: Duration,
-    ) -> Result<tokio::sync::OwnedSemaphorePermit, RpcError> {
+    /// Requests the session carries at once.
+    fn capacity(&self) -> usize {
+        if self.multiplex.get() {
+            MAX_REQUESTS
+        } else {
+            1
+        }
+    }
+
+    fn slot(self: &Rc<Self>, permit: OwnedSemaphorePermit, speculative: bool) -> SessionSlot {
+        self.speculative
+            .set(self.speculative.get() + usize::from(speculative));
+        SessionSlot {
+            permit: Some(permit),
+            session: Rc::clone(self),
+            speculative,
+        }
+    }
+
+    pub(super) async fn admit(self: &Rc<Self>, timeout: Duration) -> Result<SessionSlot, RpcError> {
         crate::runtime::timeout(timeout, self.slots.clone().acquire_owned())
             .await
             .map_err(|_| RpcError::Timeout("WebRTC admission timed out waiting for peer".into()))?
+            .map(|permit| self.slot(permit, false))
             .map_err(|_| "WebRTC session closed".to_string().into())
     }
 
+    /// Admit a GET. One a read waits for queues for a slot like any request.
+    /// A speculative one takes a free slot only within the session's
+    /// speculative share and only while no request is queued for it, and is
+    /// asked again whenever a slot is released or `read.changed` changes, so
+    /// a GET a read comes to wait for is admitted as one.
+    pub(super) async fn admit_read(
+        self: &Rc<Self>,
+        timeout: Duration,
+        read: &ReadPriority<'_>,
+    ) -> Result<SessionSlot, RpcError> {
+        let mut released = self.released.subscribe();
+        let mut changed = read.changed.clone();
+        let admission = async {
+            loop {
+                if !(read.speculative)() {
+                    return self
+                        .slots
+                        .clone()
+                        .acquire_owned()
+                        .await
+                        .map(|permit| self.slot(permit, false))
+                        .map_err(|_| "WebRTC session closed".to_string().into());
+                }
+                // At least one slot: a session without multiplexing has one,
+                // which read-ahead may then hold.
+                let share = (self.capacity() / SPECULATIVE_SHARE_DIVISOR).max(1);
+                if self.speculative.get() < share {
+                    match self.slots.clone().try_acquire_owned() {
+                        Ok(permit) => return Ok(self.slot(permit, true)),
+                        Err(TryAcquireError::Closed) => {
+                            return Err("WebRTC session closed".to_string().into())
+                        }
+                        Err(TryAcquireError::NoPermits) => {}
+                    }
+                }
+                select(
+                    Box::pin(next_change(&mut released)),
+                    Box::pin(next_change(&mut changed)),
+                )
+                .await;
+            }
+        };
+        crate::runtime::timeout(timeout, admission)
+            .await
+            .map_err(|_| RpcError::Timeout("WebRTC admission timed out waiting for peer".into()))?
+    }
+
     #[allow(clippy::too_many_arguments)]
     pub(super) async fn request(
         self: &Rc<Self>,
         id: u64,
         plaintext: Vec<u8>,
         timeout: Duration,
         read_permit: Option<ReadPermit>,
-        permit: Option<tokio::sync::OwnedSemaphorePermit>,
+        permit: Option<SessionSlot>,
         exclusive: bool,
     ) -> Result<(BrowserResponseFrame, Duration, Option<ReadPermit>), RpcError> {
         // Preserve per-lane PUT serialization: several complete incoming PUTs
@@ -117,18 +235,15 @@ impl RpcSession {
         // deadlocking when several callers already hold one slot each.
         let permit = if exclusive {
             drop(permit);
-            let count = if self.multiplex.get() {
-                MAX_REQUESTS as u32
-            } else {
-                1
-            };
-            crate::runtime::timeout(
+            let count = self.capacity() as u32;
+            let permit = crate::runtime::timeout(
                 RPC_ADMISSION_TIMEOUT,
                 self.slots.clone().acquire_many_owned(count),
             )
             .await
             .map_err(|_| RpcError::Timeout("WebRTC write admission timed out".into()))?
-            .map_err(|_| "WebRTC session closed".to_string())?
+            .map_err(|_| "WebRTC session closed".to_string())?;
+            self.slot(permit, false)
         } else {
             match permit {
                 Some(permit) => permit,

ant-core/src/browser/wasm_transport/read_ahead.rs

@@ -10,11 +10,12 @@
 //! only within those bounds; the read fetches any record refused here itself.
 //!
 //! Read-ahead fetches through a client of its own. Its GETs are speculative
-//! unless a read waits for their record, and the read budget admits speculative
-//! GETs, cancelled ones included, only within a share of its cap. Only fetches a
-//! read waited for inform the reads' adaptive concurrency, and read-ahead keeps
-//! its records out of the shared chunk cache. Readers reading ahead the same
-//! record at the same time share one fetch.
+//! unless a read waits for their record. The read budget admits speculative
+//! GETs, cancelled ones included, only within a share of its cap, and each
+//! node's session only within a share of its slots, never ahead of a GET a read
+//! waits for. Only fetches a read waited for inform the reads' adaptive
+//! concurrency, and read-ahead keeps its records out of the shared chunk cache.
+//! Readers reading ahead the same record at the same time share one fetch.
 use super::{BrowserNetworkCore, SharedNetworkAdapter};
 use crate::client_engine::read_ahead::{read_ahead_window, releasable};
 use crate::client_engine::read_budget::ReadBudget;