From efbab08a5788d28a0e39b90fffe68911b990ea3b Mon Sep 17 00:00:00 2001 From: Lei Xu Date: Mon, 29 Sep 2025 10:23:59 -0700 Subject: [PATCH 1/2] upgrade object store --- Cargo.lock | 248 +++++++++++++++++++++++++++++--------------- Cargo.toml | 6 +- benches/memory.rs | 9 +- src/memory.rs | 9 +- src/read_through.rs | 77 ++++++++++---- 5 files changed, 233 insertions(+), 116 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 11a861d..1a7d61d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -222,26 +222,22 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "criterion" -version = "0.5.1" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +checksum = "e1c047a62b0cc3e145fa84415a3191f628e980b194c2755aa12300a4e6cbd928" dependencies = [ "anes", "cast", "ciborium", "clap", "criterion-plot", - "futures", - "is-terminal", - "itertools 0.10.5", + "itertools 0.13.0", "num-traits", - "once_cell", "oorandom", "plotters", "rayon", "regex", "serde", - "serde_derive", "serde_json", "tinytemplate", "tokio", @@ -250,12 +246,12 @@ dependencies = [ [[package]] name = "criterion-plot" -version = "0.5.0" +version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +checksum = "9b1bcc0dc7dfae599d84ad0b1a55f80cde8af3725da8313b528da95ef783e338" dependencies = [ "cast", - "itertools 0.10.5", + "itertools 0.13.0", ] [[package]] @@ -352,6 +348,12 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "486f806e73c5707928240ddc295403b1b93c96a02038563881c4a2fd84b81ac4" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "form_urlencoded" version = "1.2.1" @@ -489,12 +491,6 @@ dependencies = [ "crunchy", ] -[[package]] -name = "heck" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" - [[package]] name = "hermit-abi" version = "0.3.9" @@ -502,10 +498,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024" [[package]] -name = "hermit-abi" -version = "0.4.0" +name = "http" +version = "1.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fbf6a919d6cf397374f7dfeeea91d974c7c0a7221d0d0f4f20d859d329e53fcc" +checksum = "f4a85d31aea989eead29a3aaf9e1115a180df8282431156e533de47660892565" +dependencies = [ + "bytes", + "fnv", + "itoa", +] [[package]] name = "humantime" @@ -675,31 +676,20 @@ dependencies = [ "icu_properties", ] -[[package]] -name = "is-terminal" -version = "0.4.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "261f68e344040fbd0edea105bef17c66edf46f984ddb1115b775ce31be948f4b" -dependencies = [ - "hermit-abi 0.4.0", - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "itertools" -version = "0.10.5" +version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" dependencies = [ "either", ] [[package]] name = "itertools" -version = "0.13.0" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" dependencies = [ "either", ] @@ -722,9 +712,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.171" +version = "0.2.176" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c19937216e9d3aa9956d9bb8dfc0b0c8beb6058fc4f7a4dc4d850edf86a237d6" +checksum = "58f929b4d672ea937a23a1ab494143d968337a5f47e56d0815df1e0890ddf174" [[package]] name = "linux-raw-sys" @@ -799,7 +789,7 @@ dependencies = [ "rustc_version", "smallvec", "tagptr", - "thiserror", + "thiserror 1.0.69", "triomphe", "uuid", ] @@ -828,19 +818,29 @@ version = "1.16.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43" dependencies = [ - "hermit-abi 0.3.9", + "hermit-abi", "libc", ] [[package]] name = "objc2-core-foundation" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "daeaf60f25471d26948a1c2f840e3f7d86f4109e3af4e8e4b5cd70c39690d925" +checksum = "1c10c2894a6fed806ade6027bcd50662746363a9589d3ec9d9bef30a4e4bc166" dependencies = [ "bitflags", ] +[[package]] +name = "objc2-io-kit" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71c1c64d6120e51cd86033f67176b1cb66780c2efe34dec55176f77befd93c0a" +dependencies = [ + "libc", + "objc2-core-foundation", +] + [[package]] name = "object" version = "0.36.5" @@ -852,23 +852,26 @@ dependencies = [ [[package]] name = "object_store" -version = "0.11.2" +version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3cfccb68961a56facde1163f9319e0d15743352344e7808a11795fb99698dcaf" +checksum = "4c1be0c6c22ec0817cdc77d3842f721a17fd30ab6965001415b5402a74e6b740" dependencies = [ "async-trait", "bytes", "chrono", "futures", + "http", "humantime", - "itertools 0.13.0", + "itertools 0.14.0", "parking_lot", "percent-encoding", - "snafu", + "thiserror 2.0.17", "tokio", "tracing", "url", "walkdir", + "wasm-bindgen-futures", + "web-time", ] [[package]] @@ -1238,27 +1241,6 @@ version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" -[[package]] -name = "snafu" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "223891c85e2a29c3fe8fb900c1fae5e69c2e42415e3177752e8718475efa5019" -dependencies = [ - "snafu-derive", -] - -[[package]] -name = "snafu-derive" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "03c3c6b7927ffe7ecaa769ee0e3994da3b8cafc8f444578982c83ecb161af917" -dependencies = [ - "heck", - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "socket2" version = "0.5.8" @@ -1299,14 +1281,15 @@ dependencies = [ [[package]] name = "sysinfo" -version = "0.34.2" +version = "0.37.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4b93974b3d3aeaa036504b8eefd4c039dced109171c1ae973f1dc63b2c7e4b2" +checksum = "3bddd368fda2f82ead69c03d46d351987cfa0c2a57abfa37a017f3aa3e9bf69a" dependencies = [ "libc", "memchr", "ntapi", "objc2-core-foundation", + "objc2-io-kit", "windows", ] @@ -1335,7 +1318,16 @@ version = "1.0.69" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" dependencies = [ - "thiserror-impl", + "thiserror-impl 1.0.69", +] + +[[package]] +name = "thiserror" +version = "2.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f63587ca0f12b72a0600bcba1d40081f830876000bb46dd2337a3051618f4fc8" +dependencies = [ + "thiserror-impl 2.0.17", ] [[package]] @@ -1349,6 +1341,17 @@ dependencies = [ "syn", ] +[[package]] +name = "thiserror-impl" +version = "2.0.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ff15c8ecd7de3849db632e14d18d2571fa09dfc5ed93479bc4485c7a517c913" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tinystr" version = "0.7.6" @@ -1524,6 +1527,19 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dfaf8f50e5f293737ee323940c7d8b08a66a95a419223d9f41610ca08b0833d" +dependencies = [ + "cfg-if", + "js-sys", + "once_cell", + "wasm-bindgen", + "web-sys", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.97" @@ -1563,6 +1579,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "winapi" version = "0.3.9" @@ -1596,12 +1622,24 @@ checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" [[package]] name = "windows" -version = "0.57.0" +version = "0.61.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12342cb4d8e3b046f3d80effd474a7a02447231330ef77d71daa6fbc40681143" +checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" dependencies = [ - "windows-core 0.57.0", - "windows-targets", + "windows-collections", + "windows-core 0.61.2", + "windows-future", + "windows-link", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3beeceb5e5cfd9eb1d76b381630e82c4241ccd0d27f1a39ed41b2760b255c5e8" +dependencies = [ + "windows-core 0.61.2", ] [[package]] @@ -1615,21 +1653,33 @@ dependencies = [ [[package]] name = "windows-core" -version = "0.57.0" +version = "0.61.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2ed2439a290666cd67ecce2b0ffaad89c2a56b976b736e6ece670297897832d" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" dependencies = [ "windows-implement", "windows-interface", + "windows-link", "windows-result", - "windows-targets", + "windows-strings", +] + +[[package]] +name = "windows-future" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" +dependencies = [ + "windows-core 0.61.2", + "windows-link", + "windows-threading", ] [[package]] name = "windows-implement" -version = "0.57.0" +version = "0.60.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9107ddc059d5b6fbfbffdfa7a7fe3e22a226def0b2608f72e9d552763d3e1ad7" +checksum = "edb307e42a74fb6de9bf3a02d9712678b22399c87e6fa869d6dfcd8c1b7754e0" dependencies = [ "proc-macro2", "quote", @@ -1638,22 +1688,47 @@ dependencies = [ [[package]] name = "windows-interface" -version = "0.57.0" +version = "0.59.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29bee4b38ea3cde66011baa44dba677c432a78593e202392d1e9070cf2a7fca7" +checksum = "c0abd1ddbc6964ac14db11c7213d6532ef34bd9aa042c2e5935f59d7908b46a5" dependencies = [ "proc-macro2", "quote", "syn", ] +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + +[[package]] +name = "windows-numerics" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9150af68066c4c5c07ddc0ce30421554771e528bde427614c61038bc2c92c2b1" +dependencies = [ + "windows-core 0.61.2", + "windows-link", +] + [[package]] name = "windows-result" -version = "0.1.2" +version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e383302e8ec8515204254685643de10811af0ed97ea37210dc26fb0032647f8" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" dependencies = [ - "windows-targets", + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link", ] [[package]] @@ -1690,6 +1765,15 @@ dependencies = [ "windows_x86_64_msvc", ] +[[package]] +name = "windows-threading" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b66463ad2e0ea3bbf808b7f1d371311c80e115c0b71d60efc142cafbcfb057a6" +dependencies = [ + "windows-link", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index eda84db..09cc8e6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,12 +16,12 @@ futures = "~0.3" log = "~0.4" moka = { version = "~0.12", features = ["future"] } num_cpus = "1.16" -object_store = "0.11" -sysinfo = "~0.34" +object_store = "0.12" +sysinfo = "~0.37" tokio = { version = "1", features = ["sync"] } [dev-dependencies] -criterion = { version = "~0.5", features = ["async_tokio"] } +criterion = { version = "~0.7", features = ["async_tokio"] } tempfile = "3" tokio = { version = "1", features = ["full"] } rand = "~0.8" diff --git a/benches/memory.rs b/benches/memory.rs index e0074bd..5503d0b 100644 --- a/benches/memory.rs +++ b/benches/memory.rs @@ -41,10 +41,13 @@ fn memory_cache_bench(c: &mut Criterion) { .get_with(&loc, i as u32, { let store = store.clone(); let location = loc.clone(); + let page_size = *page_size; async move { - store - .get_range(&location, i * page_size..(i + 1) * page_size) - .await + let page_size_u64 = u64::try_from(page_size).unwrap(); + let i_u64 = u64::try_from(i).unwrap(); + let start = i_u64 * page_size_u64; + let end = (i_u64 + 1) * page_size_u64; + store.get_range(&location, start..end).await } }) .await diff --git a/src/memory.rs b/src/memory.rs index 016017f..3daec44 100644 --- a/src/memory.rs +++ b/src/memory.rs @@ -342,12 +342,9 @@ mod tests { let location = location.clone(); async move { miss.fetch_add(1, Ordering::SeqCst); - local_fs - .get_range( - &location, - PAGE_SIZE * (*page_id as usize)..PAGE_SIZE * (page_id + 1) as usize, - ) - .await + let start = (PAGE_SIZE as u64) * (*page_id as u64); + let end = (PAGE_SIZE as u64) * ((*page_id as u64) + 1); + local_fs.get_range(&location, start..end).await } }) .await diff --git a/src/read_through.rs b/src/read_through.rs index a24a11e..03e8fc6 100644 --- a/src/read_through.rs +++ b/src/read_through.rs @@ -6,10 +6,10 @@ use bytes::{Bytes, BytesMut}; use futures::{stream, stream::BoxStream, StreamExt, TryStreamExt}; use object_store::{ path::Path, Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, - ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, + ObjectMeta, ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, }; -use crate::{paging::PageCache, stats::CacheStats, Result}; +use crate::{paging::PageCache, stats::CacheStats, Error, Result}; /// Read-through Page Cache. /// @@ -65,32 +65,60 @@ async fn get_range( cache: Arc, stats: Arc, location: &Path, - range: Range, + range: Range, parallelism: usize, ) -> Result { let page_size = cache.page_size(); - let start = (range.start / page_size) * page_size; + let page_size_u64 = page_size as u64; + let range_start = range.start; + let range_end = range.end; + let start = (range_start / page_size_u64) * page_size_u64; let meta = cache.head(location, store.head(location)).await?; + let meta_size = meta.size; - let pages = stream::iter((start..range.end).step_by(page_size)) + let pages = stream::iter((start..range_end).step_by(page_size)) .map(|offset| { let page_cache = cache.clone(); - let page_id = offset / page_size; - let intersection = - std::cmp::max(offset, range.start)..std::cmp::min(offset + page_size, range.end); - let range_in_page = intersection.start - offset..intersection.end - offset; - let page_end = std::cmp::min(offset + page_size, meta.size); let store = store.clone(); let stats = stats.clone(); - - stats.inc_total_reads(); + let loc = location.clone(); + let meta_size = meta_size; async move { - // Actual range in the file. + stats.inc_total_reads(); + + let page_id = + u32::try_from(offset / page_size_u64).map_err(|e| Error::Generic { + store: "ReadThroughCache", + source: Box::new(e), + })?; + + let page_end = std::cmp::min(offset + page_size_u64, meta_size); + let intersection_start = std::cmp::max(offset, range_start); + let intersection_end = std::cmp::min(page_end, range_end); + + let start_in_page = + usize::try_from(intersection_start - offset).map_err(|e| Error::Generic { + store: "ReadThroughCache", + source: Box::new(e), + })?; + let end_in_page = + usize::try_from(intersection_end - offset).map_err(|e| Error::Generic { + store: "ReadThroughCache", + source: Box::new(e), + })?; + + let range_in_page = start_in_page..end_in_page; + let stats_for_miss = stats.clone(); + let store_for_loader = store.clone(); + let loc_for_loader = loc.clone(); + page_cache - .get_range_with(location, page_id as u32, range_in_page, async { - stats.inc_total_misses(); - store.get_range(location, offset..page_end).await + .get_range_with(&loc, page_id, range_in_page, async move { + stats_for_miss.inc_total_misses(); + store_for_loader + .get_range(&loc_for_loader, offset..page_end) + .await }) .await } @@ -103,8 +131,13 @@ async fn get_range( return Ok(pages.into_iter().next().unwrap()); } + let range_len = usize::try_from(range_end - range_start).map_err(|e| Error::Generic { + store: "ReadThroughCache", + source: Box::new(e), + })?; + // stick all bytes together. - let mut buf = BytesMut::with_capacity(range.len()); + let mut buf = BytesMut::with_capacity(range_len); for page in pages { buf.extend_from_slice(&page); } @@ -127,11 +160,11 @@ impl ObjectStore for ReadThroughCache { async fn put_multipart_opts( &self, location: &Path, - _opts: PutMultipartOpts, + options: PutMultipartOptions, ) -> Result> { self.invalidate(location).await?; - self.inner.put_multipart_opts(location, _opts).await + self.inner.put_multipart_opts(location, options).await } async fn get_opts(&self, _location: &Path, _options: GetOptions) -> Result { @@ -155,7 +188,7 @@ impl ObjectStore for ReadThroughCache { let store = inner.clone(); let stats = stats.clone(); let c = cache.clone(); - let page_size = cache.page_size(); + let page_size = cache.page_size() as u64; async move { get_range( @@ -181,7 +214,7 @@ impl ObjectStore for ReadThroughCache { }) } - async fn get_range(&self, location: &Path, range: Range) -> Result { + async fn get_range(&self, location: &Path, range: Range) -> Result { get_range( self.inner.clone(), self.cache.clone(), @@ -202,7 +235,7 @@ impl ObjectStore for ReadThroughCache { self.inner.delete(location).await } - fn list(&'_ self, prefix: Option<&Path>) -> BoxStream<'_, Result> { + fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result> { self.inner.list(prefix) } From 0d84db40cd8c6e08cbe607e174a258d742478842 Mon Sep 17 00:00:00 2001 From: Lei Xu Date: Mon, 29 Sep 2025 11:28:31 -0700 Subject: [PATCH 2/2] Update src/read_through.rs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- src/read_through.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/read_through.rs b/src/read_through.rs index 03e8fc6..2c6e900 100644 --- a/src/read_through.rs +++ b/src/read_through.rs @@ -82,7 +82,6 @@ async fn get_range( let store = store.clone(); let stats = stats.clone(); let loc = location.clone(); - let meta_size = meta_size; async move { stats.inc_total_reads();