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, 10 insertions, 9 deletions

diff --git a/components/net/fetch/methods.rs b/components/net/fetch/methods.rs
index f5abcd3003c..7d64b57436d 100644
--- a/components/net/fetch/methods.rs
+++ b/components/net/fetch/methods.rs
@@ -465,7 +465,7 @@ pub async fn main_fetch(
     let mut response_loaded = false;
     let mut response = if !response.is_network_error() && !request.integrity_metadata.is_empty() {
         // Step 19.1.
-        wait_for_response(&mut response, target, done_chan).await;
+        wait_for_response(request, &mut response, target, done_chan).await;
         response_loaded = true;
 
         // Step 19.2.
@@ -487,12 +487,12 @@ pub async fn main_fetch(
     if request.synchronous {
         // process_response is not supposed to be used
         // by sync fetch, but we overload it here for simplicity
-        target.process_response(&response);
+        target.process_response(request, &response);
         if !response_loaded {
-            wait_for_response(&mut response, target, done_chan).await;
+            wait_for_response(request, &mut response, target, done_chan).await;
         }
         // overloaded similarly to process_response
-        target.process_response_eof(&response);
+        target.process_response_eof(request, &response);
         return response;
     }
 
@@ -507,15 +507,15 @@ pub async fn main_fetch(
     }
 
     // Step 22.
-    target.process_response(&response);
+    target.process_response(request, &response);
 
     // Step 23.
     if !response_loaded {
-        wait_for_response(&mut response, target, done_chan).await;
+        wait_for_response(request, &mut response, target, done_chan).await;
     }
 
     // Step 24.
-    target.process_response_eof(&response);
+    target.process_response_eof(request, &response);
 
     if let Ok(http_cache) = context.state.http_cache.write() {
         http_cache.update_awaiting_consumers(request, &response);
@@ -527,6 +527,7 @@ pub async fn main_fetch(
 }
 
 async fn wait_for_response(
+    request: &Request,
     response: &mut Response,
     target: Target<'_>,
     done_chan: &mut DoneChannel,
@@ -535,7 +536,7 @@ async fn wait_for_response(
         loop {
             match ch.1.recv().await {
                 Some(Data::Payload(vec)) => {
-                    target.process_response_chunk(vec);
+                    target.process_response_chunk(request, vec);
                 },
                 Some(Data::Done) => {
                     break;
@@ -555,7 +556,7 @@ async fn wait_for_response(
             // in case there was no channel to wait for, the body was
             // obtained synchronously via scheme_fetch for data/file/about/etc
             // We should still send the body across as a chunk
-            target.process_response_chunk(vec.clone());
+            target.process_response_chunk(request, vec.clone());
         } else {
             assert_eq!(*body, ResponseBody::Empty)
         }