revisionstore: remove unbounded scmstore queue fallback

Summary: Remove the experimental unbounded scmstore result queue escape hatch now that bounded queues have been running without issues. FileStore and TreeStore fetches always use their bounded result channels.

Reviewed By: zzl0

Differential Revision: D105911075

fbshipit-source-id: 3a7e78dcdbf2e829ba60e8688b31a1905851f394

Author: muirdm

eden/scm/lib/revisionstore/src/scmstore/builder.rs

@@ -359,10 +359,6 @@ impl<'a> FileStoreBuilder<'a> {
 
             progress_bar: AggregatingProgressBar::new("fetching from ScmStore", "files"),
 
-            unbounded_queue: self
-                .config
-                .get_or_default("experimental", "unbounded-scmstore-queue")?,
-
             lfs_buffer_in_memory: self
                 .config
                 .get_or_default("experimental", "lfs-buffer-in-memory")?,
@@ -713,9 +709,6 @@ impl<'a> TreeStoreBuilder<'a> {
             flush_on_drop: true,
             format,
             progress_bar: AggregatingProgressBar::new("fetching from ScmStore", "trees"),
-            unbounded_queue: self
-                .config
-                .get_or_default("experimental", "unbounded-scmstore-queue")?,
             verify_hash,
             restricted_tree_mode,
             acl_check_cache: new_acl_check_cache(),

eden/scm/lib/revisionstore/src/scmstore/file.rs

@@ -25,7 +25,6 @@ use anyhow::bail;
 use anyhow::ensure;
 use blob::Blob;
 use flume::bounded;
-use flume::unbounded;
 use metrics::FILE_STORE_FETCH_METRICS;
 use minibytes::Bytes;
 use parking_lot::Mutex;
@@ -106,9 +105,6 @@ pub struct FileStore {
     // threads (so that only a single progress bar shows up to the user).
     pub(crate) progress_bar: Arc<AggregatingProgressBar>,
 
-    // Temporary escape hatch to disable bounding of queue.
-    pub(crate) unbounded_queue: bool,
-
     // Temporary escape hatch to disable streaming of LFS data to caches.
     pub(crate) lfs_buffer_in_memory: bool,
 
@@ -195,14 +191,9 @@ impl FileStore {
 
         let bar = self.progress_bar.create_or_extend_local(0);
 
-        let (found_tx, found_rx) = if self.unbounded_queue {
-            // Escape hatch in case something goes wrong with bounding.
-            unbounded()
-        } else {
-            // Bound channel size so we don't use unlimited memory queueing up file content
-            // when the consumer is consuming slower than we are fetching.
-            bounded(RESULT_QUEUE_SIZE)
-        };
+        // Bound channel size so we don't use unlimited memory queueing up file content
+        // when the consumer is consuming slower than we are fetching.
+        let (found_tx, found_rx) = bounded(RESULT_QUEUE_SIZE);
 
         let indexedlog_cache = self.indexedlog_cache.clone();
 
@@ -555,8 +546,6 @@ impl FileStore {
 
             progress_bar: AggregatingProgressBar::new("", ""),
 
-            unbounded_queue: false,
-
             lfs_buffer_in_memory: false,
 
             max_fetch_count: Default::default(),
@@ -606,8 +595,6 @@ impl FileStore {
 
             progress_bar: self.progress_bar.clone(),
 
-            unbounded_queue: self.unbounded_queue,
-
             lfs_buffer_in_memory: self.lfs_buffer_in_memory,
 
             max_fetch_count: self.max_fetch_count.clone(),

eden/scm/lib/revisionstore/src/scmstore/tree.rs

@@ -35,7 +35,6 @@ use edenapi_types::FileAuxData;
 use edenapi_types::TreeAuxData;
 use fetch::FetchState;
 use flume::bounded;
-use flume::unbounded;
 use metrics::TREE_STORE_FETCH_METRICS;
 use minibytes::Bytes;
 use moka::sync::Cache;
@@ -182,9 +181,6 @@ pub struct TreeStore {
     // threads (so that only a single progress bar shows up to the user).
     pub(crate) progress_bar: Arc<AggregatingProgressBar>,
 
-    // Temporary escape hatch to disable bounding of queue.
-    pub(crate) unbounded_queue: bool,
-
     pub(crate) verify_hash: bool,
 
     pub restricted_tree_mode: RestrictedTreeMode,
@@ -365,14 +361,9 @@ impl TreeStore {
 
         let bar = self.progress_bar.create_or_extend_local(0);
 
-        let (found_tx, found_rx) = if self.unbounded_queue {
-            // Escape hatch in case something goes wrong with bounding.
-            unbounded()
-        } else {
-            // Bound channel size so we don't use unlimited memory queueing up file content
-            // when the consumer is consuming slower than we are fetching.
-            bounded(RESULT_QUEUE_SIZE)
-        };
+        // Bound channel size so we don't use unlimited memory queueing up file content
+        // when the consumer is consuming slower than we are fetching.
+        let (found_tx, found_rx) = bounded(RESULT_QUEUE_SIZE);
 
         let indexedlog_cache = self.indexedlog_cache.clone();
         let aux_cache = self.filestore.as_ref().and_then(|fs| fs.aux_cache.clone());
@@ -679,7 +670,6 @@ impl TreeStore {
             prefetch_tree_parents: false,
             format: SerializationFormat::Hg,
             progress_bar: AggregatingProgressBar::new("", ""),
-            unbounded_queue: false,
             verify_hash: true,
             restricted_tree_mode: RestrictedTreeMode::Disabled,
             acl_check_cache: new_acl_check_cache(),
@@ -761,7 +751,6 @@ impl TreeStore {
             prefetch_tree_parents: false,
             format: self.format(),
             progress_bar: self.progress_bar.clone(),
-            unbounded_queue: self.unbounded_queue,
             verify_hash: self.verify_hash,
             restricted_tree_mode: self.restricted_tree_mode,
             acl_check_cache: self.acl_check_cache.clone(),

eden/scm/lib/revisionstore/src/trait_impls.rs

@@ -387,7 +387,6 @@ mod tests {
             flush_on_drop: false,
             format: SerializationFormat::Hg,
             progress_bar: progress_model::AggregatingProgressBar::new("", ""),
-            unbounded_queue: false,
             lfs_buffer_in_memory: false,
             max_fetch_count: Default::default(),
         })