wip

Author: 0x416e746f6e

crates/rproxy/src/server/proxy/http/proxy.rs

@@ -77,7 +77,7 @@ where
     shared: ProxyHttpSharedState<C, P>,
 
     backend: ProxyHttpBackendEndpoint<C, P>,
-    requests: HashMap<Uuid, ProxiedHttpRequest>,
+    requests: Arc<HashMap<Uuid, ProxiedHttpRequest>>,
     postprocessor: actix::Addr<ProxyHttpPostprocessor<C, P>>,
 }
 
@@ -130,7 +130,7 @@ where
         }
         .start();
 
-        Self { id, shared, backend, requests: HashMap::default(), postprocessor }
+        Self { id, shared, backend, requests: Arc::new(HashMap::default()), postprocessor }
     }
 
     pub(crate) async fn run(
@@ -641,35 +641,49 @@ where
     }
 
     fn postprocess_client_request(&self, req: ProxiedHttpRequest) {
-        let id = req.info.req_id;
+        let req_id = req.info.req_id;
         let conn_id = req.info.conn_id;
+        let worker_id = self.id;
 
-        if self.requests.insert_sync(id, req).is_err() {
-            error!(
-                proxy = P::name(),
-                request_id = %id,
-                connection_id = %conn_id,
-                worker_id = %self.id,
-                "Duplicate request id",
-            );
-        };
+        let requests = self.requests.clone();
+
+        actix_web::rt::task::spawn_blocking(move || {
+            if requests.insert_sync(req_id, req).is_err() {
+                error!(
+                    proxy = P::name(),
+                    request_id = %req_id,
+                    connection_id = %conn_id,
+                    worker_id = %worker_id,
+                    "Duplicate request id",
+                );
+            };
+        });
     }
 
     fn postprocess_backend_response(&self, bknd_res: ProxiedHttpResponse) {
-        let Some((_, clnt_req)) = self.requests.remove_sync(&bknd_res.info.req_id) else {
-            error!(
-                proxy = P::name(),
-                request_id = %bknd_res.info.req_id,
-                connection_id = %bknd_res.info.conn_id,
-                worker_id = %self.id,
-                "Proxied http response for unmatching request",
-            );
-            return;
-        };
+        let req_id = bknd_res.info.req_id;
+        let conn_id = bknd_res.info.conn_id;
+        let worker_id = self.id;
+
+        let requests = self.requests.clone();
+        let postprocessor = self.postprocessor.clone();
+
+        actix_web::rt::task::spawn_blocking(move || {
+            let Some((_, clnt_req)) = requests.remove_sync(&req_id) else {
+                error!(
+                    proxy = P::name(),
+                    request_id = %req_id,
+                    connection_id = %conn_id,
+                    worker_id = %worker_id,
+                    "Proxied http response for unmatching request",
+                );
+                return;
+            };
 
-        // hand over to postprocessor asynchronously so that we can return the
-        // response to the client as early as possible
-        self.postprocessor.do_send(ProxiedHttpCombo { req: clnt_req, res: bknd_res });
+            // hand over to postprocessor asynchronously so that we can return the
+            // response to the client as early as possible
+            postprocessor.do_send(ProxiedHttpCombo { req: clnt_req, res: bknd_res });
+        });
     }
 
     fn finalise_proxying(