[testnet] backport multi_get changes to MapView and protocol code (#4789, #4788, #4634) (#4804)

Backport #4789, #4788, #4634

## Motivation

The `multi_get` was introduced in `MapView`, which provides some
improvement to the protocol code.
The most affected is maybe `block_hashes`, which showed up as a slow
part in the benchmarks.

## Proposal

The backport is clear.

## Test Plan

The CI.

## Release Plan

The backport to Conway is the point.

## Links

None

Author: MathieuDutSik

linera-chain/src/chain.rs

@@ -822,32 +822,44 @@ where
         }
 
         let recipients = block_execution_tracker.recipients();
-        let mut previous_message_blocks = BTreeMap::new();
-        for recipient in recipients {
-            if let Some(height) = previous_message_blocks_view.get(&recipient).await? {
-                let hash = confirmed_log
-                    .get(usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?)
-                    .await?
-                    .ok_or_else(|| {
-                        ChainError::InternalError("missing entry in confirmed_log".into())
-                    })?;
-                previous_message_blocks.insert(recipient, (hash, height));
+        let heights = previous_message_blocks_view.multi_get(&recipients).await?;
+        let mut recipient_heights = Vec::new();
+        let mut indices = Vec::new();
+        for (height, recipient) in heights.into_iter().zip(recipients) {
+            if let Some(height) = height {
+                let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
+                indices.push(index);
+                recipient_heights.push((recipient, height));
             }
         }
+        let hashes = confirmed_log.multi_get(indices).await?;
+        let mut previous_message_blocks = BTreeMap::new();
+        for (hash, (recipient, height)) in hashes.into_iter().zip(recipient_heights) {
+            let hash = hash.ok_or_else(|| {
+                ChainError::InternalError("missing entry in confirmed_log".into())
+            })?;
+            previous_message_blocks.insert(recipient, (hash, height));
+        }
 
         let streams = block_execution_tracker.event_streams();
-        let mut previous_event_blocks = BTreeMap::new();
-        for stream in streams {
-            if let Some(height) = previous_event_blocks_view.get(&stream).await? {
-                let hash = confirmed_log
-                    .get(usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?)
-                    .await?
-                    .ok_or_else(|| {
-                        ChainError::InternalError("missing entry in confirmed_log".into())
-                    })?;
-                previous_event_blocks.insert(stream, (hash, height));
+        let heights = previous_event_blocks_view.multi_get(&streams).await?;
+        let mut stream_heights = Vec::new();
+        let mut indices = Vec::new();
+        for (stream, height) in streams.into_iter().zip(heights) {
+            if let Some(height) = height {
+                let index = usize::try_from(height.0).map_err(|_| ArithmeticError::Overflow)?;
+                indices.push(index);
+                stream_heights.push((stream, height));
             }
         }
+        let hashes = confirmed_log.multi_get(indices).await?;
+        let mut previous_event_blocks = BTreeMap::new();
+        for (hash, (stream, height)) in hashes.into_iter().zip(stream_heights) {
+            let hash = hash.ok_or_else(|| {
+                ChainError::InternalError("missing entry in confirmed_log".into())
+            })?;
+            previous_event_blocks.insert(stream, (hash, height));
+        }
 
         let state_hash = {
             #[cfg(with_metrics)]
@@ -1068,10 +1080,17 @@ where
             Vec::new()
         };
         // Everything after (including) next_height in preprocessed_blocks if we have it.
-        for height in start.max(next_height).0..=end.0 {
-            if let Some(hash) = self.preprocessed_blocks.get(&BlockHeight(height)).await? {
-                hashes.push(hash);
-            }
+        let block_heights = (start.max(next_height).0..=end.0)
+            .map(BlockHeight)
+            .collect::<Vec<_>>();
+        for hash in self
+            .preprocessed_blocks
+            .multi_get(&block_heights)
+            .await?
+            .into_iter()
+            .flatten()
+        {
+            hashes.push(hash);
         }
         Ok(hashes)
     }

linera-views/src/views/map_view.rs

@@ -1155,6 +1155,44 @@ where
         self.map.get(&short_key).await
     }
 
+    /// Reads values at given positions, if any.
+    /// ```rust
+    /// # tokio_test::block_on(async {
+    /// # use linera_views::context::MemoryContext;
+    /// # use linera_views::map_view::MapView;
+    /// # use linera_views::views::View;
+    /// # let context = MemoryContext::new_for_testing(());
+    /// let mut map: MapView<_, u32, _> = MapView::load(context).await.unwrap();
+    /// map.insert(&(37 as u32), String::from("Hello"));
+    /// map.insert(&(49 as u32), String::from("Bonjour"));
+    /// assert_eq!(
+    ///     map.multi_get(&[37 as u32, 49 as u32, 64 as u32])
+    ///         .await
+    ///         .unwrap(),
+    ///     [
+    ///         Some(String::from("Hello")),
+    ///         Some(String::from("Bonjour")),
+    ///         None
+    ///     ]
+    /// );
+    /// assert_eq!(map.get(&(34 as u32)).await.unwrap(), None);
+    /// # })
+    /// ```
+    pub async fn multi_get<'a, Q>(
+        &self,
+        indices: impl IntoIterator<Item = &'a Q>,
+    ) -> Result<Vec<Option<V>>, ViewError>
+    where
+        I: Borrow<Q>,
+        Q: Serialize + 'a,
+    {
+        let short_keys = indices
+            .into_iter()
+            .map(|index| BaseKey::derive_short_key(index))
+            .collect::<Result<_, _>>()?;
+        self.map.multi_get(short_keys).await
+    }
+
     /// Obtains a mutable reference to a value at a given position if available
     /// ```rust
     /// # tokio_test::block_on(async {
@@ -1636,6 +1674,40 @@ where
         self.map.get(&short_key).await
     }
 
+    /// Read values at several positions, if any.
+    /// ```rust
+    /// # tokio_test::block_on(async {
+    /// # use linera_views::context::MemoryContext;
+    /// # use linera_views::map_view::CustomMapView;
+    /// # use linera_views::views::View;
+    /// # let context = MemoryContext::new_for_testing(());
+    /// let mut map: CustomMapView<MemoryContext<()>, u128, String> =
+    ///     CustomMapView::load(context).await.unwrap();
+    /// map.insert(&(34 as u128), String::from("Hello"));
+    /// map.insert(&(12 as u128), String::from("Hi"));
+    /// assert_eq!(
+    ///     map.multi_get(&[34 as u128, 12 as u128, 89 as u128])
+    ///         .await
+    ///         .unwrap(),
+    ///     [Some(String::from("Hello")), Some(String::from("Hi")), None]
+    /// );
+    /// # })
+    /// ```
+    pub async fn multi_get<'a, Q>(
+        &self,
+        indices: impl IntoIterator<Item = &'a Q>,
+    ) -> Result<Vec<Option<V>>, ViewError>
+    where
+        I: Borrow<Q>,
+        Q: CustomSerialize + 'a,
+    {
+        let short_keys = indices
+            .into_iter()
+            .map(|index| index.to_custom_bytes())
+            .collect::<Result<_, _>>()?;
+        self.map.multi_get(short_keys).await
+    }
+
     /// Obtains a mutable reference to a value at a given position if available
     /// ```rust
     /// # tokio_test::block_on(async {
@@ -2108,15 +2180,12 @@ mod graphql {
                 self.indices().await?
             };
 
-            let mut values = vec![];
-            for key in keys {
-                values.push(Entry {
-                    value: self.get(&key).await?,
-                    key,
-                })
-            }
-
-            Ok(values)
+            let values = self.multi_get(&keys).await?;
+            Ok(values
+                .into_iter()
+                .zip(keys)
+                .map(|(value, key)| Entry { value, key })
+                .collect())
         }
     }
 
@@ -2184,15 +2253,12 @@ mod graphql {
                 self.indices().await?
             };
 
-            let mut values = vec![];
-            for key in keys {
-                values.push(Entry {
-                    value: self.get(&key).await?,
-                    key,
-                })
-            }
-
-            Ok(values)
+            let values = self.multi_get(&keys).await?;
+            Ok(values
+                .into_iter()
+                .zip(keys)
+                .map(|(value, key)| Entry { value, key })
+                .collect())
         }
     }
 }