refactor(meta-service): respond mget items in stream instead of in a vector

Since this commit, mget won't block the response until all items retrieved.
Now it create a stream and return at once. The read IO will happen when
the client receives the items.

And add logging about the statistics when a read stream is closed:

```
StreamElapsed: total: 104.458µs; items: 1, items/ms: 1, item_latency: 104.458µs; ReadRequest: ListKV(ListKVReq { prefix: "__fd_marked_deleted_index/" })
```

Author: drmingdrmer

Cargo.lock

@@ -196,8 +196,8 @@ impl SMV003 {
     }
 
     pub async fn get_maybe_expired_kv(&self, key: &str) -> Result<Option<SeqV>, io::Error> {
-        let view = self.data().to_readonly_view();
-        let got = view.get(UserKey::new(key.to_string())).await?;
+        let readonly_view = self.data().to_readonly_view();
+        let got = readonly_view.get(UserKey::new(key.to_string())).await?;
         let seqv = Into::<Option<SeqV>>::into(got);
         Ok(seqv)
     }

src/meta/raft-store/src/sm_v003/sm_v003.rs

@@ -29,6 +29,7 @@ use map_api::mvcc::ScopedViewReadonly;
 use seq_marked::SeqValue;
 use state_machine_api::UserKey;
 
+use crate::leveled_store::leveled_map::ReadonlyView;
 use crate::sm_v003::SMV003;
 use crate::testing::since_epoch_millis;
 use crate::utils::add_cooperative_yielding;
@@ -50,16 +51,13 @@ impl kvapi::KVApi for SMV003KVApi<'_> {
 
     async fn get_kv_stream(&self, keys: &[String]) -> Result<KVStream<Self::Error>, Self::Error> {
         let local_now_ms = since_epoch_millis();
+        let strm = readonly_view_get_kv_stream(
+            self.sm.data().to_readonly_view(),
+            keys.to_vec(),
+            local_now_ms,
+        );
 
-        let mut items = Vec::with_capacity(keys.len());
-
-        for k in keys {
-            let got = self.sm.get_maybe_expired_kv(k).await?;
-            let v = Self::non_expired(got, local_now_ms);
-            items.push(Ok(StreamItem::from((k.clone(), v))));
-        }
-
-        Ok(futures::stream::iter(items).boxed())
+        Ok(strm)
     }
 
     async fn list_kv(&self, prefix: &str) -> Result<KVStream<Self::Error>, Self::Error> {
@@ -100,3 +98,21 @@ impl SMV003KVApi<'_> {
         }
     }
 }
+
+/// A helper function that get many keys in stream.
+#[futures_async_stream::try_stream(boxed, ok = StreamItem, error = io::Error)]
+async fn readonly_view_get_kv_stream(
+    readonly_view: ReadonlyView,
+    keys: Vec<String>,
+    local_now_ms: u64,
+) {
+    for key in keys {
+        let got = readonly_view.get(UserKey::new(key.clone())).await?;
+
+        let seqv = Into::<Option<SeqV>>::into(got);
+
+        let non_expired = SMV003KVApi::non_expired(seqv, local_now_ms);
+
+        yield StreamItem::from((key, non_expired));
+    }
+}

src/meta/raft-store/src/sm_v003/sm_v003_kv_api.rs

@@ -49,6 +49,7 @@ log = { workspace = true }
 logcall = { workspace = true }
 map-api = { workspace = true }
 maplit = { workspace = true }
+pin-project = { workspace = true}
 poem = { workspace = true }
 prometheus-client = { workspace = true }
 prost = { workspace = true }

src/meta/service/Cargo.toml

@@ -15,8 +15,10 @@
 use std::future;
 use std::io;
 use std::pin::Pin;
+use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
 use std::sync::Weak;
+use std::time::Instant;
 use std::time::SystemTime;
 
 use arrow_flight::BasicAuth;
@@ -83,6 +85,7 @@ use watcher::util::try_forward;
 use watcher::watch_stream::WatchStream;
 use watcher::watch_stream::WatchStreamSender;
 
+use crate::api::grpc::OnCompleteStream;
 use crate::message::ForwardRequest;
 use crate::message::ForwardRequestBody;
 use crate::meta_service::watcher::DispatcherHandle;
