Merge branch 'read-ahead' of github.com:numaproj/numaflow into read-a…

…head

rust/numaflow-core/src/config/components.rs

@@ -17,6 +17,8 @@ pub(crate) mod source {
 
     #[derive(Debug, Clone, PartialEq)]
     pub(crate) struct SourceConfig {
+        /// for high-throughput use-cases we read-ahead the next batch before the previous batch has
+        /// been acked (or completed). For most cases it should be set to false.
         pub(crate) read_ahead: bool,
         pub(crate) source_type: SourceType,
     }

rust/numaflow-core/src/source.rs

@@ -249,6 +249,8 @@ impl Source {
         let handle = tokio::spawn(async move {
             let mut processed_msgs_count: usize = 0;
             let mut last_logged_at = time::Instant::now();
+            // this semaphore is used only if read-ahead is disabled. we hold this semaphore to
+            // make sure we can read only if the current inflight ones are ack'ed.
             let semaphore = Arc::new(Semaphore::new(1));
 
             loop {