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