@@ -327,17 +330,36 @@ impl MetaService for MetaServiceImpl {
 
         let req: MetaGrpcReadReq = GrpcHelper::parse_req(request)?;
         let req_typ = req.type_name();
+        let req_str = format!("ReadRequest: {:?}", req);
 
         ThreadTracker::tracking_future(async move {
             let _guard = InFlightRead::guard();
+            let start = Instant::now();
             let (endpoint, strm) = self.handle_kv_read_v1(req).in_span(root).await?;
 
-            let strm = strm
-                .map(move |item| {
-                    network_metrics::incr_stream_sent_item(req_typ);
-                    item
-                })
-                .boxed();
+            // Counter to track total items sent
+            let count = Arc::new(AtomicU64::new(0));
+            let count2 = count.clone();
+
+            let strm = strm.map(move |item| {
+                network_metrics::incr_stream_sent_item(req_typ);
+                count2.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                item
+            });
+
+            // Log the total time and item count when the stream is finished.
+            let strm = OnCompleteStream::new(strm, move || {
+                let total = start.elapsed();
+                let total_items = count.load(std::sync::atomic::Ordering::Relaxed);
+                let items_per_ms = total_items / (total.as_millis() as u64 + 1);
+                let latency = total / (total_items.max(1) as u32);
+                info!(
+                    "StreamElapsed: total: {:?}; items: {}, items/ms: {}, item_latency: {:?}; {}",
+                    total, total_items, items_per_ms, latency, req_str
+                );
+            });
+
+            let strm = strm.boxed();
 
             let mut resp = Response::new(strm);
             GrpcHelper::add_response_meta_leader(&mut resp, endpoint.as_ref());

src/meta/service/src/api/grpc/grpc_service.rs

@@ -13,3 +13,47 @@
 // limitations under the License.
 
 pub mod grpc_service;
+
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+use futures::Stream;
+use pin_project::pin_project;
+
+#[pin_project]
+pub(crate) struct OnCompleteStream<S> {
+    #[pin]
+    stream: S,
+    callback: Option<Box<dyn FnOnce() + Send + 'static>>,
+}
+
+impl<S> OnCompleteStream<S> {
+    pub fn new(stream: S, callback: impl FnOnce() + Send + 'static) -> Self {
+        Self {
+            stream,
+            callback: Some(Box::new(callback)),
+        }
+    }
+}
+
+impl<S: Stream> Stream for OnCompleteStream<S> {
+    type Item = S::Item;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.project();
+
+        let res = this.stream.poll_next(cx);
+
+        match &res {
+            Poll::Ready(None) => {
+                if let Some(callback) = this.callback.take() {
+                    callback();
+                }
+            }
+            _ => {}
+        }
+
+        res
+    }
+}

src/meta/service/src/api/grpc/mod.rs

@@ -133,16 +133,9 @@ impl Handler<MetaGrpcReadReq> for MetaLeader<'_> {
 
             MetaGrpcReadReq::MGetKV(req) => {
                 // safe unwrap(): Infallible
-                let values = kv_api.mget_kv(&req.keys).await.unwrap();
+                let strm = kv_api.get_kv_stream(&req.keys).await.unwrap();
 
-                let kv_iter = req
-                    .keys
-                    .clone()
-                    .into_iter()
-                    .zip(values)
-                    .map(|(k, v)| Ok(StreamItem::from((k, v))));
-
-                let strm = futures::stream::iter(kv_iter);
+                let strm = strm.map_err(|e| Status::internal(e.to_string()));
 
                 Ok(strm.boxed())
             }

src/meta/service/src/meta_service/meta_leader.rs

@@ -1393,7 +1393,7 @@ impl MetaNode {
                 Err(e) => MetaOperationError::ForwardToLeader(e),
             };
 
-            // If needs to forward, deal with it. Otherwise return the unhandlable error.
+            // If it needs to forward, deal with it. Otherwise, return the unhandlable error.
             let to_leader = match op_err {
                 MetaOperationError::ForwardToLeader(err) => err,
                 MetaOperationError::DataError(d_err) => {