Add a multiplex mode that doesn't update the sync queue

Summary:
Backfillers and other housekeeping processes can run so far ahead of the blobstore sync queue that we can't empty it from the healer task as fast as the backfillers can fill it.

Work around this by providing a new mode that background tasks can use to avoid filling the queue if all the blobstores are writing successfully. This has a side-effect of slowing background tasks to the speed of the slowest blobstore, instead of allowing them to run ahead at the speed of the fastest blobstore and relying on the healer ensuring that all blobs are present.

Future diffs will add this mode to appropriate tasks

Reviewed By: ikostia

Differential Revision: D22866818

fbshipit-source-id: a8762528bb3f6f11c0ec63e4a3c8dac08d0b4d8e

eden/mononoke/blobstore/multiplexedblob/src/base.rs

@@ -10,7 +10,7 @@ use blobstore::{Blobstore, BlobstoreGetData};
 use blobstore_stats::{record_get_stats, record_put_stats, OperationType};
 use blobstore_sync_queue::OperationKey;
 use cloned::cloned;
-use context::{CoreContext, PerfCounterType};
+use context::{CoreContext, PerfCounterType, SessionClass};
 use futures::{
     future::{join_all, select, BoxFuture, Either as FutureEither, FutureExt},
     stream::{FuturesUnordered, StreamExt, TryStreamExt},
@@ -318,15 +318,23 @@ fn spawn_stream_completion(s: impl StreamExt + Send + 'static) {
     tokio::spawn(s.for_each(|_| async {}));
 }
 
+/// Select the next item from one of two FuturesUnordered stream.
+/// With `consider_right` set to false, this is the same as `left.next().await.map(Either::Left)`.
+/// With `consider_right` set to true, this picks the first item to complete from either stream.
+/// The idea is that `left` contains your core work, and you always want to poll futures in that
+/// stream, while `right` contains failure recovery, and you only want to poll futures in that
+/// stream if you need to do failure recovery.
 async fn select_next<F1: Future, F2: Future>(
     left: &mut FuturesUnordered<F1>,
     right: &mut FuturesUnordered<F2>,
+    consider_right: bool,
 ) -> Option<Either<F1::Output, F2::Output>> {
     use Either::*;
+    let right_empty = !consider_right || right.is_empty();
     // Can't use a match block because that infers the wrong Send + Sync bounds for this future
-    if left.is_empty() && right.is_empty() {
+    if left.is_empty() && right_empty {
         None
-    } else if right.is_empty() {
+    } else if right_empty {
         left.next().await.map(Left)
     } else if left.is_empty() {
         right.next().await.map(Right)
@@ -368,6 +376,10 @@ impl Blobstore for MultiplexedBlobstoreBase {
         let write_order = Arc::new(AtomicUsize::new(0));
         let operation_key = OperationKey::gen();
         let mut needed_handlers: usize = self.minimum_successful_writes.into();
+        let run_handlers_on_success = match ctx.session().session_class() {
+            SessionClass::UserWaiting => true,
+            SessionClass::Background => false,
+        };
 
         let mut puts: FuturesUnordered<_> = self
             .blobstores
@@ -427,15 +439,23 @@ impl Blobstore for MultiplexedBlobstoreBase {
                     let mut put_errors = HashMap::new();
                     let mut handlers = FuturesUnordered::new();
 
-                    while let Some(result) = select_next(&mut puts, &mut handlers).await {
+                    while let Some(result) = select_next(
+                        &mut puts,
+                        &mut handlers,
+                        run_handlers_on_success || !put_errors.is_empty(),
+                    )
+                    .await
+                    {
                         use Either::*;
                         match result {
                             Left(Ok(handler)) => {
                                 handlers.push(handler);
                                 // All puts have succeeded, no errors - we're done
                                 if puts.is_empty() && put_errors.is_empty() {
-                                    // Spawn off the handlers to ensure that all writes are logged.
-                                    spawn_stream_completion(handlers);
+                                    if run_handlers_on_success {
+                                        // Spawn off the handlers to ensure that all writes are logged.
+                                        spawn_stream_completion(handlers);
+                                    }
                                     return Ok(());
                                 }
                             }
@@ -444,6 +464,8 @@ impl Blobstore for MultiplexedBlobstoreBase {
                             }
                             Right(Ok(())) => {
                                 needed_handlers = needed_handlers.saturating_sub(1);
+                                // Can only get here if at least one handler has been run, therefore need to ensure all handlers
+                                // run.
                                 if needed_handlers == 0 {
                                     // Handlers were successful. Spawn off remaining puts and handler
                                     // writes, then done

eden/mononoke/blobstore/multiplexedblob/src/test.rs

@@ -23,7 +23,7 @@ use blobstore_sync_queue::{
 };
 use bytes::Bytes;
 use cloned::cloned;
-use context::CoreContext;
+use context::{CoreContext, SessionClass};
 use fbinit::FacebookInit;
 use futures::{
     channel::oneshot,
@@ -1207,3 +1207,118 @@ async fn needed_writes_bad_config(fb: FacebookInit) {
         log.clear();
     }
 }
+
+#[fbinit::test]
+async fn no_handlers(fb: FacebookInit) {
+    let bs0 = Arc::new(Tickable::new());
+    let bs1 = Arc::new(Tickable::new());
+    let bs2 = Arc::new(Tickable::new());
+    let log = Arc::new(LogHandler::new());
+    let bs = MultiplexedBlobstoreBase::new(
+        MultiplexId::new(1),
+        vec![
+            (BlobstoreId::new(0), bs0.clone()),
+            (BlobstoreId::new(1), bs1.clone()),
+            (BlobstoreId::new(2), bs2.clone()),
+        ],
+        vec![],
+        nonzero!(1usize),
+        log.clone(),
+        ScubaSampleBuilder::with_discard(),
+        nonzero!(1u64),
+    );
+    let ctx = CoreContext::test_mock_class(fb, SessionClass::Background);
+
+    let clear = {
+        cloned!(bs0, bs1, bs2, log);
+        move || {
+            bs0.tick(None);
+            bs1.tick(None);
+            bs2.tick(None);
+            log.clear();
+        }
+    };
+
+    let k = String::from("k");
+    let v = make_value("v");
+
+    // Put succeeds once all blobstores have succeded. The handlers won't run
+    {
+        let mut fut = bs
+            .put(ctx.clone(), k.clone(), v.clone())
+            .map_err(|_| ())
+            .boxed();
+
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+
+        bs0.tick(None);
+        bs1.tick(None);
+        bs2.tick(None);
+        fut.await.expect("Put should have succeeded");
+
+        log.log
+            .with(|l| assert!(l.is_empty(), "Handlers ran, yet all blobstores succeeded"));
+        clear();
+    }
+
+    // Put is still in progress after one write, because no handlers have run
+    {
+        let mut fut = bs
+            .put(ctx.clone(), k.clone(), v.clone())
+            .map_err(|_| ())
+            .boxed();
+
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+
+        bs0.tick(None);
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+        log.log
+            .with(|l| assert!(l.is_empty(), "Handlers ran, yet put in progress"));
+
+        bs1.tick(None);
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+        log.log
+            .with(|l| assert!(l.is_empty(), "Handlers ran, yet put in progress"));
+
+        bs2.tick(None);
+        fut.await.expect("Put should have succeeded");
+        log.log
+            .with(|l| assert!(l.is_empty(), "Handlers ran, yet all blobstores succeeded"));
+
+        clear();
+    }
+
+    // Put succeeds despite errors, if the queue succeeds
+    {
+        let mut fut = bs
+            .put(ctx.clone(), k.clone(), v.clone())
+            .map_err(|_| ())
+            .boxed();
+
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+
+        bs0.tick(None);
+        assert_eq!(PollOnce::new(Pin::new(&mut fut)).await, Poll::Pending);
+        bs1.tick(Some("oops"));
+        fut.await.expect("Put should have succeeded");
+
+        log.log.with(|l| {
+            assert!(
+                l.len() == 1,
+                "Handlers did not run after a blobstore failure"
+            )
+        });
+        bs2.tick(None);
+        // Yield to let the spawned puts and handlers run
+        tokio::task::yield_now().await;
+
+        log.log.with(|l| {
+            assert!(
+                l.len() == 2,
+                "Handlers did not run for both successful blobstores"
+            )
+        });
+
+        clear();
+    }
+}

eden/mononoke/server/context/src/core.rs

@@ -17,7 +17,7 @@ use tracing::TraceContext;
 
 use crate::logging::{LoggingContainer, SamplingKey};
 use crate::perf_counters::PerfCounters;
-use crate::session::SessionContainer;
+use crate::session::{SessionClass, SessionContainer};
 
 #[derive(Clone)]
 pub struct CoreContext {
@@ -35,6 +35,18 @@ impl CoreContext {
     pub fn test_mock(fb: FacebookInit) -> Self {
         let session = SessionContainer::new_with_defaults(fb);
 
+        Self::test_mock_session(session)
+    }
+
+    pub fn test_mock_class(fb: FacebookInit, session_class: SessionClass) -> Self {
+        let session = SessionContainer::builder(fb)
+            .session_class(session_class)
+            .build();
+
+        Self::test_mock_session(session)
+    }
+
+    pub fn test_mock_session(session: SessionContainer) -> Self {
         let drain = default_drain().filter_level(Level::Debug).ignore_res();
         let logger = Logger::root(drain, o![]);
         session.new_context(logger, ScubaSampleBuilder::with_discard())

eden/mononoke/server/context/src/lib.rs

@@ -17,7 +17,9 @@ pub use crate::logging::{LoggingContainer, SamplingKey};
 #[cfg(not(fbcode_build))]
 pub use crate::oss::is_quicksand;
 pub use crate::perf_counters::{PerfCounterType, PerfCounters};
-pub use crate::session::{generate_session_id, SessionContainer, SessionContainerBuilder};
+pub use crate::session::{
+    generate_session_id, SessionClass, SessionContainer, SessionContainerBuilder,
+};
 
 mod core;
 #[cfg(fbcode_build)]

eden/mononoke/server/context/src/session/builder.rs

@@ -16,7 +16,7 @@ use std::net::IpAddr;
 use std::sync::Arc;
 use tracing::TraceContext;
 
-use super::{SessionContainer, SessionContainerInner};
+use super::{SessionClass, SessionContainer, SessionContainerInner};
 
 pub fn generate_session_id() -> SessionId {
     let s: String = thread_rng().sample_iter(&Alphanumeric).take(16).collect();
@@ -50,6 +50,7 @@ impl SessionContainerBuilder {
                 blobstore_write_limiter: None,
                 blobstore_read_limiter: None,
                 user_ip: None,
+                session_class: SessionClass::UserWaiting,
             },
         }
     }
@@ -101,4 +102,9 @@ impl SessionContainerBuilder {
         self.inner.user_ip = value.into();
         self
     }
+
+    pub fn session_class(mut self, value: SessionClass) -> Self {
+        self.inner.session_class = value;
+        self
+    }
 }

eden/mononoke/server/context/src/session/mod.rs

@@ -33,6 +33,17 @@ pub struct SessionContainer {
     inner: Arc<SessionContainerInner>,
 }
 
+/// Represents the reason this session is running
+#[derive(Clone, Copy)]
+pub enum SessionClass {
+    /// There is someone waiting for this session to complete.
+    UserWaiting,
+    /// The session is doing background work (e.g. backfilling).
+    /// Wherever reasonable, prefer to slow down and wait for work to complete
+    /// fully rather than pushing work out to other tasks.
+    Background,
+}
+
 struct SessionContainerInner {
     session_id: SessionId,
     trace: TraceContext,
@@ -44,6 +55,7 @@ struct SessionContainerInner {
     blobstore_write_limiter: Option<AsyncLimiter>,
     blobstore_read_limiter: Option<AsyncLimiter>,
     user_ip: Option<IpAddr>,
+    session_class: SessionClass,
 }
 
 impl SessionContainer {
@@ -139,4 +151,8 @@ impl SessionContainer {
     pub fn blobstore_write_limiter(&self) -> &Option<AsyncLimiter> {
         &self.inner.blobstore_write_limiter
     }
+
+    pub fn session_class(&self) -> SessionClass {
+        self.inner.session_class
+    }
 }