net: Start reducing number of IPCs channels used for fetch with a `FetchThread` (#33863)

Instead of creating a `ROUTER` for each fetch, create a fetch thread
which handles all incoming and outcoming fetch requests. Now messages
involving fetches carry a "request id" which indicates which fetch is
being addressed by the message. This greatly reduces the number of file
descriptors used by fetch.

In addition, the interface for kicking off fetches is simplified when
using the `Listener` with `Document`s and the `GlobalScope`.

This does not fix all leaked file descriptors / mach ports, but greatly
eliminates the number used. Now tests can be run without limiting
procesess on modern macOS systems.

Followup work:

1. There are more instances where fetch is done using the old method.
   Some of these require more changes in order to be converted to the
   `FetchThread` approach.
2. Eliminate usage of IPC channels when doing redirects.
3. Also eliminate the IPC channel used for cancel handling.
4. This change opens up the possiblity of controlling the priority of
   fetch requests.

Fixes #29834.

Signed-off-by: Martin Robinson <mrobinson@igalia.com>

1 files changed, 35 insertions, 45 deletions

diff --git a/components/script/dom/htmlvideoelement.rs b/components/script/dom/htmlvideoelement.rs
index fa9f9c86f94..3211a7c9e37 100644
--- a/components/script/dom/htmlvideoelement.rs
+++ b/components/script/dom/htmlvideoelement.rs
@@ -3,22 +3,21 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use std::cell::Cell;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 use dom_struct::dom_struct;
 use euclid::default::Size2D;
 use html5ever::{local_name, LocalName, Prefix};
 use ipc_channel::ipc;
-use ipc_channel::router::ROUTER;
 use js::rust::HandleObject;
 use net_traits::image_cache::{
     ImageCache, ImageCacheResult, ImageOrMetadataAvailable, ImageResponse, PendingImageId,
     UsePlaceholder,
 };
-use net_traits::request::{CredentialsMode, Destination, RequestBuilder};
+use net_traits::request::{CredentialsMode, Destination, RequestBuilder, RequestId};
 use net_traits::{
-    CoreResourceMsg, FetchChannels, FetchMetadata, FetchResponseListener, FetchResponseMsg,
-    NetworkError, ResourceFetchTiming, ResourceTimingType,
+    FetchMetadata, FetchResponseListener, FetchResponseMsg, NetworkError, ResourceFetchTiming,
+    ResourceTimingType,
 };
 use servo_media::player::video::VideoFrame;
 use servo_url::ServoUrl;
@@ -41,7 +40,7 @@ use crate::dom::performanceresourcetiming::InitiatorType;
 use crate::dom::virtualmethods::VirtualMethods;
 use crate::fetch::FetchCanceller;
 use crate::image_listener::{generate_cache_listener_for_element, ImageCacheListener};
-use crate::network_listener::{self, NetworkListener, PreInvoke, ResourceTimingListener};
+use crate::network_listener::{self, PreInvoke, ResourceTimingListener};
 use crate::script_runtime::CanGc;
 
 const DEFAULT_WIDTH: u32 = 300;
@@ -213,34 +212,11 @@ impl HTMLVideoElement {
             LoadType::Image(poster_url.clone()),
         ));
 
-        let window = window_from_node(self);
-        let context = Arc::new(Mutex::new(PosterFrameFetchContext::new(
-            self, poster_url, id,
-        )));
-
-        let (action_sender, action_receiver) = ipc::channel().unwrap();
-        let (task_source, canceller) = window
-            .task_manager()
-            .networking_task_source_with_canceller();
-        let listener = NetworkListener {
-            context,
-            task_source,
-            canceller: Some(canceller),
-        };
-        ROUTER.add_route(
-            action_receiver.to_opaque(),
-            Box::new(move |message| {
-                listener.notify_fetch(message.to().unwrap());
-            }),
-        );
-        let global = self.global();
-        global
-            .core_resource_thread()
-            .send(CoreResourceMsg::Fetch(
-                request,
-                FetchChannels::ResponseMsg(action_sender, Some(cancel_receiver)),
-            ))
-            .unwrap();
+        let context = PosterFrameFetchContext::new(self, poster_url, id);
+
+        // TODO: If this is supposed to to be a "fetch" as defined in the specification
+        // this should probably be integrated into the Document's list of cancellable fetches.
+        document_from_node(self).fetch_background(request, context, Some(cancel_receiver));
     }
 }
 
@@ -326,12 +302,18 @@ struct PosterFrameFetchContext {
 }
 
 impl FetchResponseListener for PosterFrameFetchContext {
-    fn process_request_body(&mut self) {}
-    fn process_request_eof(&mut self) {}
+    fn process_request_body(&mut self, _: RequestId) {}
+    fn process_request_eof(&mut self, _: RequestId) {}
 
-    fn process_response(&mut self, metadata: Result<FetchMetadata, NetworkError>) {
-        self.image_cache
-            .notify_pending_response(self.id, FetchResponseMsg::ProcessResponse(metadata.clone()));
+    fn process_response(
+        &mut self,
+        request_id: RequestId,
+        metadata: Result<FetchMetadata, NetworkError>,
+    ) {
+        self.image_cache.notify_pending_response(
+            self.id,
+            FetchResponseMsg::ProcessResponse(request_id, metadata.clone()),
+        );
 
         let metadata = metadata.ok().map(|meta| match meta {
             FetchMetadata::Unfiltered(m) => m,
@@ -352,19 +334,27 @@ impl FetchResponseListener for PosterFrameFetchContext {
         }
     }
 
-    fn process_response_chunk(&mut self, payload: Vec<u8>) {
+    fn process_response_chunk(&mut self, request_id: RequestId, payload: Vec<u8>) {
         if self.cancelled {
             // An error was received previously, skip processing the payload.
             return;
         }
 
-        self.image_cache
-            .notify_pending_response(self.id, FetchResponseMsg::ProcessResponseChunk(payload));
+        self.image_cache.notify_pending_response(
+            self.id,
+            FetchResponseMsg::ProcessResponseChunk(request_id, payload),
+        );
     }
 
-    fn process_response_eof(&mut self, response: Result<ResourceFetchTiming, NetworkError>) {
-        self.image_cache
-            .notify_pending_response(self.id, FetchResponseMsg::ProcessResponseEOF(response));
+    fn process_response_eof(
+        &mut self,
+        request_id: RequestId,
+        response: Result<ResourceFetchTiming, NetworkError>,
+    ) {
+        self.image_cache.notify_pending_response(
+            self.id,
+            FetchResponseMsg::ProcessResponseEOF(request_id, response),
+        );
     }
 
     fn resource_timing_mut(&mut self) -> &mut ResourceFetchTiming {