From 8a9b0f82a9ca43fbbab86feb147422d3f325b233 Mon Sep 17 00:00:00 2001 From: Lachlan Deakin Date: Wed, 24 Jan 2024 11:01:15 +1100 Subject: [PATCH] Refine `UsageLogStorageTransformer` outputs and add docs --- CHANGELOG.md | 1 + examples/sharded_array_write_read.rs | 11 + src/storage/storage_transformer/usage_log.rs | 307 +++++++++++++------ 3 files changed, 219 insertions(+), 100 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 84835fc1..47094bce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking**: Rename `array::safe_transmute_to_bytes_vec` to `array::transmute_to_bytes_vec` - **Breaking**: Make `zfp` a private dependency by changing `Zfp{Bitstream,Field,Stream}` from `pub` to `pub(super)` - **Breaking**: Make `zip` a private dependency by not exposing `ZipError` in `ZipStorageAdapterCreateError` + - Refine `UsageLogStorageTransformer` outputs and add docs ### Removed - **Breaking**: Remove `StorePrefixError::new`, deprecated since `v0.7.3` diff --git a/examples/sharded_array_write_read.rs b/examples/sharded_array_write_read.rs index b3fdb960..eeb8b93a 100644 --- a/examples/sharded_array_write_read.rs +++ b/examples/sharded_array_write_read.rs @@ -1,3 +1,5 @@ +use itertools::Itertools; + fn sharded_array_write_read() -> Result<(), Box> { use zarrs::{ array::{ @@ -148,6 +150,15 @@ fn sharded_array_write_read() -> Result<(), Box> { let tree = node.hierarchy_tree(); println!("The zarr hierarchy tree is:\n{}", tree); + println!( + "The keys in the store are:\n[{}]", + store_readable_listable + .list() + .unwrap_or_default() + .iter() + .format(", ") + ); + Ok(()) } diff --git a/src/storage/storage_transformer/usage_log.rs b/src/storage/storage_transformer/usage_log.rs index 5b225363..b60313c0 100644 --- a/src/storage/storage_transformer/usage_log.rs +++ b/src/storage/storage_transformer/usage_log.rs @@ -5,6 +5,8 @@ use std::{ sync::{Arc, Mutex}, }; +use itertools::Itertools; + use crate::{ array::MaybeBytes, byte_range::ByteRange, @@ -26,7 +28,41 @@ use crate::storage::{ use super::StorageTransformerExtension; -/// The usage log storage transformer. +/// The usage log storage transformer. Logs storage method calls. +/// +/// This storage transformer is for internal use and will not to be included in `storage_transformers` array metadata. +/// It is intended to aid in debugging and optimising performance by revealing storage access patterns. +/// +/// ### Example (log to stdout) +/// ```rust +/// # use std::sync::{Arc, Mutex}; +/// # use zarrs::storage::store::MemoryStore; +/// # use zarrs::storage::storage_transformer::{UsageLogStorageTransformer, StorageTransformerExtension}; +/// let store = Arc::new(MemoryStore::new()); +/// let log_writer = Arc::new(Mutex::new( +/// // std::io::BufWriter::new( +/// std::io::stdout(), +/// // ) +/// )); +/// let usage_log = UsageLogStorageTransformer::new(log_writer, || { +/// chrono::Utc::now().format("[%T%.3f] ").to_string() +/// }); +/// let store = usage_log.create_readable_writable_transformer(store); +/// ```` +/// +/// Applying array methods with the above [`UsageLogStorageTransformer`] prints outputs like: +/// ```text +/// [23:41:19.885] set(group/array/c/1/0, len=140) -> Ok(()) +/// [23:41:19.885] get_partial_values_key(group/array/c/0/0, [-36..-0]) -> len=Ok([36]) +/// [23:41:19.886] get_partial_values_key(group/array/c/0/0, [52..104]) -> len=Ok([52]) +/// [23:41:19.887] get(group/array/c/1/0) -> len=Ok(140) +/// [23:41:19.891] get(zarr.json) -> len=Ok(0) +/// [23:41:19.891] list_dir() -> (keys:[], prefixes:[group/]) +/// [23:41:19.891] get(group/zarr.json) -> len=Ok(86) +/// [23:41:19.891] list_dir(group/) -> (keys:[group/zarr.json], prefixes:[group/array/]) +/// [23:41:19.891] get(group/array/zarr.json) -> len=Ok(1315) +/// [23:41:19.892] list() -> [group/array/c/0/0, group/array/c/1/0, group/array/zarr.json, group/zarr.json] +/// ``` pub struct UsageLogStorageTransformer { handle: Arc>, prefix_func: fn() -> String, @@ -60,6 +96,7 @@ impl UsageLogStorageTransformer { } impl StorageTransformerExtension for UsageLogStorageTransformer { + /// Returns [`None`], since this storage transformer is not intended to be included in array `storage_transformers` metadata. fn create_metadata(&self) -> Option { None } @@ -140,7 +177,7 @@ impl ReadableStorageTraits let result = self.storage.get(key); writeln!( self.handle.lock().unwrap(), - "{}get({key:?}) -> len={:?}", + "{}get({key}) -> len={:?}", (self.prefix_func)(), result .as_ref() @@ -154,54 +191,66 @@ impl ReadableStorageTraits key: &StoreKey, byte_ranges: &[ByteRange], ) -> Result>>, StorageError> { - let _ = writeln!( + let result = self.storage.get_partial_values_key(key, byte_ranges); + writeln!( self.handle.lock().unwrap(), - "{}get_partial_values_key({key}, {byte_ranges:?})", - (self.prefix_func)() - ); - self.storage.get_partial_values_key(key, byte_ranges) + "{}get_partial_values_key({key}, [{}]) -> len={:?}", + (self.prefix_func)(), + byte_ranges.iter().format(", "), + result.as_ref().map(|v| { + v.as_ref() + .map_or(vec![], |v| v.iter().map(std::vec::Vec::len).collect_vec()) + }) + )?; + result } fn get_partial_values( &self, key_ranges: &[StoreKeyRange], ) -> Result, StorageError> { - let _ = writeln!( + let result = self.storage.get_partial_values(key_ranges); + writeln!( self.handle.lock().unwrap(), - "{}get_partial_values({key_ranges:?})", - (self.prefix_func)() - ); - self.storage.get_partial_values(key_ranges) + "{}get_partial_values([{}]) -> len={:?}", + (self.prefix_func)(), + key_ranges.iter().format(", "), + result.as_ref().map(|v| { + v.iter() + .map(|v| v.iter().map(std::vec::Vec::len).collect_vec()) + }) + )?; + result } fn size(&self) -> Result { - let size = self.storage.size(); - let _ = writeln!( + let result = self.storage.size(); + writeln!( self.handle.lock().unwrap(), - "{}size() -> {size:?}", + "{}size() -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } fn size_prefix(&self, prefix: &StorePrefix) -> Result { - let size: Result = self.storage.size_prefix(prefix); - let _ = writeln!( + let result: Result = self.storage.size_prefix(prefix); + writeln!( self.handle.lock().unwrap(), - "{}size_prefix({prefix}) -> {size:?}", + "{}size_prefix({prefix}) -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } fn size_key(&self, key: &StoreKey) -> Result, StorageError> { - let size = self.storage.size_key(key); - let _ = writeln!( + let result = self.storage.size_key(key); + writeln!( self.handle.lock().unwrap(), - "{}size_key({key}) -> {size:?}", + "{}size_key({key}) -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } } @@ -209,30 +258,45 @@ impl ListableStorageTraits for UsageLogStorageTransformerImpl { fn list(&self) -> Result { + let result = self.storage.list(); writeln!( self.handle.lock().unwrap(), - "{}list()", - (self.prefix_func)() + "{}list() -> [{}]", + (self.prefix_func)(), + result.as_ref().unwrap_or(&vec![]).iter().format(", ") )?; - self.storage.list() + result } fn list_prefix(&self, prefix: &StorePrefix) -> Result { + let result = self.storage.list_prefix(prefix); writeln!( self.handle.lock().unwrap(), - "{}list_prefix({prefix:?})", - (self.prefix_func)() + "{}list_prefix({prefix}) -> [{}]", + (self.prefix_func)(), + result.as_ref().unwrap_or(&vec![]).iter().format(", ") )?; - self.storage.list_prefix(prefix) + result } fn list_dir(&self, prefix: &StorePrefix) -> Result { + let result = self.storage.list_dir(prefix); writeln!( self.handle.lock().unwrap(), - "{}list_dir({prefix:?})", - (self.prefix_func)() + "{}list_dir({prefix}) -> (keys:[{}], prefixes:[{}])", + (self.prefix_func)(), + result.as_ref().map_or(String::new(), |skp| skp + .keys() + .iter() + .format(", ") + .to_string()), + result.as_ref().map_or(String::new(), |skp| skp + .prefixes() + .iter() + .format(", ") + .to_string()), )?; - self.storage.list_dir(prefix) + result } } @@ -240,52 +304,58 @@ impl WritableStorageTraits for UsageLogStorageTransformerImpl { fn set(&self, key: &StoreKey, value: &[u8]) -> Result<(), StorageError> { + let result = self.storage.set(key, value); writeln!( self.handle.lock().unwrap(), - "{}set({key:?}, len={})", + "{}set({key}, len={}) -> {result:?}", (self.prefix_func)(), value.len() )?; - self.storage.set(key, value) + result } fn set_partial_values( &self, key_start_values: &[StoreKeyStartValue], ) -> Result<(), StorageError> { + let result = self.storage.set_partial_values(key_start_values); writeln!( self.handle.lock().unwrap(), - "{}set_partial_values({key_start_values:?}", + "{}set_partial_values({key_start_values:?}) -> {result:?}", (self.prefix_func)() )?; - self.storage.set_partial_values(key_start_values) + result } fn erase(&self, key: &StoreKey) -> Result<(), StorageError> { + let result = self.storage.erase(key); writeln!( self.handle.lock().unwrap(), - "{}erase({key:?}", + "{}erase({key}) -> {result:?}", (self.prefix_func)() )?; - self.storage.erase(key) + result } fn erase_values(&self, keys: &[StoreKey]) -> Result<(), StorageError> { + let result = self.storage.erase_values(keys); writeln!( self.handle.lock().unwrap(), - "{}erase_values({keys:?}", + "{}erase_values([{}]) -> {result:?}", + keys.iter().format(", "), (self.prefix_func)() )?; - self.storage.erase_values(keys) + result } fn erase_prefix(&self, prefix: &StorePrefix) -> Result<(), StorageError> { + let result = self.storage.erase_prefix(prefix); writeln!( self.handle.lock().unwrap(), - "{}erase_prefix({prefix:?}", + "{}erase_prefix({prefix}) -> {result:?}", (self.prefix_func)() )?; - self.storage.erase_prefix(prefix) + result } } @@ -293,12 +363,14 @@ impl ReadableWritableStorageTr for UsageLogStorageTransformerImpl { fn mutex(&self, key: &StoreKey) -> Result { + let result = self.storage.mutex(key); writeln!( self.handle.lock().unwrap(), - "{}mutex({key:?}", - (self.prefix_func)() + "{}mutex({key}) -> {:?}", + (self.prefix_func)(), + result.as_ref().map_or((), |_| ()) )?; - self.storage.mutex(key) + result } } @@ -311,7 +383,7 @@ impl AsyncReadableStorageTraits let result = self.storage.get(key).await; writeln!( self.handle.lock().unwrap(), - "{}get({key:?}) -> len={:?}", + "{}get({key}) -> len={:?}", (self.prefix_func)(), result .as_ref() @@ -325,54 +397,66 @@ impl AsyncReadableStorageTraits key: &StoreKey, byte_ranges: &[ByteRange], ) -> Result>>, StorageError> { - let _ = writeln!( + let result = self.storage.get_partial_values_key(key, byte_ranges).await; + writeln!( self.handle.lock().unwrap(), - "{}get_partial_values_key({key}, {byte_ranges:?})", - (self.prefix_func)() - ); - self.storage.get_partial_values_key(key, byte_ranges).await + "{}get_partial_values_key({key}, [{}]) -> len={:?}", + (self.prefix_func)(), + byte_ranges.iter().format(", "), + result.as_ref().map(|v| { + v.as_ref() + .map_or(vec![], |v| v.iter().map(std::vec::Vec::len).collect_vec()) + }) + )?; + result } async fn get_partial_values( &self, key_ranges: &[StoreKeyRange], ) -> Result, StorageError> { - let _ = writeln!( + let result = self.storage.get_partial_values(key_ranges).await; + writeln!( self.handle.lock().unwrap(), - "{}get_partial_values({key_ranges:?})", - (self.prefix_func)() - ); - self.storage.get_partial_values(key_ranges).await + "{}get_partial_values([{}]) -> len={:?}", + (self.prefix_func)(), + key_ranges.iter().format(", "), + result.as_ref().map(|v| { + v.iter() + .map(|v| v.iter().map(std::vec::Vec::len).collect_vec()) + }) + )?; + result } async fn size(&self) -> Result { - let size = self.storage.size().await; - let _ = writeln!( + let result = self.storage.size().await; + writeln!( self.handle.lock().unwrap(), - "{}size() -> {size:?}", + "{}size() -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } async fn size_prefix(&self, prefix: &StorePrefix) -> Result { - let size: Result = self.storage.size_prefix(prefix).await; - let _ = writeln!( + let result: Result = self.storage.size_prefix(prefix).await; + writeln!( self.handle.lock().unwrap(), - "{}size_prefix({prefix}) -> {size:?}", + "{}size_prefix({prefix}) -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } async fn size_key(&self, key: &StoreKey) -> Result, StorageError> { - let size = self.storage.size_key(key).await; - let _ = writeln!( + let result = self.storage.size_key(key).await; + writeln!( self.handle.lock().unwrap(), - "{}size_key({key}) -> {size:?}", + "{}size_key({key}) -> {result:?}", (self.prefix_func)() - ); - size + )?; + result } } @@ -382,30 +466,45 @@ impl AsyncListableStorageTraits for UsageLogStorageTransformerImpl { async fn list(&self) -> Result { + let keys = self.storage.list().await; writeln!( self.handle.lock().unwrap(), - "{}list()", - (self.prefix_func)() + "{}list() -> [{}]", + (self.prefix_func)(), + keys.as_ref().unwrap_or(&vec![]).iter().format(", "), )?; - self.storage.list().await + keys } async fn list_prefix(&self, prefix: &StorePrefix) -> Result { + let result = self.storage.list_prefix(prefix).await; writeln!( self.handle.lock().unwrap(), - "{}list_prefix({prefix:?})", - (self.prefix_func)() + "{}list_prefix({prefix}) -> [{}]", + (self.prefix_func)(), + result.as_ref().unwrap_or(&vec![]).iter().format(", ") )?; - self.storage.list_prefix(prefix).await + result } async fn list_dir(&self, prefix: &StorePrefix) -> Result { + let result = self.storage.list_dir(prefix).await; writeln!( self.handle.lock().unwrap(), - "{}list_dir({prefix:?})", - (self.prefix_func)() + "{}list_dir({prefix}) -> (keys:[{}], prefixes:[{}])", + (self.prefix_func)(), + result.as_ref().map_or(String::new(), |skp| skp + .keys() + .iter() + .format(", ") + .to_string()), + result.as_ref().map_or(String::new(), |skp| skp + .prefixes() + .iter() + .format(", ") + .to_string()), )?; - self.storage.list_dir(prefix).await + result } } @@ -415,52 +514,58 @@ impl AsyncWritableStorageTraits for UsageLogStorageTransformerImpl { async fn set(&self, key: &StoreKey, value: bytes::Bytes) -> Result<(), StorageError> { + let len = value.len(); + let result = self.storage.set(key, value).await; writeln!( self.handle.lock().unwrap(), - "{}set({key:?}, len={})", - (self.prefix_func)(), - value.len() + "{}set({key}, len={len}) -> {result:?}", + (self.prefix_func)() )?; - self.storage.set(key, value).await + result } async fn set_partial_values( &self, key_start_values: &[StoreKeyStartValue], ) -> Result<(), StorageError> { + let result = self.storage.set_partial_values(key_start_values).await; writeln!( self.handle.lock().unwrap(), - "{}set_partial_values({key_start_values:?}", + "{}set_partial_values({key_start_values:?}) -> {result:?}", (self.prefix_func)() )?; - self.storage.set_partial_values(key_start_values).await + result } async fn erase(&self, key: &StoreKey) -> Result<(), StorageError> { + let result = self.storage.erase(key).await; writeln!( self.handle.lock().unwrap(), - "{}erase({key:?}", + "{}erase({key}) -> {result:?}", (self.prefix_func)() )?; - self.storage.erase(key).await + result } async fn erase_values(&self, keys: &[StoreKey]) -> Result<(), StorageError> { + let result = self.storage.erase_values(keys).await; writeln!( self.handle.lock().unwrap(), - "{}erase_values({keys:?}", - (self.prefix_func)() + "{}erase_values([{}]) -> {result:?}", + (self.prefix_func)(), + keys.iter().format(", ") )?; - self.storage.erase_values(keys).await + result } async fn erase_prefix(&self, prefix: &StorePrefix) -> Result<(), StorageError> { + let result = self.storage.erase_prefix(prefix).await; writeln!( self.handle.lock().unwrap(), - "{}erase_prefix({prefix:?}", + "{}erase_prefix({prefix}) -> {result:?}", (self.prefix_func)() )?; - self.storage.erase_prefix(prefix).await + result } } @@ -470,11 +575,13 @@ impl AsyncReadableWritabl for UsageLogStorageTransformerImpl { async fn mutex(&self, key: &StoreKey) -> Result { + let result = self.storage.mutex(key).await; writeln!( self.handle.lock().unwrap(), - "{}mutex({key:?}", - (self.prefix_func)() + "{}mutex({key}) -> {:?}", + (self.prefix_func)(), + result.as_ref().map_or((), |_| ()) )?; - self.storage.mutex(key).await + result } }