linera_views/views/reentrant_collection_view.rs
1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5 borrow::Borrow,
6 collections::{btree_map, BTreeMap},
7 io::Write,
8 marker::PhantomData,
9 mem,
10 ops::Deref,
11 sync::Arc,
12};
13
14use allocative::{Allocative, Key, Visitor};
15use async_lock::{RwLock, RwLockReadGuardArc, RwLockWriteGuardArc};
16#[cfg(with_metrics)]
17use linera_base::prometheus_util::MeasureLatency as _;
18use serde::{de::DeserializeOwned, Serialize};
19
20use crate::{
21 batch::Batch,
22 common::{CustomSerialize, HasherOutput, SliceExt as _, Update},
23 context::{BaseKey, Context},
24 hashable_wrapper::WrappedHashableContainerView,
25 historical_hash_wrapper::HistoricallyHashableView,
26 store::ReadableKeyValueStore as _,
27 views::{ClonableView, HashableView, Hasher, ReplaceContext, View, ViewError, MIN_VIEW_TAG},
28};
29
30#[cfg(with_metrics)]
31mod metrics {
32 use std::sync::LazyLock;
33
34 use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram_vec};
35 use prometheus::HistogramVec;
36
37 /// The runtime of hash computation
38 pub static REENTRANT_COLLECTION_VIEW_HASH_RUNTIME: LazyLock<HistogramVec> =
39 LazyLock::new(|| {
40 register_histogram_vec(
41 "reentrant_collection_view_hash_runtime",
42 "ReentrantCollectionView hash runtime",
43 &[],
44 exponential_bucket_latencies(5.0),
45 )
46 });
47}
48
49/// A read-only accessor for a particular subview in a [`ReentrantCollectionView`].
50#[derive(Debug)]
51pub struct ReadGuardedView<T>(RwLockReadGuardArc<T>);
52
53impl<T> std::ops::Deref for ReadGuardedView<T> {
54 type Target = T;
55 fn deref(&self) -> &T {
56 self.0.deref()
57 }
58}
59
60/// A read-write accessor for a particular subview in a [`ReentrantCollectionView`].
61#[derive(Debug)]
62pub struct WriteGuardedView<T>(RwLockWriteGuardArc<T>);
63
64impl<T> std::ops::Deref for WriteGuardedView<T> {
65 type Target = T;
66 fn deref(&self) -> &T {
67 self.0.deref()
68 }
69}
70
71impl<T> std::ops::DerefMut for WriteGuardedView<T> {
72 fn deref_mut(&mut self) -> &mut T {
73 self.0.deref_mut()
74 }
75}
76
77/// A view that supports accessing a collection of views of the same kind, indexed by `Vec<u8>`,
78/// possibly several subviews at a time.
79#[derive(Debug)]
80pub struct ReentrantByteCollectionView<C, W> {
81 /// The view [`Context`].
82 context: C,
83 /// If the current persisted data will be completely erased and replaced on the next flush.
84 delete_storage_first: bool,
85 /// Entries that may have staged changes.
86 updates: BTreeMap<Vec<u8>, Update<Arc<RwLock<W>>>>,
87}
88
89impl<C, W: Allocative> Allocative for ReentrantByteCollectionView<C, W> {
90 fn visit<'a, 'b: 'a>(&self, visitor: &'a mut Visitor<'b>) {
91 let name = Key::new("ReentrantByteCollectionView");
92 let size = mem::size_of::<Self>();
93 let mut visitor = visitor.enter(name, size);
94
95 for (k, v) in &self.updates {
96 let key_name = Key::new("key");
97 visitor.visit_field(key_name, k);
98 match v {
99 Update::Removed => {
100 let key = Key::new("update_removed");
101 visitor.visit_field(key, &());
102 }
103 Update::Set(v) => {
104 if let Some(v) = v.try_read() {
105 let key = Key::new("update_set");
106 visitor.visit_field(key, v.deref());
107 }
108 }
109 }
110 }
111 visitor.exit();
112 }
113}
114
115impl<W, C2> ReplaceContext<C2> for ReentrantByteCollectionView<W::Context, W>
116where
117 W: View + ReplaceContext<C2>,
118 C2: Context,
119{
120 type Target = ReentrantByteCollectionView<C2, <W as ReplaceContext<C2>>::Target>;
121
122 async fn with_context(
123 &mut self,
124 ctx: impl FnOnce(&Self::Context) -> C2 + Clone,
125 ) -> Self::Target {
126 let mut updates: BTreeMap<_, Update<Arc<RwLock<W::Target>>>> = BTreeMap::new();
127 for (key, update) in &self.updates {
128 let new_value = match update {
129 Update::Removed => Update::Removed,
130 Update::Set(x) => Update::Set(Arc::new(RwLock::new(
131 x.write().await.with_context(ctx.clone()).await,
132 ))),
133 };
134 updates.insert(key.clone(), new_value);
135 }
136 ReentrantByteCollectionView {
137 context: ctx(self.context()),
138 delete_storage_first: self.delete_storage_first,
139 updates,
140 }
141 }
142}
143
144/// We need to find new base keys in order to implement the collection view.
145/// We do this by appending a value to the base key.
146///
147/// Sub-views in a collection share a common key prefix, like in other view types. However,
148/// just concatenating the shared prefix with sub-view keys makes it impossible to distinguish if a
149/// given key belongs to a child sub-view or a grandchild sub-view (consider for example if a
150/// collection is stored inside the collection).
151#[repr(u8)]
152enum KeyTag {
153 /// Prefix for specifying an index and serves to indicate the existence of an entry in the collection.
154 Index = MIN_VIEW_TAG,
155 /// Prefix for specifying as the prefix for the sub-view.
156 Subview,
157}
158
159impl<W: View> View for ReentrantByteCollectionView<W::Context, W> {
160 const NUM_INIT_KEYS: usize = 0;
161
162 type Context = W::Context;
163
164 fn context(&self) -> &Self::Context {
165 &self.context
166 }
167
168 fn pre_load(_context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
169 Ok(Vec::new())
170 }
171
172 fn post_load(context: Self::Context, _values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
173 Ok(Self {
174 context,
175 delete_storage_first: false,
176 updates: BTreeMap::new(),
177 })
178 }
179
180 fn rollback(&mut self) {
181 self.delete_storage_first = false;
182 self.updates.clear();
183 }
184
185 async fn has_pending_changes(&self) -> bool {
186 if self.delete_storage_first {
187 return true;
188 }
189 !self.updates.is_empty()
190 }
191
192 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
193 let mut delete_view = false;
194 if self.delete_storage_first {
195 delete_view = true;
196 batch.delete_key_prefix(self.context.base_key().bytes.clone());
197 for (index, update) in &self.updates {
198 if let Update::Set(view) = update {
199 let view = view
200 .try_read()
201 .ok_or_else(|| ViewError::TryLockError(index.clone()))?;
202 view.pre_save(batch)?;
203 self.add_index(batch, index);
204 delete_view = false;
205 }
206 }
207 } else {
208 for (index, update) in &self.updates {
209 match update {
210 Update::Set(view) => {
211 let view = view
212 .try_read()
213 .ok_or_else(|| ViewError::TryLockError(index.clone()))?;
214 view.pre_save(batch)?;
215 self.add_index(batch, index);
216 }
217 Update::Removed => {
218 let key_subview = self.get_subview_key(index);
219 let key_index = self.get_index_key(index);
220 batch.delete_key(key_index);
221 batch.delete_key_prefix(key_subview);
222 }
223 }
224 }
225 }
226 Ok(delete_view)
227 }
228
229 fn post_save(&mut self) {
230 for (_index, update) in mem::take(&mut self.updates) {
231 if let Update::Set(view) = update {
232 let mut view = view.try_write().expect("pre_save was called before");
233 view.post_save();
234 }
235 }
236 self.delete_storage_first = false;
237 }
238
239 fn clear(&mut self) {
240 self.delete_storage_first = true;
241 self.updates.clear();
242 }
243}
244
245impl<W: ClonableView> ClonableView for ReentrantByteCollectionView<W::Context, W> {
246 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
247 let cloned_updates = self
248 .updates
249 .iter()
250 .map(|(key, value)| {
251 let cloned_value = match value {
252 Update::Removed => Update::Removed,
253 Update::Set(view_lock) => {
254 let mut view = view_lock
255 .try_write()
256 .ok_or_else(|| ViewError::TryLockError(key.clone()))?;
257 Update::Set(Arc::new(RwLock::new(view.clone_unchecked()?)))
258 }
259 };
260 Ok::<_, ViewError>((key.clone(), cloned_value))
261 })
262 .collect::<Result<_, _>>()?;
263
264 Ok(ReentrantByteCollectionView {
265 context: self.context.clone(),
266 delete_storage_first: self.delete_storage_first,
267 updates: cloned_updates,
268 })
269 }
270}
271
272impl<C: Context, W> ReentrantByteCollectionView<C, W> {
273 fn get_index_key(&self, index: &[u8]) -> Vec<u8> {
274 self.context
275 .base_key()
276 .base_tag_index(KeyTag::Index as u8, index)
277 }
278
279 fn get_subview_key(&self, index: &[u8]) -> Vec<u8> {
280 self.context
281 .base_key()
282 .base_tag_index(KeyTag::Subview as u8, index)
283 }
284
285 fn add_index(&self, batch: &mut Batch, index: &[u8]) {
286 let key = self.get_index_key(index);
287 batch.put_key_value_bytes(key, vec![]);
288 }
289}
290
291impl<W: View> ReentrantByteCollectionView<W::Context, W> {
292 /// Reads the view and if missing returns the default view
293 async fn wrapped_view(
294 context: &W::Context,
295 delete_storage_first: bool,
296 short_key: &[u8],
297 ) -> Result<Arc<RwLock<W>>, ViewError> {
298 let key = context
299 .base_key()
300 .base_tag_index(KeyTag::Subview as u8, short_key);
301 let context = context.clone_with_base_key(key);
302 // Obtain a view and set its pending state to the default (e.g. empty) state
303 let view = if delete_storage_first {
304 W::new(context)?
305 } else {
306 W::load(context).await?
307 };
308 Ok(Arc::new(RwLock::new(view)))
309 }
310
311 /// Load the view and insert it into the updates if needed.
312 /// If the entry is missing, then it is set to default.
313 async fn try_load_view_mut(&mut self, short_key: &[u8]) -> Result<Arc<RwLock<W>>, ViewError> {
314 use btree_map::Entry::*;
315 Ok(match self.updates.entry(short_key.to_owned()) {
316 Occupied(mut entry) => match entry.get_mut() {
317 Update::Set(view) => view.clone(),
318 entry @ Update::Removed => {
319 let wrapped_view = Self::wrapped_view(&self.context, true, short_key).await?;
320 *entry = Update::Set(wrapped_view.clone());
321 wrapped_view
322 }
323 },
324 Vacant(entry) => {
325 let wrapped_view =
326 Self::wrapped_view(&self.context, self.delete_storage_first, short_key).await?;
327 entry.insert(Update::Set(wrapped_view.clone()));
328 wrapped_view
329 }
330 })
331 }
332
333 /// Load the view from the update is available.
334 /// If missing, then the entry is loaded from storage and if
335 /// missing there an error is reported.
336 async fn try_load_view(&self, short_key: &[u8]) -> Result<Option<Arc<RwLock<W>>>, ViewError> {
337 Ok(if let Some(entry) = self.updates.get(short_key) {
338 match entry {
339 Update::Set(view) => Some(view.clone()),
340 _entry @ Update::Removed => None,
341 }
342 } else if self.delete_storage_first {
343 None
344 } else {
345 let key_index = self
346 .context
347 .base_key()
348 .base_tag_index(KeyTag::Index as u8, short_key);
349 if self.context.store().contains_key(&key_index).await? {
350 let view = Self::wrapped_view(&self.context, false, short_key).await?;
351 Some(view)
352 } else {
353 None
354 }
355 })
356 }
357
358 /// Loads a subview for the data at the given index in the collection. If an entry
359 /// is absent then a default entry is added to the collection. The resulting view
360 /// can be modified.
361 /// ```rust
362 /// # tokio_test::block_on(async {
363 /// # use linera_views::context::MemoryContext;
364 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
365 /// # use linera_views::register_view::RegisterView;
366 /// # use linera_views::views::View;
367 /// # let context = MemoryContext::new_for_testing(());
368 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
369 /// ReentrantByteCollectionView::load(context).await.unwrap();
370 /// let subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
371 /// let value = subview.get();
372 /// assert_eq!(*value, String::default());
373 /// # })
374 /// ```
375 pub async fn try_load_entry_mut(
376 &mut self,
377 short_key: &[u8],
378 ) -> Result<WriteGuardedView<W>, ViewError> {
379 Ok(WriteGuardedView(
380 self.try_load_view_mut(short_key)
381 .await?
382 .try_write_arc()
383 .ok_or_else(|| ViewError::TryLockError(short_key.to_vec()))?,
384 ))
385 }
386
387 /// Loads a subview at the given index in the collection and gives read-only access to the data.
388 /// If an entry is absent then `None` is returned.
389 /// ```rust
390 /// # tokio_test::block_on(async {
391 /// # use linera_views::context::MemoryContext;
392 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
393 /// # use linera_views::register_view::RegisterView;
394 /// # use linera_views::views::View;
395 /// # let context = MemoryContext::new_for_testing(());
396 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
397 /// ReentrantByteCollectionView::load(context).await.unwrap();
398 /// {
399 /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
400 /// }
401 /// let subview = view.try_load_entry(&[0, 1]).await.unwrap().unwrap();
402 /// let value = subview.get();
403 /// assert_eq!(*value, String::default());
404 /// # })
405 /// ```
406 pub async fn try_load_entry(
407 &self,
408 short_key: &[u8],
409 ) -> Result<Option<ReadGuardedView<W>>, ViewError> {
410 match self.try_load_view(short_key).await? {
411 None => Ok(None),
412 Some(view) => Ok(Some(ReadGuardedView(
413 view.try_read_arc()
414 .ok_or_else(|| ViewError::TryLockError(short_key.to_vec()))?,
415 ))),
416 }
417 }
418
419 /// Returns `true` if the collection contains a value for the specified key.
420 /// ```rust
421 /// # tokio_test::block_on(async {
422 /// # use linera_views::context::MemoryContext;
423 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
424 /// # use linera_views::register_view::RegisterView;
425 /// # use linera_views::views::View;
426 /// # let context = MemoryContext::new_for_testing(());
427 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
428 /// ReentrantByteCollectionView::load(context).await.unwrap();
429 /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
430 /// assert!(view.contains_key(&[0, 1]).await.unwrap());
431 /// assert!(!view.contains_key(&[0, 2]).await.unwrap());
432 /// # })
433 /// ```
434 pub async fn contains_key(&self, short_key: &[u8]) -> Result<bool, ViewError> {
435 Ok(if let Some(entry) = self.updates.get(short_key) {
436 match entry {
437 Update::Set(_view) => true,
438 Update::Removed => false,
439 }
440 } else if self.delete_storage_first {
441 false
442 } else {
443 let key_index = self
444 .context
445 .base_key()
446 .base_tag_index(KeyTag::Index as u8, short_key);
447 self.context.store().contains_key(&key_index).await?
448 })
449 }
450
451 /// Removes an entry. If absent then nothing happens.
452 /// ```rust
453 /// # tokio_test::block_on(async {
454 /// # use linera_views::context::MemoryContext;
455 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
456 /// # use linera_views::register_view::RegisterView;
457 /// # use linera_views::views::View;
458 /// # let context = MemoryContext::new_for_testing(());
459 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
460 /// ReentrantByteCollectionView::load(context).await.unwrap();
461 /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
462 /// let value = subview.get_mut();
463 /// assert_eq!(*value, String::default());
464 /// view.remove_entry(vec![0, 1]);
465 /// let keys = view.keys().await.unwrap();
466 /// assert_eq!(keys.len(), 0);
467 /// # })
468 /// ```
469 pub fn remove_entry(&mut self, short_key: Vec<u8>) {
470 if self.delete_storage_first {
471 // Optimization: No need to mark `short_key` for deletion as we are going to remove all the keys at once.
472 self.updates.remove(&short_key);
473 } else {
474 self.updates.insert(short_key, Update::Removed);
475 }
476 }
477
478 /// Marks the entry so that it is removed in the next flush.
479 /// ```rust
480 /// # tokio_test::block_on(async {
481 /// # use linera_views::context::MemoryContext;
482 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
483 /// # use linera_views::register_view::RegisterView;
484 /// # use linera_views::views::View;
485 /// # let context = MemoryContext::new_for_testing(());
486 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
487 /// ReentrantByteCollectionView::load(context).await.unwrap();
488 /// {
489 /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
490 /// let value = subview.get_mut();
491 /// *value = String::from("Hello");
492 /// }
493 /// view.try_reset_entry_to_default(&[0, 1]).unwrap();
494 /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
495 /// let value = subview.get_mut();
496 /// assert_eq!(*value, String::default());
497 /// # })
498 /// ```
499 pub fn try_reset_entry_to_default(&mut self, short_key: &[u8]) -> Result<(), ViewError> {
500 let key = self
501 .context
502 .base_key()
503 .base_tag_index(KeyTag::Subview as u8, short_key);
504 let context = self.context.clone_with_base_key(key);
505 let view = W::new(context)?;
506 let view = Arc::new(RwLock::new(view));
507 let view = Update::Set(view);
508 self.updates.insert(short_key.to_vec(), view);
509 Ok(())
510 }
511
512 /// Gets the extra data.
513 pub fn extra(&self) -> &<W::Context as Context>::Extra {
514 self.context.extra()
515 }
516}
517
518impl<W: View> ReentrantByteCollectionView<W::Context, W> {
519 /// Loads multiple entries for writing at once.
520 /// The entries in `short_keys` have to be all distinct.
521 /// ```rust
522 /// # tokio_test::block_on(async {
523 /// # use linera_views::context::MemoryContext;
524 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
525 /// # use linera_views::register_view::RegisterView;
526 /// # use linera_views::views::View;
527 /// # let context = MemoryContext::new_for_testing(());
528 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
529 /// ReentrantByteCollectionView::load(context).await.unwrap();
530 /// {
531 /// let mut subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
532 /// *subview.get_mut() = "Bonjour".to_string();
533 /// }
534 /// let short_keys = vec![vec![0, 1], vec![2, 3]];
535 /// let subviews = view.try_load_entries_mut(short_keys).await.unwrap();
536 /// let value1 = subviews[0].get();
537 /// let value2 = subviews[1].get();
538 /// assert_eq!(*value1, "Bonjour".to_string());
539 /// assert_eq!(*value2, String::default());
540 /// # })
541 /// ```
542 pub async fn try_load_entries_mut(
543 &mut self,
544 short_keys: Vec<Vec<u8>>,
545 ) -> Result<Vec<WriteGuardedView<W>>, ViewError> {
546 let mut short_keys_to_load = Vec::new();
547 let mut keys = Vec::new();
548 for short_key in &short_keys {
549 let key = self
550 .context
551 .base_key()
552 .base_tag_index(KeyTag::Subview as u8, short_key);
553 let context = self.context.clone_with_base_key(key);
554 match self.updates.entry(short_key.to_vec()) {
555 btree_map::Entry::Occupied(mut entry) => {
556 if let Update::Removed = entry.get() {
557 let view = W::new(context)?;
558 let view = Arc::new(RwLock::new(view));
559 entry.insert(Update::Set(view));
560 }
561 }
562 btree_map::Entry::Vacant(entry) => {
563 if self.delete_storage_first {
564 let view = W::new(context)?;
565 let view = Arc::new(RwLock::new(view));
566 entry.insert(Update::Set(view));
567 } else {
568 keys.extend(W::pre_load(&context)?);
569 short_keys_to_load.push(short_key.to_vec());
570 }
571 }
572 }
573 }
574 let values = self.context.store().read_multi_values_bytes(&keys).await?;
575 for (loaded_values, short_key) in values
576 .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
577 .zip(short_keys_to_load)
578 {
579 let key = self
580 .context
581 .base_key()
582 .base_tag_index(KeyTag::Subview as u8, &short_key);
583 let context = self.context.clone_with_base_key(key);
584 let view = W::post_load(context, loaded_values)?;
585 let wrapped_view = Arc::new(RwLock::new(view));
586 self.updates
587 .insert(short_key.to_vec(), Update::Set(wrapped_view));
588 }
589
590 short_keys
591 .into_iter()
592 .map(|short_key| {
593 let Some(Update::Set(view)) = self.updates.get(&short_key) else {
594 unreachable!()
595 };
596 Ok(WriteGuardedView(
597 view.clone()
598 .try_write_arc()
599 .ok_or_else(|| ViewError::TryLockError(short_key))?,
600 ))
601 })
602 .collect()
603 }
604
605 /// Loads multiple entries for reading at once.
606 /// The entries in `short_keys` have to be all distinct.
607 /// ```rust
608 /// # tokio_test::block_on(async {
609 /// # use linera_views::context::MemoryContext;
610 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
611 /// # use linera_views::register_view::RegisterView;
612 /// # use linera_views::views::View;
613 /// # let context = MemoryContext::new_for_testing(());
614 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
615 /// ReentrantByteCollectionView::load(context).await.unwrap();
616 /// {
617 /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
618 /// }
619 /// let short_keys = vec![vec![0, 1], vec![2, 3]];
620 /// let subviews = view.try_load_entries(short_keys).await.unwrap();
621 /// assert!(subviews[1].is_none());
622 /// let value0 = subviews[0].as_ref().unwrap().get();
623 /// assert_eq!(*value0, String::default());
624 /// # })
625 /// ```
626 pub async fn try_load_entries(
627 &self,
628 short_keys: Vec<Vec<u8>>,
629 ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError> {
630 let mut results = vec![None; short_keys.len()];
631 let mut keys_to_check = Vec::new();
632 let mut keys_to_check_metadata = Vec::new();
633
634 for (position, short_key) in short_keys.into_iter().enumerate() {
635 if let Some(update) = self.updates.get(&short_key) {
636 if let Update::Set(view) = update {
637 results[position] = Some((short_key, view.clone()));
638 }
639 } else if !self.delete_storage_first {
640 let key_index = self
641 .context
642 .base_key()
643 .base_tag_index(KeyTag::Index as u8, &short_key);
644 keys_to_check.push(key_index);
645 keys_to_check_metadata.push((position, short_key));
646 }
647 }
648
649 let found_keys = self.context.store().contains_keys(&keys_to_check).await?;
650 let entries_to_load = keys_to_check_metadata
651 .into_iter()
652 .zip(found_keys)
653 .filter_map(|(metadata, found)| found.then_some(metadata))
654 .map(|(position, short_key)| {
655 let subview_key = self
656 .context
657 .base_key()
658 .base_tag_index(KeyTag::Subview as u8, &short_key);
659 let subview_context = self.context.clone_with_base_key(subview_key);
660 (position, short_key.to_owned(), subview_context)
661 })
662 .collect::<Vec<_>>();
663 if !entries_to_load.is_empty() {
664 let mut keys_to_load = Vec::with_capacity(entries_to_load.len() * W::NUM_INIT_KEYS);
665 for (_, _, context) in &entries_to_load {
666 keys_to_load.extend(W::pre_load(context)?);
667 }
668 let values = self
669 .context
670 .store()
671 .read_multi_values_bytes(&keys_to_load)
672 .await?;
673 for (loaded_values, (position, short_key, context)) in values
674 .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
675 .zip(entries_to_load)
676 {
677 let view = W::post_load(context, loaded_values)?;
678 let wrapped_view = Arc::new(RwLock::new(view));
679 results[position] = Some((short_key, wrapped_view));
680 }
681 }
682
683 results
684 .into_iter()
685 .map(|maybe_view| match maybe_view {
686 Some((short_key, view)) => Ok(Some(ReadGuardedView(
687 view.try_read_arc()
688 .ok_or_else(|| ViewError::TryLockError(short_key))?,
689 ))),
690 None => Ok(None),
691 })
692 .collect()
693 }
694
695 /// Loads all the entries for reading at once.
696 /// ```rust
697 /// # tokio_test::block_on(async {
698 /// # use linera_views::context::MemoryContext;
699 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
700 /// # use linera_views::register_view::RegisterView;
701 /// # use linera_views::views::View;
702 /// # let context = MemoryContext::new_for_testing(());
703 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
704 /// ReentrantByteCollectionView::load(context).await.unwrap();
705 /// {
706 /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
707 /// }
708 /// let subviews = view.try_load_all_entries().await.unwrap();
709 /// assert_eq!(subviews.len(), 1);
710 /// # })
711 /// ```
712 pub async fn try_load_all_entries(
713 &self,
714 ) -> Result<Vec<(Vec<u8>, ReadGuardedView<W>)>, ViewError> {
715 let short_keys = self.keys().await?;
716 let mut loaded_views = vec![None; short_keys.len()];
717
718 // Load views that are not in updates and not deleted
719 if !self.delete_storage_first {
720 let mut keys = Vec::new();
721 let mut short_keys_and_indexes = Vec::new();
722 for (index, short_key) in short_keys.iter().enumerate() {
723 if !self.updates.contains_key(short_key) {
724 let key = self
725 .context
726 .base_key()
727 .base_tag_index(KeyTag::Subview as u8, short_key);
728 let context = self.context.clone_with_base_key(key);
729 keys.extend(W::pre_load(&context)?);
730 short_keys_and_indexes.push((short_key.to_vec(), index));
731 }
732 }
733 let values = self.context.store().read_multi_values_bytes(&keys).await?;
734 for (loaded_values, (short_key, index)) in values
735 .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
736 .zip(short_keys_and_indexes)
737 {
738 let key = self
739 .context
740 .base_key()
741 .base_tag_index(KeyTag::Subview as u8, &short_key);
742 let context = self.context.clone_with_base_key(key);
743 let view = W::post_load(context, loaded_values)?;
744 let wrapped_view = Arc::new(RwLock::new(view));
745 loaded_views[index] = Some(wrapped_view);
746 }
747 }
748
749 // Create result from updates and loaded views
750 short_keys
751 .into_iter()
752 .zip(loaded_views)
753 .map(|(short_key, loaded_view)| {
754 let view = if let Some(Update::Set(view)) = self.updates.get(&short_key) {
755 view.clone()
756 } else if let Some(view) = loaded_view {
757 view
758 } else {
759 unreachable!("All entries should have been loaded into memory");
760 };
761 let guard = ReadGuardedView(
762 view.try_read_arc()
763 .ok_or_else(|| ViewError::TryLockError(short_key.clone()))?,
764 );
765 Ok((short_key, guard))
766 })
767 .collect()
768 }
769
770 /// Loads all the entries for writing at once.
771 /// ```rust
772 /// # tokio_test::block_on(async {
773 /// # use linera_views::context::MemoryContext;
774 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
775 /// # use linera_views::register_view::RegisterView;
776 /// # use linera_views::views::View;
777 /// # let context = MemoryContext::new_for_testing(());
778 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
779 /// ReentrantByteCollectionView::load(context).await.unwrap();
780 /// {
781 /// let _subview = view.try_load_entry_mut(&[0, 1]).await.unwrap();
782 /// }
783 /// let subviews = view.try_load_all_entries_mut().await.unwrap();
784 /// assert_eq!(subviews.len(), 1);
785 /// # })
786 /// ```
787 pub async fn try_load_all_entries_mut(
788 &mut self,
789 ) -> Result<Vec<(Vec<u8>, WriteGuardedView<W>)>, ViewError> {
790 let short_keys = self.keys().await?;
791 if !self.delete_storage_first {
792 let mut keys = Vec::new();
793 let mut short_keys_to_load = Vec::new();
794
795 for short_key in &short_keys {
796 if !self.updates.contains_key(short_key) {
797 let key = self
798 .context
799 .base_key()
800 .base_tag_index(KeyTag::Subview as u8, short_key);
801 let context = self.context.clone_with_base_key(key);
802 keys.extend(W::pre_load(&context)?);
803 short_keys_to_load.push(short_key.to_vec());
804 }
805 }
806
807 let values = self.context.store().read_multi_values_bytes(&keys).await?;
808 for (loaded_values, short_key) in values
809 .chunks_exact_or_repeat(W::NUM_INIT_KEYS)
810 .zip(short_keys_to_load)
811 {
812 let key = self
813 .context
814 .base_key()
815 .base_tag_index(KeyTag::Subview as u8, &short_key);
816 let context = self.context.clone_with_base_key(key);
817 let view = W::post_load(context, loaded_values)?;
818 let wrapped_view = Arc::new(RwLock::new(view));
819 self.updates
820 .insert(short_key.to_vec(), Update::Set(wrapped_view));
821 }
822 }
823 short_keys
824 .into_iter()
825 .map(|short_key| {
826 let Some(Update::Set(view)) = self.updates.get(&short_key) else {
827 unreachable!("All entries should have been loaded into `updates`")
828 };
829 let guard = WriteGuardedView(
830 view.clone()
831 .try_write_arc()
832 .ok_or_else(|| ViewError::TryLockError(short_key.clone()))?,
833 );
834 Ok((short_key, guard))
835 })
836 .collect()
837 }
838}
839
840impl<W: View> ReentrantByteCollectionView<W::Context, W> {
841 /// Returns the list of indices in the collection in lexicographic order.
842 /// ```rust
843 /// # tokio_test::block_on(async {
844 /// # use linera_views::context::MemoryContext;
845 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
846 /// # use linera_views::register_view::RegisterView;
847 /// # use linera_views::views::View;
848 /// # let context = MemoryContext::new_for_testing(());
849 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
850 /// ReentrantByteCollectionView::load(context).await.unwrap();
851 /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
852 /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
853 /// let keys = view.keys().await.unwrap();
854 /// assert_eq!(keys, vec![vec![0, 1], vec![0, 2]]);
855 /// # })
856 /// ```
857 pub async fn keys(&self) -> Result<Vec<Vec<u8>>, ViewError> {
858 let mut keys = Vec::new();
859 self.for_each_key(|key| {
860 keys.push(key.to_vec());
861 Ok(())
862 })
863 .await?;
864 Ok(keys)
865 }
866
867 /// Returns the number of indices of the collection.
868 /// ```rust
869 /// # tokio_test::block_on(async {
870 /// # use linera_views::context::MemoryContext;
871 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
872 /// # use linera_views::register_view::RegisterView;
873 /// # use linera_views::views::View;
874 /// # let context = MemoryContext::new_for_testing(());
875 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
876 /// ReentrantByteCollectionView::load(context).await.unwrap();
877 /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
878 /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
879 /// assert_eq!(view.count().await.unwrap(), 2);
880 /// # })
881 /// ```
882 pub async fn count(&self) -> Result<usize, ViewError> {
883 let mut count = 0;
884 self.for_each_key(|_key| {
885 count += 1;
886 Ok(())
887 })
888 .await?;
889 Ok(count)
890 }
891
892 /// Applies a function f on each index (aka key). Keys are visited in a
893 /// lexicographic order. If the function returns false then the loop
894 /// ends prematurely.
895 /// ```rust
896 /// # tokio_test::block_on(async {
897 /// # use linera_views::context::MemoryContext;
898 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
899 /// # use linera_views::register_view::RegisterView;
900 /// # use linera_views::views::View;
901 /// # let context = MemoryContext::new_for_testing(());
902 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
903 /// ReentrantByteCollectionView::load(context).await.unwrap();
904 /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
905 /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
906 /// let mut count = 0;
907 /// view.for_each_key_while(|_key| {
908 /// count += 1;
909 /// Ok(count < 1)
910 /// })
911 /// .await
912 /// .unwrap();
913 /// assert_eq!(count, 1);
914 /// # })
915 /// ```
916 pub async fn for_each_key_while<F>(&self, mut f: F) -> Result<(), ViewError>
917 where
918 F: FnMut(&[u8]) -> Result<bool, ViewError> + Send,
919 {
920 let mut updates = self.updates.iter();
921 let mut update = updates.next();
922 if !self.delete_storage_first {
923 let base = self.get_index_key(&[]);
924 for index in self.context.store().find_keys_by_prefix(&base).await? {
925 loop {
926 match update {
927 Some((key, value)) if key <= &index => {
928 if let Update::Set(_) = value {
929 if !f(key)? {
930 return Ok(());
931 }
932 }
933 update = updates.next();
934 if key == &index {
935 break;
936 }
937 }
938 _ => {
939 if !f(&index)? {
940 return Ok(());
941 }
942 break;
943 }
944 }
945 }
946 }
947 }
948 while let Some((key, value)) = update {
949 if let Update::Set(_) = value {
950 if !f(key)? {
951 return Ok(());
952 }
953 }
954 update = updates.next();
955 }
956 Ok(())
957 }
958
959 /// Applies a function f on each index (aka key). Keys are visited in a
960 /// lexicographic order.
961 /// ```rust
962 /// # tokio_test::block_on(async {
963 /// # use linera_views::context::MemoryContext;
964 /// # use linera_views::reentrant_collection_view::ReentrantByteCollectionView;
965 /// # use linera_views::register_view::RegisterView;
966 /// # use linera_views::views::View;
967 /// # let context = MemoryContext::new_for_testing(());
968 /// let mut view: ReentrantByteCollectionView<_, RegisterView<_, String>> =
969 /// ReentrantByteCollectionView::load(context).await.unwrap();
970 /// view.try_load_entry_mut(&[0, 1]).await.unwrap();
971 /// view.try_load_entry_mut(&[0, 2]).await.unwrap();
972 /// let mut count = 0;
973 /// view.for_each_key(|_key| {
974 /// count += 1;
975 /// Ok(())
976 /// })
977 /// .await
978 /// .unwrap();
979 /// assert_eq!(count, 2);
980 /// # })
981 /// ```
982 pub async fn for_each_key<F>(&self, mut f: F) -> Result<(), ViewError>
983 where
984 F: FnMut(&[u8]) -> Result<(), ViewError> + Send,
985 {
986 self.for_each_key_while(|key| {
987 f(key)?;
988 Ok(true)
989 })
990 .await
991 }
992}
993
994impl<W: HashableView> HashableView for ReentrantByteCollectionView<W::Context, W> {
995 type Hasher = sha3::Sha3_256;
996
997 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
998 #[cfg(with_metrics)]
999 let _hash_latency = metrics::REENTRANT_COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
1000 let mut hasher = sha3::Sha3_256::default();
1001 let keys = self.keys().await?;
1002 let count = keys.len() as u32;
1003 hasher.update_with_bcs_bytes(&count)?;
1004 for key in keys {
1005 hasher.update_with_bytes(&key)?;
1006 let hash = if let Some(entry) = self.updates.get_mut(&key) {
1007 let Update::Set(view) = entry else {
1008 unreachable!();
1009 };
1010 let mut view = view
1011 .try_write_arc()
1012 .ok_or_else(|| ViewError::TryLockError(key))?;
1013 view.hash_mut().await?
1014 } else {
1015 let key = self
1016 .context
1017 .base_key()
1018 .base_tag_index(KeyTag::Subview as u8, &key);
1019 let context = self.context.clone_with_base_key(key);
1020 let mut view = W::load(context).await?;
1021 view.hash_mut().await?
1022 };
1023 hasher.write_all(hash.as_ref())?;
1024 }
1025 Ok(hasher.finalize())
1026 }
1027
1028 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1029 #[cfg(with_metrics)]
1030 let _hash_latency = metrics::REENTRANT_COLLECTION_VIEW_HASH_RUNTIME.measure_latency();
1031 let mut hasher = sha3::Sha3_256::default();
1032 let keys = self.keys().await?;
1033 let count = keys.len() as u32;
1034 hasher.update_with_bcs_bytes(&count)?;
1035 for key in keys {
1036 hasher.update_with_bytes(&key)?;
1037 let hash = if let Some(entry) = self.updates.get(&key) {
1038 let Update::Set(view) = entry else {
1039 unreachable!();
1040 };
1041 let view = view
1042 .try_read_arc()
1043 .ok_or_else(|| ViewError::TryLockError(key))?;
1044 view.hash().await?
1045 } else {
1046 let key = self
1047 .context
1048 .base_key()
1049 .base_tag_index(KeyTag::Subview as u8, &key);
1050 let context = self.context.clone_with_base_key(key);
1051 let view = W::load(context).await?;
1052 view.hash().await?
1053 };
1054 hasher.write_all(hash.as_ref())?;
1055 }
1056 Ok(hasher.finalize())
1057 }
1058}
1059
1060/// A view that supports accessing a collection of views of the same kind, indexed by keys,
1061/// possibly several subviews at a time.
1062#[derive(Debug, Allocative)]
1063#[allocative(bound = "C, I, W: Allocative")]
1064pub struct ReentrantCollectionView<C, I, W> {
1065 collection: ReentrantByteCollectionView<C, W>,
1066 #[allocative(skip)]
1067 _phantom: PhantomData<I>,
1068}
1069
1070impl<I, W, C2> ReplaceContext<C2> for ReentrantCollectionView<W::Context, I, W>
1071where
1072 W: View + ReplaceContext<C2>,
1073 I: Send + Sync + Serialize + DeserializeOwned,
1074 C2: Context,
1075{
1076 type Target = ReentrantCollectionView<C2, I, <W as ReplaceContext<C2>>::Target>;
1077
1078 async fn with_context(
1079 &mut self,
1080 ctx: impl FnOnce(&Self::Context) -> C2 + Clone,
1081 ) -> Self::Target {
1082 ReentrantCollectionView {
1083 collection: self.collection.with_context(ctx).await,
1084 _phantom: self._phantom,
1085 }
1086 }
1087}
1088
1089impl<I, W> View for ReentrantCollectionView<W::Context, I, W>
1090where
1091 W: View,
1092 I: Send + Sync + Serialize + DeserializeOwned,
1093{
1094 const NUM_INIT_KEYS: usize = ReentrantByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;
1095
1096 type Context = W::Context;
1097
1098 fn context(&self) -> &Self::Context {
1099 self.collection.context()
1100 }
1101
1102 fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
1103 ReentrantByteCollectionView::<W::Context, W>::pre_load(context)
1104 }
1105
1106 fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
1107 let collection = ReentrantByteCollectionView::post_load(context, values)?;
1108 Ok(ReentrantCollectionView {
1109 collection,
1110 _phantom: PhantomData,
1111 })
1112 }
1113
1114 fn rollback(&mut self) {
1115 self.collection.rollback()
1116 }
1117
1118 async fn has_pending_changes(&self) -> bool {
1119 self.collection.has_pending_changes().await
1120 }
1121
1122 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
1123 self.collection.pre_save(batch)
1124 }
1125
1126 fn post_save(&mut self) {
1127 self.collection.post_save()
1128 }
1129
1130 fn clear(&mut self) {
1131 self.collection.clear()
1132 }
1133}
1134
1135impl<I, W> ClonableView for ReentrantCollectionView<W::Context, I, W>
1136where
1137 W: ClonableView,
1138 I: Send + Sync + Serialize + DeserializeOwned,
1139{
1140 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
1141 Ok(ReentrantCollectionView {
1142 collection: self.collection.clone_unchecked()?,
1143 _phantom: PhantomData,
1144 })
1145 }
1146}
1147
1148impl<I, W> ReentrantCollectionView<W::Context, I, W>
1149where
1150 W: View,
1151 I: Sync + Send + Serialize + DeserializeOwned,
1152{
1153 /// Loads a subview for the data at the given index in the collection. If an entry
1154 /// is absent then a default entry is put on the collection. The obtained view can
1155 /// then be modified.
1156 /// ```rust
1157 /// # tokio_test::block_on(async {
1158 /// # use linera_views::context::MemoryContext;
1159 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1160 /// # use linera_views::register_view::RegisterView;
1161 /// # use linera_views::views::View;
1162 /// # let context = MemoryContext::new_for_testing(());
1163 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1164 /// ReentrantCollectionView::load(context).await.unwrap();
1165 /// let subview = view.try_load_entry_mut(&23).await.unwrap();
1166 /// let value = subview.get();
1167 /// assert_eq!(*value, String::default());
1168 /// # })
1169 /// ```
1170 pub async fn try_load_entry_mut<Q>(
1171 &mut self,
1172 index: &Q,
1173 ) -> Result<WriteGuardedView<W>, ViewError>
1174 where
1175 I: Borrow<Q>,
1176 Q: Serialize + ?Sized,
1177 {
1178 let short_key = BaseKey::derive_short_key(index)?;
1179 self.collection.try_load_entry_mut(&short_key).await
1180 }
1181
1182 /// Loads a subview at the given index in the collection and gives read-only access to the data.
1183 /// If an entry is absent then `None` is returned.
1184 /// ```rust
1185 /// # tokio_test::block_on(async {
1186 /// # use linera_views::context::MemoryContext;
1187 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1188 /// # use linera_views::register_view::RegisterView;
1189 /// # use linera_views::views::View;
1190 /// # let context = MemoryContext::new_for_testing(());
1191 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1192 /// ReentrantCollectionView::load(context).await.unwrap();
1193 /// {
1194 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1195 /// }
1196 /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
1197 /// let value = subview.get();
1198 /// assert_eq!(*value, String::default());
1199 /// # })
1200 /// ```
1201 pub async fn try_load_entry<Q>(
1202 &self,
1203 index: &Q,
1204 ) -> Result<Option<ReadGuardedView<W>>, ViewError>
1205 where
1206 I: Borrow<Q>,
1207 Q: Serialize + ?Sized,
1208 {
1209 let short_key = BaseKey::derive_short_key(index)?;
1210 self.collection.try_load_entry(&short_key).await
1211 }
1212
1213 /// Returns `true` if the collection contains a value for the specified key.
1214 /// ```rust
1215 /// # tokio_test::block_on(async {
1216 /// # use linera_views::context::MemoryContext;
1217 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1218 /// # use linera_views::register_view::RegisterView;
1219 /// # use linera_views::views::View;
1220 /// # let context = MemoryContext::new_for_testing(());
1221 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1222 /// ReentrantCollectionView::load(context).await.unwrap();
1223 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1224 /// assert!(view.contains_key(&23).await.unwrap());
1225 /// assert!(!view.contains_key(&24).await.unwrap());
1226 /// # })
1227 /// ```
1228 pub async fn contains_key<Q>(&self, index: &Q) -> Result<bool, ViewError>
1229 where
1230 I: Borrow<Q>,
1231 Q: Serialize + ?Sized,
1232 {
1233 let short_key = BaseKey::derive_short_key(index)?;
1234 self.collection.contains_key(&short_key).await
1235 }
1236
1237 /// Marks the entry so that it is removed in the next flush.
1238 /// ```rust
1239 /// # tokio_test::block_on(async {
1240 /// # use linera_views::context::MemoryContext;
1241 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1242 /// # use linera_views::register_view::RegisterView;
1243 /// # use linera_views::views::View;
1244 /// # let context = MemoryContext::new_for_testing(());
1245 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1246 /// ReentrantCollectionView::load(context).await.unwrap();
1247 /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
1248 /// let value = subview.get_mut();
1249 /// assert_eq!(*value, String::default());
1250 /// view.remove_entry(&23);
1251 /// let keys = view.indices().await.unwrap();
1252 /// assert_eq!(keys.len(), 0);
1253 /// # })
1254 /// ```
1255 pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1256 where
1257 I: Borrow<Q>,
1258 Q: Serialize + ?Sized,
1259 {
1260 let short_key = BaseKey::derive_short_key(index)?;
1261 self.collection.remove_entry(short_key);
1262 Ok(())
1263 }
1264
1265 /// Marks the entry so that it is removed in the next flush.
1266 /// ```rust
1267 /// # tokio_test::block_on(async {
1268 /// # use linera_views::context::MemoryContext;
1269 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1270 /// # use linera_views::register_view::RegisterView;
1271 /// # use linera_views::views::View;
1272 /// # let context = MemoryContext::new_for_testing(());
1273 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1274 /// ReentrantCollectionView::load(context).await.unwrap();
1275 /// {
1276 /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
1277 /// let value = subview.get_mut();
1278 /// *value = String::from("Hello");
1279 /// }
1280 /// view.try_reset_entry_to_default(&23).unwrap();
1281 /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
1282 /// let value = subview.get_mut();
1283 /// assert_eq!(*value, String::default());
1284 /// # })
1285 /// ```
1286 pub fn try_reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1287 where
1288 I: Borrow<Q>,
1289 Q: Serialize + ?Sized,
1290 {
1291 let short_key = BaseKey::derive_short_key(index)?;
1292 self.collection.try_reset_entry_to_default(&short_key)
1293 }
1294
1295 /// Gets the extra data.
1296 pub fn extra(&self) -> &<W::Context as Context>::Extra {
1297 self.collection.extra()
1298 }
1299}
1300
1301impl<I, W> ReentrantCollectionView<W::Context, I, W>
1302where
1303 W: View,
1304 I: Sync + Send + Serialize + DeserializeOwned,
1305{
1306 /// Load multiple entries for writing at once.
1307 /// The entries in indices have to be all distinct.
1308 /// ```rust
1309 /// # tokio_test::block_on(async {
1310 /// # use linera_views::context::MemoryContext;
1311 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1312 /// # use linera_views::register_view::RegisterView;
1313 /// # use linera_views::views::View;
1314 /// # let context = MemoryContext::new_for_testing(());
1315 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1316 /// ReentrantCollectionView::load(context).await.unwrap();
1317 /// let indices = vec![23, 42];
1318 /// let subviews = view.try_load_entries_mut(&indices).await.unwrap();
1319 /// let value1 = subviews[0].get();
1320 /// let value2 = subviews[1].get();
1321 /// assert_eq!(*value1, String::default());
1322 /// assert_eq!(*value2, String::default());
1323 /// # })
1324 /// ```
1325 pub async fn try_load_entries_mut<'a, Q>(
1326 &'a mut self,
1327 indices: impl IntoIterator<Item = &'a Q>,
1328 ) -> Result<Vec<WriteGuardedView<W>>, ViewError>
1329 where
1330 I: Borrow<Q>,
1331 Q: Serialize + 'a,
1332 {
1333 let short_keys = indices
1334 .into_iter()
1335 .map(|index| BaseKey::derive_short_key(index))
1336 .collect::<Result<_, _>>()?;
1337 self.collection.try_load_entries_mut(short_keys).await
1338 }
1339
1340 /// Load multiple entries for reading at once.
1341 /// The entries in indices have to be all distinct.
1342 /// ```rust
1343 /// # tokio_test::block_on(async {
1344 /// # use linera_views::context::MemoryContext;
1345 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1346 /// # use linera_views::register_view::RegisterView;
1347 /// # use linera_views::views::View;
1348 /// # let context = MemoryContext::new_for_testing(());
1349 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1350 /// ReentrantCollectionView::load(context).await.unwrap();
1351 /// {
1352 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1353 /// }
1354 /// let indices = vec![23, 42];
1355 /// let subviews = view.try_load_entries(&indices).await.unwrap();
1356 /// assert!(subviews[1].is_none());
1357 /// let value0 = subviews[0].as_ref().unwrap().get();
1358 /// assert_eq!(*value0, String::default());
1359 /// # })
1360 /// ```
1361 pub async fn try_load_entries<'a, Q>(
1362 &'a self,
1363 indices: impl IntoIterator<Item = &'a Q>,
1364 ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError>
1365 where
1366 I: Borrow<Q>,
1367 Q: Serialize + 'a,
1368 {
1369 let short_keys = indices
1370 .into_iter()
1371 .map(|index| BaseKey::derive_short_key(index))
1372 .collect::<Result<_, _>>()?;
1373 self.collection.try_load_entries(short_keys).await
1374 }
1375
1376 /// Loads all entries for writing at once.
1377 /// The entries in indices have to be all distinct.
1378 /// ```rust
1379 /// # tokio_test::block_on(async {
1380 /// # use linera_views::context::MemoryContext;
1381 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1382 /// # use linera_views::register_view::RegisterView;
1383 /// # use linera_views::views::View;
1384 /// # let context = MemoryContext::new_for_testing(());
1385 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1386 /// ReentrantCollectionView::load(context).await.unwrap();
1387 /// {
1388 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1389 /// }
1390 /// let subviews = view.try_load_all_entries_mut().await.unwrap();
1391 /// assert_eq!(subviews.len(), 1);
1392 /// # })
1393 /// ```
1394 pub async fn try_load_all_entries_mut(
1395 &mut self,
1396 ) -> Result<Vec<(I, WriteGuardedView<W>)>, ViewError> {
1397 let results = self.collection.try_load_all_entries_mut().await?;
1398 results
1399 .into_iter()
1400 .map(|(short_key, view)| {
1401 let index = BaseKey::deserialize_value(&short_key)?;
1402 Ok((index, view))
1403 })
1404 .collect()
1405 }
1406
1407 /// Load multiple entries for reading at once.
1408 /// The entries in indices have to be all distinct.
1409 /// ```rust
1410 /// # tokio_test::block_on(async {
1411 /// # use linera_views::context::MemoryContext;
1412 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1413 /// # use linera_views::register_view::RegisterView;
1414 /// # use linera_views::views::View;
1415 /// # let context = MemoryContext::new_for_testing(());
1416 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1417 /// ReentrantCollectionView::load(context).await.unwrap();
1418 /// {
1419 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1420 /// }
1421 /// let subviews = view.try_load_all_entries().await.unwrap();
1422 /// assert_eq!(subviews.len(), 1);
1423 /// # })
1424 /// ```
1425 pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<W>)>, ViewError> {
1426 let results = self.collection.try_load_all_entries().await?;
1427 results
1428 .into_iter()
1429 .map(|(short_key, view)| {
1430 let index = BaseKey::deserialize_value(&short_key)?;
1431 Ok((index, view))
1432 })
1433 .collect()
1434 }
1435}
1436
1437impl<I, W> ReentrantCollectionView<W::Context, I, W>
1438where
1439 W: View,
1440 I: Sync + Send + Serialize + DeserializeOwned,
1441{
1442 /// Returns the list of indices in the collection in an order determined
1443 /// by serialization.
1444 /// ```rust
1445 /// # tokio_test::block_on(async {
1446 /// # use linera_views::context::MemoryContext;
1447 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1448 /// # use linera_views::register_view::RegisterView;
1449 /// # use linera_views::views::View;
1450 /// # let context = MemoryContext::new_for_testing(());
1451 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1452 /// ReentrantCollectionView::load(context).await.unwrap();
1453 /// view.try_load_entry_mut(&23).await.unwrap();
1454 /// view.try_load_entry_mut(&25).await.unwrap();
1455 /// let indices = view.indices().await.unwrap();
1456 /// assert_eq!(indices.len(), 2);
1457 /// # })
1458 /// ```
1459 pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
1460 let mut indices = Vec::new();
1461 self.for_each_index(|index| {
1462 indices.push(index);
1463 Ok(())
1464 })
1465 .await?;
1466 Ok(indices)
1467 }
1468
1469 /// Returns the number of indices in the collection.
1470 /// ```rust
1471 /// # tokio_test::block_on(async {
1472 /// # use linera_views::context::MemoryContext;
1473 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1474 /// # use linera_views::register_view::RegisterView;
1475 /// # use linera_views::views::View;
1476 /// # let context = MemoryContext::new_for_testing(());
1477 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1478 /// ReentrantCollectionView::load(context).await.unwrap();
1479 /// view.try_load_entry_mut(&23).await.unwrap();
1480 /// view.try_load_entry_mut(&25).await.unwrap();
1481 /// assert_eq!(view.count().await.unwrap(), 2);
1482 /// # })
1483 /// ```
1484 pub async fn count(&self) -> Result<usize, ViewError> {
1485 self.collection.count().await
1486 }
1487
1488 /// Applies a function f on each index. Indices are visited in an order
1489 /// determined by the serialization. If the function f returns false then
1490 /// the loop ends prematurely.
1491 /// ```rust
1492 /// # tokio_test::block_on(async {
1493 /// # use linera_views::context::MemoryContext;
1494 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1495 /// # use linera_views::register_view::RegisterView;
1496 /// # use linera_views::views::View;
1497 /// # let context = MemoryContext::new_for_testing(());
1498 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1499 /// ReentrantCollectionView::load(context).await.unwrap();
1500 /// view.try_load_entry_mut(&23).await.unwrap();
1501 /// view.try_load_entry_mut(&24).await.unwrap();
1502 /// let mut count = 0;
1503 /// view.for_each_index_while(|_key| {
1504 /// count += 1;
1505 /// Ok(count < 1)
1506 /// })
1507 /// .await
1508 /// .unwrap();
1509 /// assert_eq!(count, 1);
1510 /// # })
1511 /// ```
1512 pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
1513 where
1514 F: FnMut(I) -> Result<bool, ViewError> + Send,
1515 {
1516 self.collection
1517 .for_each_key_while(|key| {
1518 let index = BaseKey::deserialize_value(key)?;
1519 f(index)
1520 })
1521 .await?;
1522 Ok(())
1523 }
1524
1525 /// Applies a function f on each index. Indices are visited in an order
1526 /// determined by the serialization.
1527 /// ```rust
1528 /// # tokio_test::block_on(async {
1529 /// # use linera_views::context::MemoryContext;
1530 /// # use linera_views::reentrant_collection_view::ReentrantCollectionView;
1531 /// # use linera_views::register_view::RegisterView;
1532 /// # use linera_views::views::View;
1533 /// # let context = MemoryContext::new_for_testing(());
1534 /// let mut view: ReentrantCollectionView<_, u64, RegisterView<_, String>> =
1535 /// ReentrantCollectionView::load(context).await.unwrap();
1536 /// view.try_load_entry_mut(&23).await.unwrap();
1537 /// view.try_load_entry_mut(&28).await.unwrap();
1538 /// let mut count = 0;
1539 /// view.for_each_index(|_key| {
1540 /// count += 1;
1541 /// Ok(())
1542 /// })
1543 /// .await
1544 /// .unwrap();
1545 /// assert_eq!(count, 2);
1546 /// # })
1547 /// ```
1548 pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
1549 where
1550 F: FnMut(I) -> Result<(), ViewError> + Send,
1551 {
1552 self.collection
1553 .for_each_key(|key| {
1554 let index = BaseKey::deserialize_value(key)?;
1555 f(index)
1556 })
1557 .await?;
1558 Ok(())
1559 }
1560}
1561
1562impl<I, W> HashableView for ReentrantCollectionView<W::Context, I, W>
1563where
1564 W: HashableView,
1565 I: Send + Sync + Serialize + DeserializeOwned,
1566{
1567 type Hasher = sha3::Sha3_256;
1568
1569 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1570 self.collection.hash_mut().await
1571 }
1572
1573 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
1574 self.collection.hash().await
1575 }
1576}
1577
1578/// A view that supports accessing a collection of views of the same kind, indexed by an ordered key,
1579/// possibly several subviews at a time.
1580#[derive(Debug, Allocative)]
1581#[allocative(bound = "C, I, W: Allocative")]
1582pub struct ReentrantCustomCollectionView<C, I, W> {
1583 collection: ReentrantByteCollectionView<C, W>,
1584 #[allocative(skip)]
1585 _phantom: PhantomData<I>,
1586}
1587
1588impl<I, W> View for ReentrantCustomCollectionView<W::Context, I, W>
1589where
1590 W: View,
1591 I: Send + Sync + CustomSerialize,
1592{
1593 const NUM_INIT_KEYS: usize = ReentrantByteCollectionView::<W::Context, W>::NUM_INIT_KEYS;
1594
1595 type Context = W::Context;
1596
1597 fn context(&self) -> &Self::Context {
1598 self.collection.context()
1599 }
1600
1601 fn pre_load(context: &Self::Context) -> Result<Vec<Vec<u8>>, ViewError> {
1602 ReentrantByteCollectionView::<_, W>::pre_load(context)
1603 }
1604
1605 fn post_load(context: Self::Context, values: &[Option<Vec<u8>>]) -> Result<Self, ViewError> {
1606 let collection = ReentrantByteCollectionView::post_load(context, values)?;
1607 Ok(ReentrantCustomCollectionView {
1608 collection,
1609 _phantom: PhantomData,
1610 })
1611 }
1612
1613 fn rollback(&mut self) {
1614 self.collection.rollback()
1615 }
1616
1617 async fn has_pending_changes(&self) -> bool {
1618 self.collection.has_pending_changes().await
1619 }
1620
1621 fn pre_save(&self, batch: &mut Batch) -> Result<bool, ViewError> {
1622 self.collection.pre_save(batch)
1623 }
1624
1625 fn post_save(&mut self) {
1626 self.collection.post_save()
1627 }
1628
1629 fn clear(&mut self) {
1630 self.collection.clear()
1631 }
1632}
1633
1634impl<I, W> ClonableView for ReentrantCustomCollectionView<W::Context, I, W>
1635where
1636 W: ClonableView,
1637 Self: View,
1638{
1639 fn clone_unchecked(&mut self) -> Result<Self, ViewError> {
1640 Ok(ReentrantCustomCollectionView {
1641 collection: self.collection.clone_unchecked()?,
1642 _phantom: PhantomData,
1643 })
1644 }
1645}
1646
1647impl<I, W> ReentrantCustomCollectionView<W::Context, I, W>
1648where
1649 W: View,
1650 I: Sync + Send + CustomSerialize,
1651{
1652 /// Loads a subview for the data at the given index in the collection. If an entry
1653 /// is absent then a default entry is put in the collection on this index.
1654 /// ```rust
1655 /// # tokio_test::block_on(async {
1656 /// # use linera_views::context::MemoryContext;
1657 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1658 /// # use linera_views::register_view::RegisterView;
1659 /// # use linera_views::views::View;
1660 /// # let context = MemoryContext::new_for_testing(());
1661 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1662 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1663 /// let subview = view.try_load_entry_mut(&23).await.unwrap();
1664 /// let value = subview.get();
1665 /// assert_eq!(*value, String::default());
1666 /// # })
1667 /// ```
1668 pub async fn try_load_entry_mut<Q>(
1669 &mut self,
1670 index: &Q,
1671 ) -> Result<WriteGuardedView<W>, ViewError>
1672 where
1673 I: Borrow<Q>,
1674 Q: CustomSerialize,
1675 {
1676 let short_key = index.to_custom_bytes()?;
1677 self.collection.try_load_entry_mut(&short_key).await
1678 }
1679
1680 /// Loads a subview at the given index in the collection and gives read-only access to the data.
1681 /// If an entry is absent then `None` is returned.
1682 /// ```rust
1683 /// # tokio_test::block_on(async {
1684 /// # use linera_views::context::MemoryContext;
1685 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1686 /// # use linera_views::register_view::RegisterView;
1687 /// # use linera_views::views::View;
1688 /// # let context = MemoryContext::new_for_testing(());
1689 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1690 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1691 /// {
1692 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1693 /// }
1694 /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
1695 /// let value = subview.get();
1696 /// assert_eq!(*value, String::default());
1697 /// # })
1698 /// ```
1699 pub async fn try_load_entry<Q>(
1700 &self,
1701 index: &Q,
1702 ) -> Result<Option<ReadGuardedView<W>>, ViewError>
1703 where
1704 I: Borrow<Q>,
1705 Q: CustomSerialize,
1706 {
1707 let short_key = index.to_custom_bytes()?;
1708 self.collection.try_load_entry(&short_key).await
1709 }
1710
1711 /// Returns `true` if the collection contains a value for the specified key.
1712 /// ```rust
1713 /// # tokio_test::block_on(async {
1714 /// # use linera_views::context::MemoryContext;
1715 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1716 /// # use linera_views::register_view::RegisterView;
1717 /// # use linera_views::views::View;
1718 /// # let context = MemoryContext::new_for_testing(());
1719 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1720 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1721 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1722 /// assert!(view.contains_key(&23).await.unwrap());
1723 /// assert!(!view.contains_key(&24).await.unwrap());
1724 /// # })
1725 /// ```
1726 pub async fn contains_key<Q>(&self, index: &Q) -> Result<bool, ViewError>
1727 where
1728 I: Borrow<Q>,
1729 Q: CustomSerialize,
1730 {
1731 let short_key = index.to_custom_bytes()?;
1732 self.collection.contains_key(&short_key).await
1733 }
1734
1735 /// Removes an entry. If absent then nothing happens.
1736 /// ```rust
1737 /// # tokio_test::block_on(async {
1738 /// # use linera_views::context::MemoryContext;
1739 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1740 /// # use linera_views::register_view::RegisterView;
1741 /// # use linera_views::views::View;
1742 /// # let context = MemoryContext::new_for_testing(());
1743 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1744 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1745 /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
1746 /// let value = subview.get_mut();
1747 /// assert_eq!(*value, String::default());
1748 /// view.remove_entry(&23);
1749 /// let keys = view.indices().await.unwrap();
1750 /// assert_eq!(keys.len(), 0);
1751 /// # })
1752 /// ```
1753 pub fn remove_entry<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1754 where
1755 I: Borrow<Q>,
1756 Q: CustomSerialize,
1757 {
1758 let short_key = index.to_custom_bytes()?;
1759 self.collection.remove_entry(short_key);
1760 Ok(())
1761 }
1762
1763 /// Marks the entry so that it is removed in the next flush.
1764 /// ```rust
1765 /// # tokio_test::block_on(async {
1766 /// # use linera_views::context::MemoryContext;
1767 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1768 /// # use linera_views::register_view::RegisterView;
1769 /// # use linera_views::views::View;
1770 /// # let context = MemoryContext::new_for_testing(());
1771 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1772 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1773 /// {
1774 /// let mut subview = view.try_load_entry_mut(&23).await.unwrap();
1775 /// let value = subview.get_mut();
1776 /// *value = String::from("Hello");
1777 /// }
1778 /// {
1779 /// view.try_reset_entry_to_default(&23).unwrap();
1780 /// let subview = view.try_load_entry(&23).await.unwrap().unwrap();
1781 /// let value = subview.get();
1782 /// assert_eq!(*value, String::default());
1783 /// }
1784 /// # })
1785 /// ```
1786 pub fn try_reset_entry_to_default<Q>(&mut self, index: &Q) -> Result<(), ViewError>
1787 where
1788 I: Borrow<Q>,
1789 Q: CustomSerialize,
1790 {
1791 let short_key = index.to_custom_bytes()?;
1792 self.collection.try_reset_entry_to_default(&short_key)
1793 }
1794
1795 /// Gets the extra data.
1796 pub fn extra(&self) -> &<W::Context as Context>::Extra {
1797 self.collection.extra()
1798 }
1799}
1800
1801impl<I, W: View> ReentrantCustomCollectionView<W::Context, I, W>
1802where
1803 I: Sync + Send + CustomSerialize,
1804{
1805 /// Load multiple entries for writing at once.
1806 /// The entries in indices have to be all distinct.
1807 /// ```rust
1808 /// # tokio_test::block_on(async {
1809 /// # use linera_views::context::MemoryContext;
1810 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1811 /// # use linera_views::register_view::RegisterView;
1812 /// # use linera_views::views::View;
1813 /// # let context = MemoryContext::new_for_testing(());
1814 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1815 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1816 /// let subviews = view.try_load_entries_mut(&[23, 42]).await.unwrap();
1817 /// let value1 = subviews[0].get();
1818 /// let value2 = subviews[1].get();
1819 /// assert_eq!(*value1, String::default());
1820 /// assert_eq!(*value2, String::default());
1821 /// # })
1822 /// ```
1823 pub async fn try_load_entries_mut<'a, Q>(
1824 &mut self,
1825 indices: impl IntoIterator<Item = &'a Q>,
1826 ) -> Result<Vec<WriteGuardedView<W>>, ViewError>
1827 where
1828 I: Borrow<Q>,
1829 Q: CustomSerialize + 'a,
1830 {
1831 let short_keys = indices
1832 .into_iter()
1833 .map(|index| index.to_custom_bytes())
1834 .collect::<Result<_, _>>()?;
1835 self.collection.try_load_entries_mut(short_keys).await
1836 }
1837
1838 /// Load multiple entries for reading at once.
1839 /// The entries in indices have to be all distinct.
1840 /// ```rust
1841 /// # tokio_test::block_on(async {
1842 /// # use linera_views::context::MemoryContext;
1843 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1844 /// # use linera_views::register_view::RegisterView;
1845 /// # use linera_views::views::View;
1846 /// # let context = MemoryContext::new_for_testing(());
1847 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1848 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1849 /// {
1850 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1851 /// }
1852 /// let subviews = view.try_load_entries(&[23, 42]).await.unwrap();
1853 /// assert!(subviews[1].is_none());
1854 /// let value0 = subviews[0].as_ref().unwrap().get();
1855 /// assert_eq!(*value0, String::default());
1856 /// # })
1857 /// ```
1858 pub async fn try_load_entries<'a, Q>(
1859 &self,
1860 indices: impl IntoIterator<Item = &'a Q>,
1861 ) -> Result<Vec<Option<ReadGuardedView<W>>>, ViewError>
1862 where
1863 I: Borrow<Q>,
1864 Q: CustomSerialize + 'a,
1865 {
1866 let short_keys = indices
1867 .into_iter()
1868 .map(|index| index.to_custom_bytes())
1869 .collect::<Result<_, _>>()?;
1870 self.collection.try_load_entries(short_keys).await
1871 }
1872
1873 /// Loads all entries for writing at once.
1874 /// The entries in indices have to be all distinct.
1875 /// ```rust
1876 /// # tokio_test::block_on(async {
1877 /// # use linera_views::context::MemoryContext;
1878 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1879 /// # use linera_views::register_view::RegisterView;
1880 /// # use linera_views::views::View;
1881 /// # let context = MemoryContext::new_for_testing(());
1882 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1883 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1884 /// {
1885 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1886 /// }
1887 /// let subviews = view.try_load_all_entries_mut().await.unwrap();
1888 /// assert_eq!(subviews.len(), 1);
1889 /// # })
1890 /// ```
1891 pub async fn try_load_all_entries_mut(
1892 &mut self,
1893 ) -> Result<Vec<(I, WriteGuardedView<W>)>, ViewError> {
1894 let results = self.collection.try_load_all_entries_mut().await?;
1895 results
1896 .into_iter()
1897 .map(|(short_key, view)| {
1898 let index = I::from_custom_bytes(&short_key)?;
1899 Ok((index, view))
1900 })
1901 .collect()
1902 }
1903
1904 /// Load multiple entries for reading at once.
1905 /// The entries in indices have to be all distinct.
1906 /// ```rust
1907 /// # tokio_test::block_on(async {
1908 /// # use linera_views::context::MemoryContext;
1909 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1910 /// # use linera_views::register_view::RegisterView;
1911 /// # use linera_views::views::View;
1912 /// # let context = MemoryContext::new_for_testing(());
1913 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1914 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1915 /// {
1916 /// let _subview = view.try_load_entry_mut(&23).await.unwrap();
1917 /// }
1918 /// let subviews = view.try_load_all_entries().await.unwrap();
1919 /// assert_eq!(subviews.len(), 1);
1920 /// # })
1921 /// ```
1922 pub async fn try_load_all_entries(&self) -> Result<Vec<(I, ReadGuardedView<W>)>, ViewError> {
1923 let results = self.collection.try_load_all_entries().await?;
1924 results
1925 .into_iter()
1926 .map(|(short_key, view)| {
1927 let index = I::from_custom_bytes(&short_key)?;
1928 Ok((index, view))
1929 })
1930 .collect()
1931 }
1932}
1933
1934impl<I, W> ReentrantCustomCollectionView<W::Context, I, W>
1935where
1936 W: View,
1937 I: Sync + Send + CustomSerialize,
1938{
1939 /// Returns the list of indices in the collection. The order is determined by
1940 /// the custom serialization.
1941 /// ```rust
1942 /// # tokio_test::block_on(async {
1943 /// # use linera_views::context::MemoryContext;
1944 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1945 /// # use linera_views::register_view::RegisterView;
1946 /// # use linera_views::views::View;
1947 /// # let context = MemoryContext::new_for_testing(());
1948 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1949 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1950 /// view.try_load_entry_mut(&23).await.unwrap();
1951 /// view.try_load_entry_mut(&25).await.unwrap();
1952 /// let indices = view.indices().await.unwrap();
1953 /// assert_eq!(indices, vec![23, 25]);
1954 /// # })
1955 /// ```
1956 pub async fn indices(&self) -> Result<Vec<I>, ViewError> {
1957 let mut indices = Vec::new();
1958 self.for_each_index(|index| {
1959 indices.push(index);
1960 Ok(())
1961 })
1962 .await?;
1963 Ok(indices)
1964 }
1965
1966 /// Returns the number of entries in the collection.
1967 /// ```rust
1968 /// # tokio_test::block_on(async {
1969 /// # use linera_views::context::MemoryContext;
1970 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1971 /// # use linera_views::register_view::RegisterView;
1972 /// # use linera_views::views::View;
1973 /// # let context = MemoryContext::new_for_testing(());
1974 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1975 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1976 /// view.try_load_entry_mut(&23).await.unwrap();
1977 /// view.try_load_entry_mut(&25).await.unwrap();
1978 /// assert_eq!(view.count().await.unwrap(), 2);
1979 /// # })
1980 /// ```
1981 pub async fn count(&self) -> Result<usize, ViewError> {
1982 self.collection.count().await
1983 }
1984
1985 /// Applies a function f on each index. Indices are visited in an order
1986 /// determined by the custom serialization. If the function f returns false
1987 /// then the loop ends prematurely.
1988 /// ```rust
1989 /// # tokio_test::block_on(async {
1990 /// # use linera_views::context::MemoryContext;
1991 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
1992 /// # use linera_views::register_view::RegisterView;
1993 /// # use linera_views::views::View;
1994 /// # let context = MemoryContext::new_for_testing(());
1995 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
1996 /// ReentrantCustomCollectionView::load(context).await.unwrap();
1997 /// view.try_load_entry_mut(&28).await.unwrap();
1998 /// view.try_load_entry_mut(&24).await.unwrap();
1999 /// view.try_load_entry_mut(&23).await.unwrap();
2000 /// let mut part_indices = Vec::new();
2001 /// view.for_each_index_while(|index| {
2002 /// part_indices.push(index);
2003 /// Ok(part_indices.len() < 2)
2004 /// })
2005 /// .await
2006 /// .unwrap();
2007 /// assert_eq!(part_indices, vec![23, 24]);
2008 /// # })
2009 /// ```
2010 pub async fn for_each_index_while<F>(&self, mut f: F) -> Result<(), ViewError>
2011 where
2012 F: FnMut(I) -> Result<bool, ViewError> + Send,
2013 {
2014 self.collection
2015 .for_each_key_while(|key| {
2016 let index = I::from_custom_bytes(key)?;
2017 f(index)
2018 })
2019 .await?;
2020 Ok(())
2021 }
2022
2023 /// Applies a function f on each index. Indices are visited in an order
2024 /// determined by the custom serialization.
2025 /// ```rust
2026 /// # tokio_test::block_on(async {
2027 /// # use linera_views::context::MemoryContext;
2028 /// # use linera_views::reentrant_collection_view::ReentrantCustomCollectionView;
2029 /// # use linera_views::register_view::RegisterView;
2030 /// # use linera_views::views::View;
2031 /// # let context = MemoryContext::new_for_testing(());
2032 /// let mut view: ReentrantCustomCollectionView<_, u128, RegisterView<_, String>> =
2033 /// ReentrantCustomCollectionView::load(context).await.unwrap();
2034 /// view.try_load_entry_mut(&28).await.unwrap();
2035 /// view.try_load_entry_mut(&24).await.unwrap();
2036 /// view.try_load_entry_mut(&23).await.unwrap();
2037 /// let mut indices = Vec::new();
2038 /// view.for_each_index(|index| {
2039 /// indices.push(index);
2040 /// Ok(())
2041 /// })
2042 /// .await
2043 /// .unwrap();
2044 /// assert_eq!(indices, vec![23, 24, 28]);
2045 /// # })
2046 /// ```
2047 pub async fn for_each_index<F>(&self, mut f: F) -> Result<(), ViewError>
2048 where
2049 F: FnMut(I) -> Result<(), ViewError> + Send,
2050 {
2051 self.collection
2052 .for_each_key(|key| {
2053 let index = I::from_custom_bytes(key)?;
2054 f(index)
2055 })
2056 .await?;
2057 Ok(())
2058 }
2059}
2060
2061impl<I, W> HashableView for ReentrantCustomCollectionView<W::Context, I, W>
2062where
2063 W: HashableView,
2064 I: Send + Sync + CustomSerialize,
2065{
2066 type Hasher = sha3::Sha3_256;
2067
2068 async fn hash_mut(&mut self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
2069 self.collection.hash_mut().await
2070 }
2071
2072 async fn hash(&self) -> Result<<Self::Hasher as Hasher>::Output, ViewError> {
2073 self.collection.hash().await
2074 }
2075}
2076
2077/// Type wrapping `ReentrantByteCollectionView` while memoizing the hash.
2078pub type HashedReentrantByteCollectionView<C, W> =
2079 WrappedHashableContainerView<C, ReentrantByteCollectionView<C, W>, HasherOutput>;
2080
2081/// Type wrapping `ReentrantCollectionView` while memoizing the hash.
2082pub type HashedReentrantCollectionView<C, I, W> =
2083 WrappedHashableContainerView<C, ReentrantCollectionView<C, I, W>, HasherOutput>;
2084
2085/// Type wrapping `ReentrantCustomCollectionView` while memoizing the hash.
2086pub type HashedReentrantCustomCollectionView<C, I, W> =
2087 WrappedHashableContainerView<C, ReentrantCustomCollectionView<C, I, W>, HasherOutput>;
2088
2089/// Wrapper around `ReentrantByteCollectionView` to compute hashes based on the history of changes.
2090pub type HistoricallyHashedReentrantByteCollectionView<C, W> =
2091 HistoricallyHashableView<C, ReentrantByteCollectionView<C, W>>;
2092
2093/// Wrapper around `ReentrantCollectionView` to compute hashes based on the history of changes.
2094pub type HistoricallyHashedReentrantCollectionView<C, I, W> =
2095 HistoricallyHashableView<C, ReentrantCollectionView<C, I, W>>;
2096
2097/// Wrapper around `ReentrantCustomCollectionView` to compute hashes based on the history of changes.
2098pub type HistoricallyHashedReentrantCustomCollectionView<C, I, W> =
2099 HistoricallyHashableView<C, ReentrantCustomCollectionView<C, I, W>>;
2100
2101#[cfg(with_graphql)]
2102mod graphql {
2103 use std::borrow::Cow;
2104
2105 use super::{ReadGuardedView, ReentrantCollectionView};
2106 use crate::{
2107 graphql::{hash_name, mangle, missing_key_error, Entry, MapInput},
2108 views::View,
2109 };
2110
2111 impl<T: async_graphql::OutputType> async_graphql::OutputType for ReadGuardedView<T> {
2112 fn type_name() -> Cow<'static, str> {
2113 T::type_name()
2114 }
2115
2116 fn create_type_info(registry: &mut async_graphql::registry::Registry) -> String {
2117 T::create_type_info(registry)
2118 }
2119
2120 async fn resolve(
2121 &self,
2122 ctx: &async_graphql::ContextSelectionSet<'_>,
2123 field: &async_graphql::Positioned<async_graphql::parser::types::Field>,
2124 ) -> async_graphql::ServerResult<async_graphql::Value> {
2125 (**self).resolve(ctx, field).await
2126 }
2127 }
2128
2129 impl<C: Send + Sync, K: async_graphql::OutputType, V: async_graphql::OutputType>
2130 async_graphql::TypeName for ReentrantCollectionView<C, K, V>
2131 {
2132 fn type_name() -> Cow<'static, str> {
2133 format!(
2134 "ReentrantCollectionView_{}_{}_{:08x}",
2135 mangle(K::type_name()),
2136 mangle(V::type_name()),
2137 hash_name::<(K, V)>(),
2138 )
2139 .into()
2140 }
2141 }
2142
2143 #[async_graphql::Object(cache_control(no_cache), name_type)]
2144 impl<K, V> ReentrantCollectionView<V::Context, K, V>
2145 where
2146 K: async_graphql::InputType
2147 + async_graphql::OutputType
2148 + serde::ser::Serialize
2149 + serde::de::DeserializeOwned
2150 + std::fmt::Debug,
2151 V: View + async_graphql::OutputType,
2152 {
2153 async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
2154 Ok(self.indices().await?)
2155 }
2156
2157 #[graphql(derived(name = "count"))]
2158 async fn count_(&self) -> Result<u32, async_graphql::Error> {
2159 Ok(self.count().await? as u32)
2160 }
2161
2162 async fn entry(
2163 &self,
2164 key: K,
2165 ) -> Result<Entry<K, ReadGuardedView<V>>, async_graphql::Error> {
2166 let value = self
2167 .try_load_entry(&key)
2168 .await?
2169 .ok_or_else(|| missing_key_error(&key))?;
2170 Ok(Entry { value, key })
2171 }
2172
2173 async fn entries(
2174 &self,
2175 input: Option<MapInput<K>>,
2176 ) -> Result<Vec<Entry<K, ReadGuardedView<V>>>, async_graphql::Error> {
2177 let keys = if let Some(keys) = input
2178 .and_then(|input| input.filters)
2179 .and_then(|filters| filters.keys)
2180 {
2181 keys
2182 } else {
2183 self.indices().await?
2184 };
2185
2186 let values = self.try_load_entries(&keys).await?;
2187 Ok(values
2188 .into_iter()
2189 .zip(keys)
2190 .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
2191 .collect())
2192 }
2193 }
2194
2195 use crate::reentrant_collection_view::ReentrantCustomCollectionView;
2196 impl<C: Send + Sync, K: async_graphql::OutputType, V: async_graphql::OutputType>
2197 async_graphql::TypeName for ReentrantCustomCollectionView<C, K, V>
2198 {
2199 fn type_name() -> Cow<'static, str> {
2200 format!(
2201 "ReentrantCustomCollectionView_{}_{}_{:08x}",
2202 mangle(K::type_name()),
2203 mangle(V::type_name()),
2204 hash_name::<(K, V)>(),
2205 )
2206 .into()
2207 }
2208 }
2209
2210 #[async_graphql::Object(cache_control(no_cache), name_type)]
2211 impl<K, V> ReentrantCustomCollectionView<V::Context, K, V>
2212 where
2213 K: async_graphql::InputType
2214 + async_graphql::OutputType
2215 + crate::common::CustomSerialize
2216 + std::fmt::Debug,
2217 V: View + async_graphql::OutputType,
2218 {
2219 async fn keys(&self) -> Result<Vec<K>, async_graphql::Error> {
2220 Ok(self.indices().await?)
2221 }
2222
2223 async fn entry(
2224 &self,
2225 key: K,
2226 ) -> Result<Entry<K, ReadGuardedView<V>>, async_graphql::Error> {
2227 let value = self
2228 .try_load_entry(&key)
2229 .await?
2230 .ok_or_else(|| missing_key_error(&key))?;
2231 Ok(Entry { value, key })
2232 }
2233
2234 async fn entries(
2235 &self,
2236 input: Option<MapInput<K>>,
2237 ) -> Result<Vec<Entry<K, ReadGuardedView<V>>>, async_graphql::Error> {
2238 let keys = if let Some(keys) = input
2239 .and_then(|input| input.filters)
2240 .and_then(|filters| filters.keys)
2241 {
2242 keys
2243 } else {
2244 self.indices().await?
2245 };
2246
2247 let values = self.try_load_entries(&keys).await?;
2248 Ok(values
2249 .into_iter()
2250 .zip(keys)
2251 .filter_map(|(value, key)| value.map(|value| Entry { value, key }))
2252 .collect())
2253 }
2254 }
2255}