Log slow poll times

Summary:
## This stack

We noticed that the poll time is pretty high on some async methods, and that is affecting other calls as well. Let's try to optimize that.

## This diff

Log the max poll time when it's excessive.

Reviewed By: mitrandir77

Differential Revision: D67135149

fbshipit-source-id: cea1d23ba27026731e958bf24048139d47decc7d

eden/mononoke/async_requests/worker/Cargo.toml

@@ -25,6 +25,7 @@ executor_lib = { version = "0.1.0", path = "../../cmdlib/sharding" }
 fbinit = { version = "0.2.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 futures = { version = "0.3.30", features = ["async-await", "compat"] }
 futures_stats = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
+futures_watchdog = { version = "0.1.0", path = "../../common/futures_watchdog" }
 hostname = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 megarepo_api = { version = "0.1.0", path = "../../megarepo_api" }
 megarepo_error = { version = "0.1.0", path = "../../megarepo_api/megarepo_error" }

eden/mononoke/async_requests/worker/src/methods.rs

@@ -28,6 +28,7 @@ use futures::future::BoxFuture;
 use futures::try_join;
 use futures::Future;
 use futures::FutureExt;
+use futures_watchdog::WatchdogExt;
 use megarepo_api::MegarepoApi;
 use megarepo_error::MegarepoError;
 use mononoke_api::ChangesetContext;
@@ -48,6 +49,8 @@ use scs_methods::specifiers::SpecifierExt;
 use source_control as thrift;
 use source_control::CommitSpecifier;
 
+const METHOD_MAX_POLL_TIME_MS: u64 = 100;
+
 #[cfg(not(fbcode_build))]
 pub async fn commit_sparse_profile_delta_impl(
     ctx: &CoreContext,
@@ -89,6 +92,7 @@ async fn megarepo_sync_changeset<R: MononokeRepo>(
             params.target.into_config_format(&megarepo_api.mononoke())?,
             target_location,
         )
+        .watched(ctx.logger())
         .await?
         .as_ref()
         .into();
@@ -247,30 +251,45 @@ pub(crate) async fn megarepo_async_request_compute<R: MononokeRepo>(
     match params.into() {
         async_requests_types_thrift::AsynchronousRequestParams::megarepo_add_target_params(params) => {
             Ok(megarepo_add_sync_target(ctx, megarepo_api, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("megarepo_add_sync_target")
                 .await
                 .map_err(|e| e.into())
                 .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::megarepo_add_branching_target_params(params) => {
             Ok(megarepo_add_branching_sync_target(ctx, megarepo_api, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("megarepo_add_branching_sync_target")
                 .await
                 .map_err(|e| e.into())
                 .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::megarepo_change_target_params(params) => {
             Ok(megarepo_change_target_config(ctx, megarepo_api, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("megarepo_change_target_config")
                 .await
                 .map_err(|e| e.into())
                 .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::megarepo_remerge_source_params(params) => {
             Ok(megarepo_remerge_source(ctx, megarepo_api, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("megarepo_remerge_source")
                 .await
                 .map_err(|e| e.into())
                 .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::megarepo_sync_changeset_params(params) => {
             Ok(megarepo_sync_changeset(ctx, megarepo_api, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("megarepo_sync_changeset")
                 .await
                 .map_err(|e| e.into())
                 .into())
@@ -282,10 +301,20 @@ pub(crate) async fn megarepo_async_request_compute<R: MononokeRepo>(
             }).into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::commit_sparse_profile_size_params(params) => {
-            Ok(commit_sparse_profile_size(ctx, mononoke, params).await.into())
+            Ok(commit_sparse_profile_size(ctx, mononoke, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("commit_sparse_profile_size")
+                .await
+                .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::commit_sparse_profile_delta_params(params) => {
-            Ok(commit_sparse_profile_delta(ctx, mononoke, params).await.into())
+            Ok(commit_sparse_profile_delta(ctx, mononoke, params)
+                .watched(ctx.logger())
+                .with_max_poll(METHOD_MAX_POLL_TIME_MS)
+                .with_label("commit_sparse_profile_delta")
+                .await
+                .into())
         }
         async_requests_types_thrift::AsynchronousRequestParams::UnknownField(union_tag) => {
              bail!(