use a parking-lot mutex in Context

.changesets/fix_geal_context_mutex.md

@@ -0,0 +1,5 @@
+### use a parking-lot mutex in Context ([Issue #2751](https://github.com/apollographql/router/issues/2751))
+
+The context requires synchronized access to the busy timer, and precedently we used a futures aware mutex for that, but those are susceptible to contention. This replaces that mutex with a parking-lot synchronous mutex that is much faster.
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/2885

apollo-router/src/axum_factory/axum_http_server_factory.rs

@@ -436,7 +436,7 @@ async fn handle_graphql(
     let context = request.context.clone();
 
     let res = service.oneshot(request).await;
-    let dur = context.busy_time().await;
+    let dur = context.busy_time();
     let processing_seconds = dur.as_secs_f64();
 
     tracing::info!(histogram.apollo_router_processing_time = processing_seconds,);

apollo-router/src/context/mod.rs

@@ -10,7 +10,8 @@ use std::time::Instant;
 use dashmap::mapref::multiple::RefMulti;
 use dashmap::mapref::multiple::RefMutMulti;
 use dashmap::DashMap;
-use futures::lock::Mutex;
+use derivative::Derivative;
+use parking_lot::Mutex;
 use serde::Deserialize;
 use serde::Serialize;
 use tower::BoxError;
@@ -34,7 +35,8 @@ pub(crate) type Entries = Arc<DashMap<String, Value>>;
 /// [`crate::services::SubgraphResponse`] processing. At such times,
 /// plugins should restrict themselves to the [`Context::get`] and [`Context::upsert`]
 /// functions to minimise the possibility of mis-sequenced updates.
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Clone, Deserialize, Serialize, Derivative)]
+#[derivative(Debug)]
 pub struct Context {
     // Allows adding custom entries to the context.
     entries: Entries,
@@ -48,6 +50,7 @@ pub struct Context {
     pub(crate) created_at: Instant,
 
     #[serde(skip)]
+    #[derivative(Debug = "ignore")]
     busy_timer: Arc<Mutex<BusyTimer>>,
 }
 
@@ -193,18 +196,18 @@ impl Context {
     }
 
     /// Notify the busy timer that we're waiting on a network request
-    pub(crate) async fn enter_active_request(&self) {
-        self.busy_timer.lock().await.increment_active_requests()
+    pub(crate) fn enter_active_request(&self) {
+        self.busy_timer.lock().increment_active_requests()
     }
 
     /// Notify the busy timer that we stopped waiting on a network request
-    pub(crate) async fn leave_active_request(&self) {
-        self.busy_timer.lock().await.decrement_active_requests()
+    pub(crate) fn leave_active_request(&self) {
+        self.busy_timer.lock().decrement_active_requests()
     }
 
     /// How much time was spent working on the request
-    pub(crate) async fn busy_time(&self) -> Duration {
-        self.busy_timer.lock().await.current()
+    pub(crate) fn busy_time(&self) -> Duration {
+        self.busy_timer.lock().current()
     }
 }
 

apollo-router/src/plugins/coprocessor.rs

@@ -516,9 +516,9 @@ where
     };
 
     tracing::debug!(?payload, "externalized output");
-    request.context.enter_active_request().await;
+    request.context.enter_active_request();
     let co_processor_result = payload.call(http_client, &coprocessor_url).await;
-    request.context.leave_active_request().await;
+    request.context.leave_active_request();
     tracing::debug!(?co_processor_result, "co-processor returned");
     let co_processor_output = co_processor_result?;
 
@@ -642,9 +642,9 @@ where
 
     // Second, call our co-processor and get a reply.
     tracing::debug!(?payload, "externalized output");
-    response.context.enter_active_request().await;
+    response.context.enter_active_request();
     let co_processor_result = payload.call(http_client, &coprocessor_url).await;
-    response.context.leave_active_request().await;
+    response.context.leave_active_request();
     tracing::debug!(?co_processor_result, "co-processor returned");
     let co_processor_output = co_processor_result?;
 
@@ -728,9 +728,9 @@ where
     };
 
     tracing::debug!(?payload, "externalized output");
-    request.context.enter_active_request().await;
+    request.context.enter_active_request();
     let co_processor_result = payload.call(http_client, &coprocessor_url).await;
-    request.context.leave_active_request().await;
+    request.context.leave_active_request();
     tracing::debug!(?co_processor_result, "co-processor returned");
     let co_processor_output = co_processor_result?;
     validate_coprocessor_output(&co_processor_output, PipelineStep::SubgraphRequest)?;
@@ -858,9 +858,9 @@ where
     };
 
     tracing::debug!(?payload, "externalized output");
-    response.context.enter_active_request().await;
+    response.context.enter_active_request();
     let co_processor_result = payload.call(http_client, &coprocessor_url).await;
-    response.context.leave_active_request().await;
+    response.context.leave_active_request();
     tracing::debug!(?co_processor_result, "co-processor returned");
     let co_processor_output = co_processor_result?;
 

apollo-router/src/services/subgraph_service.rs

@@ -329,13 +329,13 @@ async fn call_http(
     let cloned_service_name = service_name.clone();
     let cloned_context = context.clone();
     let (parts, body) = async move {
-        cloned_context.enter_active_request().await;
+        cloned_context.enter_active_request();
         let response = match client
             .call(request)
             .await {
                 Err(err) => {
                     tracing::error!(fetch_error = format!("{err:?}").as_str());
-                    cloned_context.leave_active_request().await;
+                    cloned_context.leave_active_request();
 
                     return Err(FetchError::SubrequestHttpError {
                         status_code: None,
@@ -359,7 +359,7 @@ async fn call_http(
                 if !content_type_str.contains(APPLICATION_JSON.essence_str())
                     && !content_type_str.contains(GRAPHQL_JSON_RESPONSE_HEADER_VALUE)
                 {
-                    cloned_context.leave_active_request().await;
+                    cloned_context.leave_active_request();
 
                     return if !parts.status.is_success() {
 
@@ -387,7 +387,7 @@ async fn call_http(
             .instrument(tracing::debug_span!("aggregate_response_data"))
             .await {
                 Err(err) => {
-                    cloned_context.leave_active_request().await;
+                    cloned_context.leave_active_request();
 
                     tracing::error!(fetch_error = format!("{err:?}").as_str());
 
@@ -400,7 +400,7 @@ async fn call_http(
                 }, Ok(body) => body,
             };
 
-            cloned_context.leave_active_request().await;
+            cloned_context.leave_active_request();
 
         Ok((parts, body))
     }.instrument(subgraph_req_span).await?;