awoo.ren / attic
A fork of attic a self-hostable Nix Binary Cache server
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

Merge pull request #160 from zhaofengli/tokio-drift

Various cleanup and WebAssembly CI

authored by

Zhaofeng Li and committed by
GitHub 6d9aeaef acf3c351

+558 -256
+13
.github/workflows/build.yml
··· 52 52 tests=$(nix build .#internalMatrix."$system".\"${{ matrix.nix }}\".attic-tests --no-link --print-out-paths -L) 53 53 find "$tests/bin" -exec {} \; 54 54 55 + - name: Build WebAssembly crates 56 + if: runner.os == 'Linux' 57 + run: | 58 + # https://github.com/rust-lang/rust/issues/122357 59 + export RUST_MIN_STACK=16777216 60 + 61 + pushd attic 62 + nix develop .# --command -- cargo build --target wasm32-unknown-unknown --no-default-features -F chunking -F stream 63 + popd 64 + pushd token 65 + nix develop .# --command -- cargo build --target wasm32-unknown-unknown 66 + popd 67 + 55 68 # TODO: Just take a diff of the list of store paths, also abstract all of this out 56 69 - name: Push build artifacts 57 70 run: |
+205 -17
Cargo.lock
··· 93 93 ] 94 94 95 95 [[package]] 96 + name = "anes" 97 + version = "0.1.6" 98 + source = "registry+https://github.com/rust-lang/crates.io-index" 99 + checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" 100 + 101 + [[package]] 96 102 name = "anstream" 97 103 version = "0.6.14" 98 104 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 231 237 "base64 0.22.1", 232 238 "bytes", 233 239 "cc", 240 + "criterion", 234 241 "cxx", 235 242 "cxx-build", 236 243 "digest", 237 244 "displaydoc", 238 245 "ed25519-compact", 246 + "fastcdc", 239 247 "futures", 240 248 "hex", 241 249 "lazy_static", ··· 309 317 "digest", 310 318 "displaydoc", 311 319 "enum-as-inner", 312 - "fastcdc", 313 320 "futures", 314 321 "hex", 315 322 "http-body-util", ··· 327 334 "serde_with", 328 335 "sha2", 329 336 "tokio", 330 - "tokio-test", 331 337 "tokio-util", 332 338 "toml", 333 339 "tower-http", ··· 1051 1057 ] 1052 1058 1053 1059 [[package]] 1060 + name = "cast" 1061 + version = "0.3.0" 1062 + source = "registry+https://github.com/rust-lang/crates.io-index" 1063 + checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" 1064 + 1065 + [[package]] 1054 1066 name = "cc" 1055 1067 version = "1.1.13" 1056 1068 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1086 1098 "serde", 1087 1099 "wasm-bindgen", 1088 1100 "windows-targets 0.52.5", 1101 + ] 1102 + 1103 + [[package]] 1104 + name = "ciborium" 1105 + version = "0.2.2" 1106 + source = "registry+https://github.com/rust-lang/crates.io-index" 1107 + checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" 1108 + dependencies = [ 1109 + "ciborium-io", 1110 + "ciborium-ll", 1111 + "serde", 1112 + ] 1113 + 1114 + [[package]] 1115 + name = "ciborium-io" 1116 + version = "0.2.2" 1117 + source = "registry+https://github.com/rust-lang/crates.io-index" 1118 + checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" 1119 + 1120 + [[package]] 1121 + name = "ciborium-ll" 1122 + version = "0.2.2" 1123 + source = "registry+https://github.com/rust-lang/crates.io-index" 1124 + checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" 1125 + dependencies = [ 1126 + "ciborium-io", 1127 + "half", 1089 1128 ] 1090 1129 1091 1130 [[package]] ··· 1308 1347 ] 1309 1348 1310 1349 [[package]] 1350 + name = "criterion" 1351 + version = "0.5.1" 1352 + source = "registry+https://github.com/rust-lang/crates.io-index" 1353 + checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" 1354 + dependencies = [ 1355 + "anes", 1356 + "cast", 1357 + "ciborium", 1358 + "clap", 1359 + "criterion-plot", 1360 + "futures", 1361 + "is-terminal", 1362 + "itertools 0.10.5", 1363 + "num-traits", 1364 + "once_cell", 1365 + "oorandom", 1366 + "plotters", 1367 + "rayon", 1368 + "regex", 1369 + "serde", 1370 + "serde_derive", 1371 + "serde_json", 1372 + "tinytemplate", 1373 + "tokio", 1374 + "walkdir", 1375 + ] 1376 + 1377 + [[package]] 1378 + name = "criterion-plot" 1379 + version = "0.5.0" 1380 + source = "registry+https://github.com/rust-lang/crates.io-index" 1381 + checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" 1382 + dependencies = [ 1383 + "cast", 1384 + "itertools 0.10.5", 1385 + ] 1386 + 1387 + [[package]] 1311 1388 name = "crossbeam-channel" 1312 1389 version = "0.5.13" 1313 1390 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1317 1394 ] 1318 1395 1319 1396 [[package]] 1397 + name = "crossbeam-deque" 1398 + version = "0.8.5" 1399 + source = "registry+https://github.com/rust-lang/crates.io-index" 1400 + checksum = "613f8cc01fe9cf1a3eb3d7f488fd2fa8388403e97039e2f73692932e291a770d" 1401 + dependencies = [ 1402 + "crossbeam-epoch", 1403 + "crossbeam-utils", 1404 + ] 1405 + 1406 + [[package]] 1407 + name = "crossbeam-epoch" 1408 + version = "0.9.18" 1409 + source = "registry+https://github.com/rust-lang/crates.io-index" 1410 + checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" 1411 + dependencies = [ 1412 + "crossbeam-utils", 1413 + ] 1414 + 1415 + [[package]] 1320 1416 name = "crossbeam-queue" 1321 1417 version = "0.3.11" 1322 1418 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1330 1426 version = "0.8.20" 1331 1427 source = "registry+https://github.com/rust-lang/crates.io-index" 1332 1428 checksum = "22ec99545bb0ed0ea7bb9b8e1e9122ea386ff8a48c0922e43f36d45ab09e0e80" 1429 + 1430 + [[package]] 1431 + name = "crunchy" 1432 + version = "0.2.2" 1433 + source = "registry+https://github.com/rust-lang/crates.io-index" 1434 + checksum = "7a81dae078cea95a014a339291cec439d2f232ebe854a9d672b796c6afafa9b7" 1333 1435 1334 1436 [[package]] 1335 1437 name = "crypto-bigint" ··· 1698 1800 version = "3.1.0" 1699 1801 source = "registry+https://github.com/rust-lang/crates.io-index" 1700 1802 checksum = "a71061d097bfa9a5a4d2efdec57990d9a88745020b365191d37e48541a1628f2" 1803 + dependencies = [ 1804 + "async-stream", 1805 + "tokio", 1806 + "tokio-stream", 1807 + ] 1701 1808 1702 1809 [[package]] 1703 1810 name = "fastrand" ··· 1957 2064 ] 1958 2065 1959 2066 [[package]] 2067 + name = "half" 2068 + version = "2.4.1" 2069 + source = "registry+https://github.com/rust-lang/crates.io-index" 2070 + checksum = "6dd08c532ae367adf81c312a4580bc67f1d0fe8bc9c460520283f4c0ff277888" 2071 + dependencies = [ 2072 + "cfg-if", 2073 + "crunchy", 2074 + ] 2075 + 2076 + [[package]] 1960 2077 name = "hashbrown" 1961 2078 version = "0.12.3" 1962 2079 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2389 2506 checksum = "8f518f335dce6725a761382244631d86cf0ccb2863413590b31338feb467f9c3" 2390 2507 2391 2508 [[package]] 2509 + name = "is-terminal" 2510 + version = "0.4.12" 2511 + source = "registry+https://github.com/rust-lang/crates.io-index" 2512 + checksum = "f23ff5ef2b80d608d61efee834934d862cd92461afc0560dedf493e4c033738b" 2513 + dependencies = [ 2514 + "hermit-abi", 2515 + "libc", 2516 + "windows-sys 0.52.0", 2517 + ] 2518 + 2519 + [[package]] 2392 2520 name = "is_terminal_polyfill" 2393 2521 version = "1.70.0" 2394 2522 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2396 2524 2397 2525 [[package]] 2398 2526 name = "itertools" 2527 + version = "0.10.5" 2528 + source = "registry+https://github.com/rust-lang/crates.io-index" 2529 + checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" 2530 + dependencies = [ 2531 + "either", 2532 + ] 2533 + 2534 + [[package]] 2535 + name = "itertools" 2399 2536 version = "0.12.1" 2400 2537 source = "registry+https://github.com/rust-lang/crates.io-index" 2401 2538 checksum = "ba291022dbbd398a455acf126c1e341954079855bc60dfdda641363bd6922569" ··· 2777 2914 checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92" 2778 2915 2779 2916 [[package]] 2917 + name = "oorandom" 2918 + version = "11.1.4" 2919 + source = "registry+https://github.com/rust-lang/crates.io-index" 2920 + checksum = "b410bbe7e14ab526a0e86877eb47c6996a2bd7746f027ba551028c925390e4e9" 2921 + 2922 + [[package]] 2780 2923 name = "openssl-probe" 2781 2924 version = "0.1.5" 2782 2925 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 3003 3146 checksum = "d231b230927b5e4ad203db57bbcbee2802f6bce620b1e4a9024a07d94e2907ec" 3004 3147 3005 3148 [[package]] 3149 + name = "plotters" 3150 + version = "0.3.6" 3151 + source = "registry+https://github.com/rust-lang/crates.io-index" 3152 + checksum = "a15b6eccb8484002195a3e44fe65a4ce8e93a625797a063735536fd59cb01cf3" 3153 + dependencies = [ 3154 + "num-traits", 3155 + "plotters-backend", 3156 + "plotters-svg", 3157 + "wasm-bindgen", 3158 + "web-sys", 3159 + ] 3160 + 3161 + [[package]] 3162 + name = "plotters-backend" 3163 + version = "0.3.6" 3164 + source = "registry+https://github.com/rust-lang/crates.io-index" 3165 + checksum = "414cec62c6634ae900ea1c56128dfe87cf63e7caece0852ec76aba307cebadb7" 3166 + 3167 + [[package]] 3168 + name = "plotters-svg" 3169 + version = "0.3.6" 3170 + source = "registry+https://github.com/rust-lang/crates.io-index" 3171 + checksum = "81b30686a7d9c3e010b84284bdd26a29f2138574f52f5eb6f794fc0ad924e705" 3172 + dependencies = [ 3173 + "plotters-backend", 3174 + ] 3175 + 3176 + [[package]] 3006 3177 name = "portable-atomic" 3007 3178 version = "1.6.0" 3008 3179 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 3088 3259 checksum = "81bddcdb20abf9501610992b6759a4c888aef7d1a7247ef75e2404275ac24af1" 3089 3260 dependencies = [ 3090 3261 "anyhow", 3091 - "itertools", 3262 + "itertools 0.12.1", 3092 3263 "proc-macro2", 3093 3264 "quote", 3094 3265 "syn 2.0.66", ··· 3166 3337 checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" 3167 3338 dependencies = [ 3168 3339 "getrandom", 3340 + ] 3341 + 3342 + [[package]] 3343 + name = "rayon" 3344 + version = "1.10.0" 3345 + source = "registry+https://github.com/rust-lang/crates.io-index" 3346 + checksum = "b418a60154510ca1a002a752ca9714984e21e4241e804d32555251faf8b78ffa" 3347 + dependencies = [ 3348 + "either", 3349 + "rayon-core", 3350 + ] 3351 + 3352 + [[package]] 3353 + name = "rayon-core" 3354 + version = "1.12.1" 3355 + source = "registry+https://github.com/rust-lang/crates.io-index" 3356 + checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2" 3357 + dependencies = [ 3358 + "crossbeam-deque", 3359 + "crossbeam-utils", 3169 3360 ] 3170 3361 3171 3362 [[package]] ··· 4069 4260 source = "registry+https://github.com/rust-lang/crates.io-index" 4070 4261 checksum = "ce81b7bd7c4493975347ef60d8c7e8b742d4694f4c49f93e0a12ea263938176c" 4071 4262 dependencies = [ 4072 - "itertools", 4263 + "itertools 0.12.1", 4073 4264 "nom", 4074 4265 "unicode_categories", 4075 4266 ] ··· 4460 4651 ] 4461 4652 4462 4653 [[package]] 4654 + name = "tinytemplate" 4655 + version = "1.2.1" 4656 + source = "registry+https://github.com/rust-lang/crates.io-index" 4657 + checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" 4658 + dependencies = [ 4659 + "serde", 4660 + "serde_json", 4661 + ] 4662 + 4663 + [[package]] 4463 4664 name = "tinyvec" 4464 4665 version = "1.6.0" 4465 4666 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 4544 4745 "futures-core", 4545 4746 "pin-project-lite", 4546 4747 "tokio", 4547 - ] 4548 - 4549 - [[package]] 4550 - name = "tokio-test" 4551 - version = "0.4.4" 4552 - source = "registry+https://github.com/rust-lang/crates.io-index" 4553 - checksum = "2468baabc3311435b55dd935f702f42cd1b8abb7e754fb7dfb16bd36aa88f9f7" 4554 - dependencies = [ 4555 - "async-stream", 4556 - "bytes", 4557 - "futures-core", 4558 - "tokio", 4559 - "tokio-stream", 4560 4748 ] 4561 4749 4562 4750 [[package]]
+29 -9
attic/Cargo.toml
··· 11 11 displaydoc = "0.2.4" 12 12 digest = "0.10.7" 13 13 ed25519-compact = "2.0.4" 14 + fastcdc = "3.0.3" 14 15 futures = "0.3.28" 15 16 hex = "0.4.3" 16 17 lazy_static = "1.4.0" ··· 32 33 version = "1.28.2" 33 34 optional = true 34 35 features = [ 35 - "fs", 36 36 "io-util", 37 - "process", 37 + "macros", 38 38 "sync", 39 39 ] 40 40 41 41 [dev-dependencies] 42 + criterion = { version = "0.5", features = ["html_reports", "async_tokio"] } 43 + fastcdc = { version = "*", features = ["tokio"] } 42 44 serde_json = "1.0.96" 43 - tokio-test = "0.4.2" 44 45 45 46 [build-dependencies] 46 47 cc = "1.1.13" ··· 50 51 version-compare = "0.2.0" 51 52 52 53 [features] 53 - default = [ "nix_store", "tokio" ] 54 + default = [ 55 + "chunking", 56 + "nix_store", 57 + "stream", 58 + "tokio", 59 + ] 60 + 61 + # Chunking. 62 + chunking = ["tokio", "stream", "dep:async-stream"] 54 63 55 64 # Native libnixstore bindings. 56 65 # 57 66 # When disabled, the native Rust portions of nix_store can still be used. 58 - nix_store = [ "dep:cxx", "dep:cxx-build", "tokio/rt" ] 67 + nix_store = [ 68 + "tokio", 69 + "tokio/fs", 70 + "tokio/process", 71 + "dep:cxx", 72 + "dep:cxx-build", 73 + ] 59 74 60 - # Tokio. 61 - # 62 - # When disabled, any part depending on tokio is unavailable. 63 - tokio = [ "dep:tokio", "dep:async-stream" ] 75 + # Stream utilities. 76 + stream = ["tokio", "dep:async-stream"] 77 + 78 + # Tokio runtime. 79 + tokio = ["dep:tokio", "tokio/rt"] 80 + 81 + [[bench]] 82 + name = "chunking" 83 + harness = false
+84
attic/benches/chunking.rs
··· 1 + use std::io::Cursor; 2 + 3 + use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, Throughput}; 4 + use futures::StreamExt; 5 + 6 + use attic::chunking::chunk_stream; 7 + use attic::testing::{get_fake_data, get_runtime}; 8 + 9 + struct Parameters { 10 + min_size: u32, 11 + avg_size: u32, 12 + max_size: u32, 13 + } 14 + 15 + pub fn bench_chunking(c: &mut Criterion) { 16 + let rt = get_runtime(); 17 + let data = get_fake_data(128 * 1024 * 1024); // 128 MiB 18 + 19 + let cases = [ 20 + ( 21 + "2K,4K,8K", 22 + Parameters { 23 + min_size: 2 * 1024, 24 + avg_size: 4 * 1024, 25 + max_size: 8 * 1024, 26 + }, 27 + ), 28 + ( 29 + "8K,16K,32K", 30 + Parameters { 31 + min_size: 8 * 1024, 32 + avg_size: 16 * 1024, 33 + max_size: 32 * 1024, 34 + }, 35 + ), 36 + ( 37 + "1M,4M,16M", 38 + Parameters { 39 + min_size: 1024 * 1024, 40 + avg_size: 4 * 1024 * 1024, 41 + max_size: 16 * 1024 * 1024, 42 + }, 43 + ), 44 + ]; 45 + 46 + let mut group = c.benchmark_group("chunking"); 47 + group.throughput(Throughput::Bytes(data.len() as u64)); 48 + 49 + for (case, params) in cases { 50 + group.bench_with_input(BenchmarkId::new("ronomon", case), &params, |b, params| { 51 + b.to_async(&rt).iter(|| async { 52 + let cursor = Cursor::new(&data); 53 + let mut chunks = chunk_stream( 54 + cursor, 55 + params.min_size as usize, 56 + params.avg_size as usize, 57 + params.max_size as usize, 58 + ); 59 + while let Some(chunk) = chunks.next().await { 60 + black_box(chunk).unwrap(); 61 + } 62 + }) 63 + }); 64 + group.bench_with_input(BenchmarkId::new("v2020", case), &params, |b, params| { 65 + b.to_async(&rt).iter(|| async { 66 + let cursor = Cursor::new(&data); 67 + let mut chunks = fastcdc::v2020::AsyncStreamCDC::new( 68 + cursor, 69 + params.min_size, 70 + params.avg_size, 71 + params.max_size, 72 + ); 73 + let mut chunks = Box::pin(chunks.as_stream()); 74 + while let Some(chunk) = chunks.next().await { 75 + black_box(chunk).unwrap(); 76 + } 77 + }) 78 + }); 79 + } 80 + group.finish(); 81 + } 82 + 83 + criterion_group!(benches, bench_chunking); 84 + criterion_main!(benches);
+101
attic/src/chunking/mod.rs
··· 1 + //! Chunking. 2 + //! 3 + //! We perform chunking on uncompressed NARs using the FastCDC 4 + //! algorithm. 5 + 6 + use async_stream::try_stream; 7 + use bytes::{BufMut, Bytes, BytesMut}; 8 + use fastcdc::ronomon::FastCDC; 9 + use futures::stream::Stream; 10 + use tokio::io::AsyncRead; 11 + 12 + use crate::stream::read_chunk_async; 13 + 14 + /// Splits a streams into content-defined chunks. 15 + /// 16 + /// This is a wrapper over fastcdc-rs that takes an `AsyncRead` and 17 + /// returns a `Stream` of chunks as `Bytes`s. 18 + pub fn chunk_stream<R>( 19 + mut stream: R, 20 + min_size: usize, 21 + avg_size: usize, 22 + max_size: usize, 23 + ) -> impl Stream<Item = std::io::Result<Bytes>> 24 + where 25 + R: AsyncRead + Unpin + Send, 26 + { 27 + let s = try_stream! { 28 + let mut buf = BytesMut::with_capacity(max_size); 29 + 30 + loop { 31 + let read = read_chunk_async(&mut stream, buf).await?; 32 + 33 + let mut eof = false; 34 + if read.is_empty() { 35 + // Already EOF 36 + break; 37 + } else if read.len() < max_size { 38 + // Last read 39 + eof = true; 40 + } 41 + 42 + let chunks = FastCDC::with_eof(&read, min_size, avg_size, max_size, eof); 43 + let mut consumed = 0; 44 + 45 + for chunk in chunks { 46 + consumed += chunk.length; 47 + 48 + let slice = read.slice(chunk.offset..chunk.offset + chunk.length); 49 + yield slice; 50 + } 51 + 52 + if eof { 53 + break; 54 + } 55 + 56 + buf = BytesMut::with_capacity(max_size); 57 + 58 + if consumed < read.len() { 59 + // remaining bytes for the next read 60 + buf.put_slice(&read[consumed..]); 61 + } 62 + } 63 + }; 64 + 65 + Box::pin(s) 66 + } 67 + 68 + #[cfg(test)] 69 + mod tests { 70 + use super::*; 71 + 72 + use std::io::Cursor; 73 + 74 + use futures::StreamExt; 75 + 76 + use crate::testing::get_fake_data; 77 + 78 + /// Chunks and reconstructs a file. 79 + #[tokio::test] 80 + async fn test_chunking_basic() { 81 + async fn case(size: usize) { 82 + let test_file = get_fake_data(size); // 32 MiB 83 + let mut reconstructed_file = Vec::new(); 84 + 85 + let cursor = Cursor::new(&test_file); 86 + let mut chunks = chunk_stream(cursor, 8 * 1024, 16 * 1024, 32 * 1024); 87 + 88 + while let Some(chunk) = chunks.next().await { 89 + let chunk = chunk.unwrap(); 90 + eprintln!("Got a {}-byte chunk", chunk.len()); 91 + reconstructed_file.extend(chunk); 92 + } 93 + 94 + assert_eq!(reconstructed_file, test_file); 95 + } 96 + 97 + case(32 * 1024 * 1024 - 1).await; 98 + case(32 * 1024 * 1024).await; 99 + case(32 * 1024 * 1024 + 1).await; 100 + } 101 + }
+1 -1
attic/src/hash/mod.rs
··· 118 118 } 119 119 120 120 /// Decodes a base16 or base32 encoded hash containing a specified number of bytes. 121 - fn decode_hash<'s>(s: &'s str, typ: &'static str, expected_bytes: usize) -> AtticResult<Vec<u8>> { 121 + fn decode_hash(s: &str, typ: &'static str, expected_bytes: usize) -> AtticResult<Vec<u8>> { 122 122 let base16_len = expected_bytes * 2; 123 123 let base32_len = (expected_bytes * 8 - 1) / 5 + 1; 124 124
+3 -1
attic/src/lib.rs
··· 17 17 18 18 pub mod api; 19 19 pub mod cache; 20 + #[cfg(feature = "chunking")] 21 + pub mod chunking; 20 22 pub mod error; 21 23 pub mod hash; 22 24 pub mod mime; 23 25 pub mod nix_store; 24 26 pub mod signing; 25 - #[cfg(feature = "tokio")] 27 + #[cfg(feature = "stream")] 26 28 pub mod stream; 27 29 #[cfg(target_family = "unix")] 28 30 pub mod testing;
+1 -1
attic/src/nix_store/bindings/mod.rs
··· 18 18 unsafe impl Sync for FfiNixStore {} 19 19 20 20 impl FfiNixStore { 21 - pub fn store<'a>(&'a self) -> Pin<&'a mut ffi::CNixStore> { 21 + pub fn store(&self) -> Pin<&mut ffi::CNixStore> { 22 22 unsafe { 23 23 let ptr = self.0.get().as_mut().unwrap(); 24 24 ptr.pin_mut()
+72 -81
attic/src/nix_store/tests/mod.rs
··· 6 6 use std::process::Command; 7 7 8 8 use serde::de::DeserializeOwned; 9 - use tokio_test::block_on; 10 9 11 10 pub mod test_nar; 12 11 ··· 143 142 StorePathHash::new(h).unwrap_err(); 144 143 } 145 144 146 - #[test] 147 - fn test_nar_streaming() { 145 + #[tokio::test] 146 + async fn test_nar_streaming() { 148 147 let store = NixStore::connect().expect("Failed to connect to the Nix store"); 149 148 150 - block_on(async move { 151 - let test_nar = test_nar::NO_DEPS; 152 - test_nar.import().await.expect("Could not import test NAR"); 149 + let test_nar = test_nar::NO_DEPS; 150 + test_nar.import().await.expect("Could not import test NAR"); 153 151 154 - let target = test_nar.get_target().expect("Could not create dump target"); 155 - let writer = target.get_writer().await.expect("Could not get writer"); 152 + let target = test_nar.get_target().expect("Could not create dump target"); 153 + let writer = target.get_writer().await.expect("Could not get writer"); 156 154 157 - let store_path = store.parse_store_path(test_nar.path()).unwrap(); 155 + let store_path = store.parse_store_path(test_nar.path()).unwrap(); 158 156 159 - let stream = store.nar_from_path(store_path); 160 - stream.write_all(writer).await.unwrap(); 157 + let stream = store.nar_from_path(store_path); 158 + stream.write_all(writer).await.unwrap(); 161 159 162 - target 163 - .validate() 164 - .await 165 - .expect("Could not validate resulting dump"); 166 - }); 160 + target 161 + .validate() 162 + .await 163 + .expect("Could not validate resulting dump"); 167 164 } 168 165 169 - #[test] 170 - fn test_compute_fs_closure() { 166 + #[tokio::test] 167 + async fn test_compute_fs_closure() { 168 + use test_nar::{WITH_DEPS_A, WITH_DEPS_B, WITH_DEPS_C}; 169 + 171 170 let store = NixStore::connect().expect("Failed to connect to the Nix store"); 172 171 173 - block_on(async move { 174 - use test_nar::{WITH_DEPS_A, WITH_DEPS_B, WITH_DEPS_C}; 175 - 176 - for nar in [WITH_DEPS_C, WITH_DEPS_B, WITH_DEPS_A] { 177 - nar.import().await.expect("Could not import test NAR"); 172 + for nar in [WITH_DEPS_C, WITH_DEPS_B, WITH_DEPS_A] { 173 + nar.import().await.expect("Could not import test NAR"); 178 174 179 - let path = store 180 - .parse_store_path(nar.path()) 181 - .expect("Could not parse store path"); 175 + let path = store 176 + .parse_store_path(nar.path()) 177 + .expect("Could not parse store path"); 182 178 183 - let actual: HashSet<StorePath> = store 184 - .compute_fs_closure(path, false, false, false) 185 - .await 186 - .expect("Could not compute closure") 187 - .into_iter() 188 - .collect(); 179 + let actual: HashSet<StorePath> = store 180 + .compute_fs_closure(path, false, false, false) 181 + .await 182 + .expect("Could not compute closure") 183 + .into_iter() 184 + .collect(); 189 185 190 - assert_eq!(nar.closure(), actual); 191 - } 192 - }); 186 + assert_eq!(nar.closure(), actual); 187 + } 193 188 } 194 189 195 - #[test] 196 - fn test_compute_fs_closure_multi() { 190 + #[tokio::test] 191 + async fn test_compute_fs_closure_multi() { 192 + use test_nar::{NO_DEPS, WITH_DEPS_A, WITH_DEPS_B, WITH_DEPS_C}; 193 + 197 194 let store = NixStore::connect().expect("Failed to connect to the Nix store"); 198 195 199 - block_on(async move { 200 - use test_nar::{NO_DEPS, WITH_DEPS_A, WITH_DEPS_B, WITH_DEPS_C}; 201 - 202 - for nar in [NO_DEPS, WITH_DEPS_C, WITH_DEPS_B, WITH_DEPS_A] { 203 - nar.import().await.expect("Could not import test NAR"); 204 - } 196 + for nar in [NO_DEPS, WITH_DEPS_C, WITH_DEPS_B, WITH_DEPS_A] { 197 + nar.import().await.expect("Could not import test NAR"); 198 + } 205 199 206 - let mut expected = NO_DEPS.closure(); 207 - expected.extend(WITH_DEPS_A.closure()); 200 + let mut expected = NO_DEPS.closure(); 201 + expected.extend(WITH_DEPS_A.closure()); 208 202 209 - let paths = vec![ 210 - store.parse_store_path(WITH_DEPS_A.path()).unwrap(), 211 - store.parse_store_path(NO_DEPS.path()).unwrap(), 212 - ]; 203 + let paths = vec![ 204 + store.parse_store_path(WITH_DEPS_A.path()).unwrap(), 205 + store.parse_store_path(NO_DEPS.path()).unwrap(), 206 + ]; 213 207 214 - let actual: HashSet<StorePath> = store 215 - .compute_fs_closure_multi(paths, false, false, false) 216 - .await 217 - .expect("Could not compute closure") 218 - .into_iter() 219 - .collect(); 208 + let actual: HashSet<StorePath> = store 209 + .compute_fs_closure_multi(paths, false, false, false) 210 + .await 211 + .expect("Could not compute closure") 212 + .into_iter() 213 + .collect(); 220 214 221 - eprintln!("Closure: {:#?}", actual); 215 + eprintln!("Closure: {:#?}", actual); 222 216 223 - assert_eq!(expected, actual); 224 - }); 217 + assert_eq!(expected, actual); 225 218 } 226 219 227 - #[test] 228 - fn test_query_path_info() { 229 - let store = NixStore::connect().expect("Failed to connect to the Nix store"); 220 + #[tokio::test] 221 + async fn test_query_path_info() { 222 + use test_nar::{WITH_DEPS_B, WITH_DEPS_C}; 230 223 231 - block_on(async move { 232 - use test_nar::{WITH_DEPS_B, WITH_DEPS_C}; 224 + let store = NixStore::connect().expect("Failed to connect to the Nix store"); 233 225 234 - for nar in [WITH_DEPS_C, WITH_DEPS_B] { 235 - nar.import().await.expect("Could not import test NAR"); 236 - } 226 + for nar in [WITH_DEPS_C, WITH_DEPS_B] { 227 + nar.import().await.expect("Could not import test NAR"); 228 + } 237 229 238 - let nar = WITH_DEPS_B; 239 - let path = store.parse_store_path(nar.path()).unwrap(); 240 - let path_info = store 241 - .query_path_info(path) 242 - .await 243 - .expect("Could not query path info"); 230 + let nar = WITH_DEPS_B; 231 + let path = store.parse_store_path(nar.path()).unwrap(); 232 + let path_info = store 233 + .query_path_info(path) 234 + .await 235 + .expect("Could not query path info"); 244 236 245 - eprintln!("Path info: {:?}", path_info); 237 + eprintln!("Path info: {:?}", path_info); 246 238 247 - assert_eq!(nar.nar().len() as u64, path_info.nar_size); 248 - assert_eq!( 249 - vec![PathBuf::from( 250 - "3k1wymic8p7h5pfcqfhh0jan8ny2a712-attic-test-with-deps-c-final" 251 - ),], 252 - path_info.references 253 - ); 254 - }); 239 + assert_eq!(nar.nar().len() as u64, path_info.nar_size); 240 + assert_eq!( 241 + vec![PathBuf::from( 242 + "3k1wymic8p7h5pfcqfhh0jan8ny2a712-attic-test-with-deps-c-final" 243 + ),], 244 + path_info.references 245 + ); 255 246 }
+13 -16
attic/src/stream.rs
··· 176 176 use bytes::{BufMut, BytesMut}; 177 177 use futures::future; 178 178 use tokio::io::AsyncReadExt; 179 - use tokio_test::block_on; 180 179 181 - #[test] 182 - fn test_stream_hasher() { 180 + #[tokio::test] 181 + async fn test_stream_hasher() { 183 182 let expected = b"hello world"; 184 183 let expected_sha256 = 185 184 hex::decode("b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9") ··· 191 190 // force multiple reads 192 191 let mut buf = vec![0u8; 100]; 193 192 let mut bytes_read = 0; 194 - bytes_read += block_on(read.read(&mut buf[bytes_read..bytes_read + 5])).unwrap(); 195 - bytes_read += block_on(read.read(&mut buf[bytes_read..bytes_read + 5])).unwrap(); 196 - bytes_read += block_on(read.read(&mut buf[bytes_read..bytes_read + 5])).unwrap(); 197 - bytes_read += block_on(read.read(&mut buf[bytes_read..bytes_read + 5])).unwrap(); 193 + bytes_read += read.read(&mut buf[bytes_read..bytes_read + 5]).await.unwrap(); 194 + bytes_read += read.read(&mut buf[bytes_read..bytes_read + 5]).await.unwrap(); 195 + bytes_read += read.read(&mut buf[bytes_read..bytes_read + 5]).await.unwrap(); 196 + bytes_read += read.read(&mut buf[bytes_read..bytes_read + 5]).await.unwrap(); 198 197 199 198 assert_eq!(expected.len(), bytes_read); 200 199 assert_eq!(expected, &buf[..bytes_read]); ··· 206 205 eprintln!("finalized = {:x?}", finalized); 207 206 } 208 207 209 - #[test] 210 - fn test_merge_chunks() { 208 + #[tokio::test] 209 + async fn test_merge_chunks() { 211 210 let chunk_a: BoxStream<Result<Bytes, ()>> = { 212 211 let s = stream! { 213 212 yield Ok(Bytes::from_static(b"Hello")); ··· 236 235 let streamer = |c, _| future::ok(c); 237 236 let mut merged = merge_chunks(chunks, streamer, (), 2); 238 237 239 - let bytes = block_on(async move { 240 - let mut bytes = BytesMut::with_capacity(100); 241 - while let Some(item) = merged.next().await { 242 - bytes.put(item.unwrap()); 243 - } 244 - bytes.freeze() 245 - }); 238 + let mut bytes = BytesMut::with_capacity(100); 239 + while let Some(item) = merged.next().await { 240 + bytes.put(item.unwrap()); 241 + } 242 + let bytes = bytes.freeze(); 246 243 247 244 assert_eq!(&*bytes, b"Hello, world!"); 248 245 }
+24
attic/src/testing/mod.rs
··· 1 1 //! Utilities for testing. 2 2 3 3 pub mod shadow_store; 4 + 5 + use tokio::runtime::Runtime; 6 + 7 + /// Returns a new Tokio runtime. 8 + pub fn get_runtime() -> Runtime { 9 + tokio::runtime::Builder::new_current_thread() 10 + .enable_time() 11 + .build() 12 + .unwrap() 13 + } 14 + 15 + /// Returns some fake data. 16 + pub fn get_fake_data(len: usize) -> Vec<u8> { 17 + let mut state = 42u32; 18 + let mut data = vec![0u8; len]; 19 + 20 + for (i, byte) in data.iter_mut().enumerate() { 21 + (state, _) = state.overflowing_mul(1664525u32); 22 + (state, _) = state.overflowing_add(1013904223u32); 23 + *byte = ((state >> (i % 24)) & 0xff) as u8; 24 + } 25 + 26 + data 27 + }
+3
flake.nix
··· 138 138 flyctl 139 139 140 140 wrk 141 + 142 + llvmPackages_latest.bintools 143 + wrangler worker-build wasm-pack wasm-bindgen-cli 141 144 ] ++ (lib.optionals pkgs.stdenv.isLinux [ 142 145 linuxPackages.perf 143 146 ]);
+1 -1
integration-tests/default.nix
··· 33 33 } 34 34 ]; 35 35 }; 36 - }) (lib.cartesianProductOfSets matrix)); 36 + }) (lib.cartesianProduct matrix)); 37 37 in { 38 38 } // basicTests
+1 -5
server/Cargo.toml
··· 19 19 doc = false 20 20 21 21 [dependencies] 22 - attic = { path = "../attic", default-features = false, features = [ "tokio" ] } 22 + attic = { path = "../attic", default-features = false, features = ["chunking", "stream", "tokio"] } 23 23 attic-token = { path = "../token" } 24 24 25 25 anyhow = "1.0.71" ··· 37 37 digest = "0.10.7" 38 38 displaydoc = "0.2.4" 39 39 enum-as-inner = "0.6.0" 40 - fastcdc = "3.0.3" 41 40 futures = "0.3.28" 42 41 hex = "0.4.3" 43 42 http-body-util = "0.1.1" ··· 95 94 "rt-multi-thread", 96 95 "sync", 97 96 ] 98 - 99 - [dev-dependencies] 100 - tokio-test = "0.4.2"
+1 -1
server/src/api/v1/upload_path.rs
··· 37 37 UploadPathNarInfo, UploadPathResult, UploadPathResultKind, ATTIC_NAR_INFO, 38 38 ATTIC_NAR_INFO_PREAMBLE_SIZE, 39 39 }; 40 + use attic::chunking::chunk_stream; 40 41 use attic::hash::Hash; 41 42 use attic::stream::{read_chunk_async, StreamHasher}; 42 43 use attic::util::Finally; 43 44 44 - use crate::chunking::chunk_stream; 45 45 use crate::database::entity::cache; 46 46 use crate::database::entity::chunk::{self, ChunkState, Entity as Chunk}; 47 47 use crate::database::entity::chunkref::{self, Entity as ChunkRef};
-116
server/src/chunking/mod.rs
··· 1 - //! Chunking. 2 - //! 3 - //! We perform chunking on uncompressed NARs using the FastCDC 4 - //! algorithm. 5 - 6 - use async_stream::try_stream; 7 - use bytes::{BufMut, Bytes, BytesMut}; 8 - use fastcdc::ronomon::FastCDC; 9 - use futures::stream::Stream; 10 - use tokio::io::AsyncRead; 11 - 12 - use attic::stream::read_chunk_async; 13 - 14 - /// Splits a streams into content-defined chunks. 15 - /// 16 - /// This is a wrapper over fastcdc-rs that takes an `AsyncRead` and 17 - /// returns a `Stream` of chunks as `Bytes`s. 18 - pub fn chunk_stream<R>( 19 - mut stream: R, 20 - min_size: usize, 21 - avg_size: usize, 22 - max_size: usize, 23 - ) -> impl Stream<Item = std::io::Result<Bytes>> 24 - where 25 - R: AsyncRead + Unpin + Send, 26 - { 27 - let s = try_stream! { 28 - let mut buf = BytesMut::with_capacity(max_size); 29 - 30 - loop { 31 - let read = read_chunk_async(&mut stream, buf).await?; 32 - 33 - let mut eof = false; 34 - if read.is_empty() { 35 - // Already EOF 36 - break; 37 - } else if read.len() < max_size { 38 - // Last read 39 - eof = true; 40 - } 41 - 42 - let chunks = FastCDC::with_eof(&read, min_size, avg_size, max_size, eof); 43 - let mut consumed = 0; 44 - 45 - for chunk in chunks { 46 - consumed += chunk.length; 47 - 48 - let slice = read.slice(chunk.offset..chunk.offset + chunk.length); 49 - yield slice; 50 - } 51 - 52 - if eof { 53 - break; 54 - } 55 - 56 - buf = BytesMut::with_capacity(max_size); 57 - 58 - if consumed < read.len() { 59 - // remaining bytes for the next read 60 - buf.put_slice(&read[consumed..]); 61 - } 62 - } 63 - }; 64 - 65 - Box::pin(s) 66 - } 67 - 68 - #[cfg(test)] 69 - mod tests { 70 - use super::*; 71 - 72 - use std::io::Cursor; 73 - 74 - use futures::StreamExt; 75 - use tokio_test::block_on; 76 - 77 - /// Chunks and reconstructs a file. 78 - #[test] 79 - fn test_chunking_basic() { 80 - fn case(size: usize) { 81 - block_on(async move { 82 - let test_file = get_data(size); // 32 MiB 83 - let mut reconstructed_file = Vec::new(); 84 - 85 - let cursor = Cursor::new(&test_file); 86 - let mut chunks = chunk_stream(cursor, 8 * 1024, 16 * 1024, 32 * 1024); 87 - 88 - while let Some(chunk) = chunks.next().await { 89 - let chunk = chunk.unwrap(); 90 - eprintln!("Got a {}-byte chunk", chunk.len()); 91 - reconstructed_file.extend(chunk); 92 - } 93 - 94 - assert_eq!(reconstructed_file, test_file); 95 - }); 96 - } 97 - 98 - case(32 * 1024 * 1024 - 1); 99 - case(32 * 1024 * 1024); 100 - case(32 * 1024 * 1024 + 1); 101 - } 102 - 103 - /// Returns some fake data. 104 - fn get_data(len: usize) -> Vec<u8> { 105 - let mut state = 42u32; 106 - let mut data = vec![0u8; len]; 107 - 108 - for i in 0..data.len() { 109 - (state, _) = state.overflowing_mul(1664525u32); 110 - (state, _) = state.overflowing_add(1013904223u32); 111 - data[i] = ((state >> (i % 24)) & 0xff) as u8; 112 - } 113 - 114 - data 115 - } 116 - }
+4 -4
server/src/gc.rs
··· 159 159 let storage = state.storage().await?; 160 160 161 161 let orphan_chunk_limit = match db.get_database_backend() { 162 - // Arbitrarily chosen sensible value since there's no good default to choose from for MySQL 163 - sea_orm::DatabaseBackend::MySql => 1000, 162 + // Arbitrarily chosen sensible value since there's no good default to choose from for MySQL 163 + sea_orm::DatabaseBackend::MySql => 1000, 164 164 // Panic limit set by sqlx for postgresql: https://github.com/launchbadge/sqlx/issues/671#issuecomment-687043510 165 165 sea_orm::DatabaseBackend::Postgres => u64::from(u16::MAX), 166 - // Default statement limit imposed by sqlite: https://www.sqlite.org/limits.html#max_variable_number 167 - sea_orm::DatabaseBackend::Sqlite => 500, 166 + // Default statement limit imposed by sqlite: https://www.sqlite.org/limits.html#max_variable_number 167 + sea_orm::DatabaseBackend::Sqlite => 500, 168 168 }; 169 169 170 170 // find all orphan chunks...
-1
server/src/lib.rs
··· 15 15 16 16 pub mod access; 17 17 mod api; 18 - mod chunking; 19 18 pub mod config; 20 19 pub mod database; 21 20 pub mod error;
+2 -2
server/src/storage/local.rs
··· 79 79 let name = file.file_name(); 80 80 let name_bytes = name.as_os_str().as_bytes(); 81 81 let parents = storage_path 82 - .join(OsStr::from_bytes(&name_bytes[0..1])) 83 - .join(OsStr::from_bytes(&name_bytes[0..2])); 82 + .join(OsStr::from_bytes(&name_bytes[0..1])) 83 + .join(OsStr::from_bytes(&name_bytes[0..2])); 84 84 let new_path = parents.join(name); 85 85 fs::create_dir_all(&parents).await.map_err(|e| { 86 86 ErrorKind::StorageError(anyhow::anyhow!("Failed to create directory {}", e))