chore: buffer channels for better performance

.dockerignore

@@ -0,0 +1,2 @@
+target/
+**/target/

src/sink.rs

@@ -18,7 +18,7 @@ const DEFAULT_FB_SOCK_ADDR: &str = "/var/run/numaflow/fb-sink.sock";
 const DEFAULT_FB_SERVER_INFO_FILE: &str = "/var/run/numaflow/fb-sinker-server-info";
 const ENV_UD_CONTAINER_TYPE: &str = "NUMAFLOW_UD_CONTAINER_TYPE";
 const UD_CONTAINER_FB_SINK: &str = "fb-udsink";
-
+const DEFAULT_CHANNEL_SIZE: usize = 1000;
 /// Numaflow Sink Proto definitions.
 pub mod proto {
     tonic::include_proto!("sink.v1");
@@ -185,8 +185,8 @@ where
         let sink_handle = self.handler.clone();
         let cancellation_token = self.cancellation_token.clone();
         let shutdown_tx = self.shutdown_tx.clone();
-        // TODO: what should be the idle buffer size?
-        let (tx, rx) = mpsc::channel::<SinkRequest>(1);
+        // FIXME: we should be using the batch size as the channel size
+        let (tx, rx) = mpsc::channel::<SinkRequest>(DEFAULT_CHANNEL_SIZE);
 
         let reader_shutdown_tx = shutdown_tx.clone();
         // spawn a task to read messages from the stream and send them to the user's sink handle

src/source.rs

@@ -87,11 +87,12 @@ where
         let sr = request.into_inner().request.unwrap();
 
         // tx,rx pair for sending data over to user-defined source
-        let (stx, mut srx) = mpsc::channel::<Message>(1);
+        let (stx, mut srx) = mpsc::channel::<Message>(sr.num_records as usize);
         // tx,rx pair for gRPC response
-        let (tx, rx) = mpsc::channel::<Result<proto::ReadResponse, Status>>(1);
+        let (tx, rx) =
+            mpsc::channel::<Result<proto::ReadResponse, Status>>(sr.num_records as usize);
 
-        // start the ud-source rx asynchronously and start populating the gRPC response so it can be streamed to the gRPC client (numaflow).
+        // start the ud-source rx asynchronously and start populating the gRPC response, so it can be streamed to the gRPC client (numaflow).
         tokio::spawn(async move {
             while let Some(resp) = srx.recv().await {
                 tx.send(Ok(proto::ReadResponse {