Stop passing MononokeApp to the worker.

Summary:
## This stack

Running the worker for all repos is causing OOMs; sharding should help.

## This diff

The `MononokeApp` instance is supposed to be a singleton and remain in the top context, not passed around. With the preparation from the previous diff, we can clean this up now.

Reviewed By: singhsrb

Differential Revision: D64410195

fbshipit-source-id: 90b6e27429f9ee765f70ba975e516fe75c4c1dc2

eden/mononoke/async_requests/async_requests_client/src/lib.rs

@@ -40,7 +40,7 @@ pub async fn build(
     ))
 }
 
-async fn open_sql_connection(
+pub async fn open_sql_connection(
     fb: FacebookInit,
     app: &MononokeApp,
 ) -> Result<SqlLongRunningRequestsQueue, Error> {
@@ -61,7 +61,10 @@ async fn open_sql_connection(
     }
 }
 
-async fn open_blobstore(fb: FacebookInit, app: &MononokeApp) -> Result<Arc<dyn Blobstore>, Error> {
+pub async fn open_blobstore(
+    fb: FacebookInit,
+    app: &MononokeApp,
+) -> Result<Arc<dyn Blobstore>, Error> {
     let config = app.repo_configs().common.async_requests_config.clone();
     if let Some(config) = config.blobstore {
         let options = app.blobstore_options();

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

@@ -89,12 +89,11 @@ fn main(fb: FacebookInit) -> Result<()> {
     } else {
         Some(repos)
     };
+    let queue = Arc::new(runtime.block_on(async_requests_client::build(fb, &app, filter_repos))?);
     let worker = runtime.block_on(worker::AsyncMethodRequestWorker::new(
-        app.fb,
-        &app,
         args.clone(),
         Arc::new(ctx),
-        filter_repos,
+        queue,
         mononoke,
         megarepo,
         will_exit.clone(),

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

@@ -18,7 +18,6 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::Context;
 use anyhow::Error;
 use anyhow::Result;
 use async_requests::types::AsynchronousRequestParams;
@@ -31,7 +30,6 @@ use async_trait::async_trait;
 use cloned::cloned;
 use context::CoreContext;
 use executor_lib::RepoShardedProcessExecutor;
-use fbinit::FacebookInit;
 use futures::future::abortable;
 use futures::future::select;
 use futures::future::Either;
@@ -44,8 +42,6 @@ use megarepo_api::MegarepoApi;
 use mononoke_api::Mononoke;
 use mononoke_api::MononokeRepo;
 use mononoke_api::Repo;
-use mononoke_api::RepositoryId;
-use mononoke_app::MononokeApp;
 use mononoke_types::Timestamp;
 use slog::debug;
 use slog::error;
@@ -85,23 +81,17 @@ pub struct AsyncMethodRequestWorker {
     mononoke: Arc<Mononoke<Repo>>,
     megarepo: Arc<MegarepoApi<Repo>>,
     name: String,
-    queue: AsyncMethodRequestQueue,
+    queue: Arc<AsyncMethodRequestQueue>,
     will_exit: Arc<AtomicBool>,
     limit: Option<usize>,
     concurrency_limit: usize,
 }
 
 impl AsyncMethodRequestWorker {
-    /// Creates a new tailer instance that's going to use provided megarepo API
-    /// The name argument should uniquely identify tailer instance and will be put
-    /// in the queue table so it's possible to find out which instance is working on
-    /// a given task (for debugging purposes).
     pub(crate) async fn new(
-        fb: FacebookInit,
-        app: &MononokeApp,
         args: Arc<AsyncRequestsWorkerArgs>,
         ctx: Arc<CoreContext>,
-        repos: Option<Vec<RepositoryId>>,
+        queue: Arc<AsyncMethodRequestQueue>,
         mononoke: Arc<Mononoke<Repo>>,
         megarepo: Arc<MegarepoApi<Repo>>,
         will_exit: Arc<AtomicBool>,
@@ -121,9 +111,6 @@ impl AsyncMethodRequestWorker {
             }
         };
 
-        let queue = async_requests_client::build(fb, app, repos)
-            .await
-            .context("acquiring the async requests queue")?;
         Ok(Self {
             ctx,
             mononoke,
@@ -199,7 +186,7 @@ impl AsyncMethodRequestWorker {
     pub fn request_stream(
         &self,
         ctx: &CoreContext,
-        queue: AsyncMethodRequestQueue,
+        queue: Arc<AsyncMethodRequestQueue>,
         will_exit: Arc<AtomicBool>,
     ) -> impl Stream<Item = Result<(RequestId, AsynchronousRequestParams), AsyncRequestsError>>
     {
@@ -218,7 +205,7 @@ impl AsyncMethodRequestWorker {
     fn request_stream_inner(
         ctx: CoreContext,
         claimed_by: ClaimedBy,
-        queue: AsyncMethodRequestQueue,
+        queue: Arc<AsyncMethodRequestQueue>,
         will_exit: Arc<AtomicBool>,
         sleep_time: Duration,
         abandoned_threshold_secs: i64,
@@ -489,7 +476,7 @@ mod test {
 
     #[mononoke::fbinit_test]
     async fn test_request_stream_simple(fb: FacebookInit) -> Result<(), Error> {
-        let q = AsyncMethodRequestQueue::new_test_in_memory().unwrap();
+        let q = Arc::new(AsyncMethodRequestQueue::new_test_in_memory().unwrap());
         let ctx = CoreContext::test_mock(fb);
 
         let params = thrift::MegarepoSyncChangesetParams {
@@ -529,7 +516,7 @@ mod test {
 
     #[mononoke::fbinit_test]
     async fn test_request_stream_clear_abandoned(fb: FacebookInit) -> Result<(), Error> {
-        let q = AsyncMethodRequestQueue::new_test_in_memory().unwrap();
+        let q = Arc::new(AsyncMethodRequestQueue::new_test_in_memory().unwrap());
         let ctx = CoreContext::test_mock(fb);
 
         let params = thrift::MegarepoSyncChangesetParams {