Run sync operations in parallel

Summary:
## This stack

Track and down methods that are holding up the reactor and optimize them.

## This diff

This code is loading both sides of a diff from blobstore concurrently is the right thing. In practice, most of the cost comes from the Thrift deserialization, which is synchronous. Worse, `try_join` does not yield back to Tokio. The combination means blocking the reactor for 150-200ms or even longer.

With this diff we'll actually run them in parallel, saving both max poll time and quite likely wall clock time too.

Reviewed By: markbt

Differential Revision: D66761433

fbshipit-source-id: 5d44506a453697d7a45afe3bd811152ba0e21982

eden/mononoke/manifest/src/ops.rs

@@ -22,6 +22,7 @@ use futures::FutureExt;
 use futures::StreamExt;
 use futures::TryFutureExt;
 use futures::TryStreamExt;
+use futures_watchdog::WatchdogExt;
 use mononoke_types::path::MPath;
 use mononoke_types::NonRootMPath;
 
@@ -316,11 +317,16 @@ where
 
                     match input {
                         Diff::Changed(path, left, right) => {
-                            let (left_mf, right_mf) = future::try_join(
-                                left.load(ctx, &store),
-                                right.load(ctx, &other_store),
-                            )
-                            .await?;
+                            let l = tokio::spawn({
+                                cloned!(ctx, left, store);
+                                async move { left.load(&ctx, &store).watched(ctx.logger()).await }
+                            });
+                            let r = tokio::spawn({
+                                cloned!(ctx, right, other_store);
+                                async move { right.load(&ctx, &other_store).watched(ctx.logger()).await }
+                            });
+                            let (left_mf, right_mf) = future::try_join(l, r).await?;
+                            let (left_mf, right_mf) = (left_mf?, right_mf?);
 
                             let mut stream = left_mf.list(ctx, &store).await?;
                             while let Some((name, left)) = stream.try_next().await? {