diff --git a/tools/edge-bench-rs/Cargo.lock b/tools/edge-bench-rs/Cargo.lock index d7faff5..93df9e0 100644 --- a/tools/edge-bench-rs/Cargo.lock +++ b/tools/edge-bench-rs/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aho-corasick" version = "1.1.5" @@ -32,6 +38,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "block-buffer" version = "0.10.4" @@ -47,6 +59,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "cc" version = "1.6.0" @@ -93,6 +111,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01a7799fd6b852db0e61728dde9a204c423b44d689dbd432522543614b490e78" +dependencies = [ + "cfg-if", +] + [[package]] name = "crypto-common" version = "0.1.7" @@ -117,7 +144,7 @@ dependencies = [ name = "edge-bench-rs" version = "0.1.0" dependencies = [ - "base64", + "base64 0.22.1", "chrono", "flate2", "libdeflater", @@ -126,10 +153,24 @@ dependencies = [ "mimalloc", "regex", "rustc-hash", + "serde", + "serde_json", "sha1", + "signal-hook", + "ureq", "zstd", ] +[[package]] +name = "errno" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "find-msvc-tools" version = "0.1.14" @@ -142,6 +183,8 @@ version = "1.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" dependencies = [ + "crc32fast", + "miniz_oxide", "zlib-rs", ] @@ -179,6 +222,17 @@ dependencies = [ "version_check", ] +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -190,6 +244,22 @@ dependencies = [ "r-efi", ] +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -214,13 +284,19 @@ dependencies = [ "cc", ] +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "jobserver" version = "0.1.35" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" dependencies = [ - "getrandom", + "getrandom 0.4.3", "libc", ] @@ -299,6 +375,16 @@ dependencies = [ "libmimalloc-sys", ] +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -314,6 +400,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -379,18 +471,110 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rustc-hash" version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" +[[package]] +name = "rustls" +version = "0.23.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" +dependencies = [ + "log", + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "sha1" version = "0.10.7" @@ -408,12 +592,44 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b29ffa27d3d59a049bb328258de100f8012848d1b08d444e08bbc02ac468435e" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "slab" version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -448,12 +664,59 @@ version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "ureq" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a7ac20be9b7726e0bbdbf974c059676d9acb1cd414961f570a4e8231cacd7fc" +dependencies = [ + "base64 0.23.1", + "flate2", + "log", + "percent-encoding", + "rustls", + "rustls-pki-types", + "ureq-proto", + "utf8-zero", + "webpki-roots", +] + +[[package]] +name = "ureq-proto" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f86fd172ccca569e458f61b6bdd6220965a9ef36e672a6852953b51a0e1583be" +dependencies = [ + "base64 0.23.1", + "http", + "httparse", + "log", +] + +[[package]] +name = "utf8-zero" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8c0a043c9540bae7c578c88f91dda8bd82e59ae27c21baca69c8b191aaf5a6e" + [[package]] name = "version_check" version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasm-bindgen" version = "0.2.129" @@ -499,6 +762,15 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "webpki-roots" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -558,12 +830,106 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "zeroize" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e13084392c5e4bc371903e2935a5eaeed24905a7511356b883835e18a78f6879" + [[package]] name = "zlib-rs" version = "0.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" + [[package]] name = "zstd" version = "0.13.3" diff --git a/tools/edge-bench-rs/Cargo.toml b/tools/edge-bench-rs/Cargo.toml index 0a3f914..ea31165 100644 --- a/tools/edge-bench-rs/Cargo.toml +++ b/tools/edge-bench-rs/Cargo.toml @@ -14,7 +14,11 @@ memchr = "2.8.3" mimalloc = "0.1.52" regex = "1" rustc-hash = "2" +serde = { version = "1.0.229", features = ["derive"] } +serde_json = "1.0.151" sha1 = "0.10" +signal-hook = "0.4.5" +ureq = "3.4.2" zstd = "0.13" [profile.release] diff --git a/tools/edge-bench-rs/check.ts b/tools/edge-bench-rs/check.ts index c4a4a51..45fb705 100644 --- a/tools/edge-bench-rs/check.ts +++ b/tools/edge-bench-rs/check.ts @@ -4,13 +4,16 @@ * payload decoded with lib/chunks), samples, probe candidates, rule counts, * fragments and batch boundaries. * - * node --import=@poppinss/ts-exec tools/edge-bench-rs/check.ts [--batch-chunks 10] [--idle 5] + * node --import=@poppinss/ts-exec tools/edge-bench-rs/check.ts [--wire batches.jsonl] [--batch-chunks 10] [--idle 5] + * + * With --wire (edge-bench-rs --wire): also each batch as the Rust worker + * posts it, against toWire of the TypeScript batch, field by field. */ import { readFile, readdir } from 'node:fs/promises' import { join } from 'node:path' import { unzipSync } from 'node:zlib' import { isDeepStrictEqual } from 'node:util' -import { RangeIndexer, type EdgeBatch } from '#lib/edge/index' +import { RangeIndexer, toWire, type EdgeBatch } from '#lib/edge/index' import { DEFAULT_RULES, type Rules } from '#lib/rules/index' import { decodeChunk } from '#lib/chunks/index' @@ -97,4 +100,40 @@ for (const kind of Object.keys(keyOf)) { } } console.log(`payload bytes: ${payloadBytes.ts} TypeScript, ${payloadBytes.rs} Rust`) + +const wireIndex = process.argv.indexOf('--wire') +if (wireIndex !== -1) { + // Order within a batch doesn't count; payloads by what they decode to. + const by = (f: (x: any) => string) => (a: any, b: any) => (f(a) < f(b) ? -1 : f(a) > f(b) ? 1 : 0) + const normal = (w: any) => ({ + ...w, + releases: w.releases.map((r: any) => ({ ...r, payload: sortFiles(decodeChunk(Buffer.from(r.payload, 'base64'))) })).sort(by((r) => `${r.key} ${r.upload}`)), + samples: w.samples.map((r: any) => ({ ...r, payload: sortFiles(decodeChunk(Buffer.from(r.payload, 'base64'))) })).sort(by((r) => `${r.key} ${r.upload}`)), + candidates: [...w.candidates].sort(by((c) => JSON.stringify(c))), + counts: [...w.counts].sort(by((c) => `${c.rule} ${c.key}`)), + fragments: [...w.fragments].sort(by((f) => `${f.fileKey} ${f.segments[0]?.[1]}`)), + }) + const rust = (await readFile(process.argv[wireIndex + 1], 'utf8')).split('\n').filter(Boolean).map((l) => normal(JSON.parse(l))) + const typescript = batches.map((b) => normal(JSON.parse(JSON.stringify(toWire(b))))) + let wireSame = rust.length === typescript.length + const firstDiff = (x: any, y: any, path: string): string | null => { + if (isDeepStrictEqual(x, y)) return null + if (typeof x !== 'object' || typeof y !== 'object' || x === null || y === null) return `${path}: ${JSON.stringify(x)?.slice(0, 120)} vs ${JSON.stringify(y)?.slice(0, 120)}` + for (const k of new Set([...Object.keys(x), ...Object.keys(y)])) { + const d = firstDiff(x[k], y[k], `${path}.${k}`) + if (d) return d + } + return path + } + for (let i = 0; i < Math.max(rust.length, typescript.length); i++) { + const d = firstDiff(typescript[i], rust[i], `batch ${i}`) + if (d) { + wireSame = false + console.log(`DIFF wire: ${d}`) + break + } + } + if (wireSame) console.log(`same wire batches: ${typescript.length} TypeScript, ${rust.length} Rust`) + failed ||= !wireSame +} process.exit(failed ? 1 : 0) diff --git a/tools/edge-bench-rs/src/indexer.rs b/tools/edge-bench-rs/src/indexer.rs index c170614..40fd5a7 100644 --- a/tools/edge-bench-rs/src/indexer.rs +++ b/tools/edge-bench-rs/src/indexer.rs @@ -4,7 +4,7 @@ use crate::codec::{encode_chunk, ChunkFile}; use crate::js::{self, sha1_hex}; -use crate::names::{self, is_random_file, JudgedFile, Release, Verdict}; +use crate::names::{self, is_random_file, JudgedFile, Release, Rules, Verdict}; use rustc_hash::{FxHashMap, FxHashSet}; use std::rc::Rc; @@ -60,7 +60,11 @@ pub struct Candidate { pub struct Fragment { pub file_key: String, pub name: String, + pub poster: String, + pub subject: String, pub total_parts: u32, + pub groups: Vec, + pub release: Release, pub posted_at: Option, pub segments: Vec<(u32, String, i64, i64)>, } @@ -441,6 +445,7 @@ fn uploads_of(files: Vec>) -> Vec>> { pub struct RangeIndexer { group: String, + rules: Rules, start: i64, end: i64, priorities: FxHashMap, @@ -456,17 +461,19 @@ pub struct RangeIndexer { } impl RangeIndexer { - pub fn new(group: &str, start: i64, end: i64, priorities: FxHashMap, idle: u64) -> Self { + /// `resume_at`: where this worker starts (a range taken over midway); its first batch starts there. + pub fn new(group: &str, start: i64, end: i64, resume_at: i64, rules: Rules, idle: u64) -> Self { RangeIndexer { group: group.to_string(), start, end, - priorities, + priorities: rules.priorities.clone(), + rules, idle, chunk_articles: 0, scanner: Scanner::default(), aggregator: Aggregator::default(), - batch: Batch { from: start, next: start, ..Default::default() }, + batch: Batch { from: resume_at, next: resume_at, ..Default::default() }, counts: FxHashMap::default(), count_order: Vec::new(), completed: FxHashMap::default(), @@ -531,6 +538,11 @@ impl RangeIndexer { self.batch.next = self.end + 1; } + /// Files completed and fragments cut since the last take (RangeIndexer.pending). + pub fn pending(&self) -> usize { + self.batch.fragments.len() + self.completed.values().map(|l| l.len()).sum::() + } + pub fn take(&mut self) -> Batch { let t = std::time::Instant::now(); self.judge_releases(); @@ -567,7 +579,7 @@ impl RangeIndexer { .iter() .map(|f| JudgedFile { name: &f[0].file.name, total_parts: f[0].file.total_parts, groups: &f[0].file.groups, first_message_id: &f[0].message_id }) .collect(); - match names::judge(&key, &judged) { + match names::judge(&key, &judged, &self.rules) { Verdict::Keep => { let candidates = candidates_of(&key, &list, upload); self.batch.releases.push(chunk_of(&key, &list, upload)); @@ -603,7 +615,17 @@ impl RangeIndexer { continue; } let f = &list[0].file; - let mut fragment = Fragment { file_key: f.file_key.clone(), name: f.name.clone(), total_parts: f.total_parts, posted_at: None, segments: Vec::with_capacity(list.len()) }; + let mut fragment = Fragment { + file_key: f.file_key.clone(), + name: f.name.clone(), + poster: f.poster.clone(), + subject: f.subject.clone(), + total_parts: f.total_parts, + groups: f.groups.clone(), + release: f.release.clone(), + posted_at: None, + segments: Vec::with_capacity(list.len()), + }; for s in &list { fragment.segments.push((s.number, s.message_id.clone(), s.bytes, s.article)); if let Some(t) = s.posted_at { @@ -736,7 +758,7 @@ mod tests { fn uploads_split_as_in_the_typescript() { let mut priorities = FxHashMap::default(); priorities.insert(GROUP.to_string(), 1); - let mut ix = RangeIndexer::new(GROUP, 1, 10_000, priorities, 5); + let mut ix = RangeIndexer::new(GROUP, 1, 10_000, 1, Rules::defaults(priorities), 5); let mut lines: Vec = (0..5).map(|i| line(100 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), i * 60)).collect(); lines.push(line(200, r#"Big.Release [2/2] - "Big.Release.par2" yEnc (1/1)"#, 245)); lines.extend((0..5).map(|i| line(300 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), 1440 + i))); diff --git a/tools/edge-bench-rs/src/lib.rs b/tools/edge-bench-rs/src/lib.rs new file mode 100644 index 0000000..3b8ca99 --- /dev/null +++ b/tools/edge-bench-rs/src/lib.rs @@ -0,0 +1,9 @@ +//! The edge worker's indexing (lib/edge RangeIndexer and what it uses) in +//! Rust, kept identical to the TypeScript (check.ts): the edge-bench-rs +//! benchmark and the edge-node worker share it. + +pub mod codec; +pub mod indexer; +pub mod js; +pub mod names; +pub mod wire; diff --git a/tools/edge-bench-rs/src/main.rs b/tools/edge-bench-rs/src/main.rs index 7f7241c..f7f8264 100644 --- a/tools/edge-bench-rs/src/main.rs +++ b/tools/edge-bench-rs/src/main.rs @@ -4,16 +4,12 @@ //! idle ones, and every N chunks judge and encode the releases. `--dump` //! writes every output as JSON lines for tools/edge-bench-rs/check.ts. //! -//! edge-bench-rs [--batch-chunks 10] [--idle 5] [--dump out.jsonl] - -mod codec; -mod indexer; -mod js; -mod names; +//! edge-bench-rs [--batch-chunks 10] [--idle 5] [--dump out.jsonl] [--wire batches.jsonl] use base64::Engine; -use indexer::{Batch, RangeIndexer}; -use js::json_str; +use edge_bench_rs::indexer::{self, Batch, RangeIndexer}; +use edge_bench_rs::js::json_str; +use edge_bench_rs::{names, wire}; use rustc_hash::FxHashMap; use std::io::Write; use std::time::Instant; @@ -159,6 +155,8 @@ fn main() { let batch_chunks: usize = flag("--batch-chunks").map(|v| v.parse().unwrap()).unwrap_or(10); let idle: u64 = flag("--idle").map(|v| v.parse().unwrap()).unwrap_or(5); let mut dump = flag("--dump").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap())); + // Each batch as the worker posts it (wire.rs), one JSON per line: check.ts compares it with toWire. + let mut wire_out = flag("--wire").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap())); let range = std::fs::read_to_string(format!("{dir}/range.json")).unwrap(); let group = json_field(&range, "group"); @@ -169,7 +167,7 @@ fn main() { let mut priorities = FxHashMap::default(); priorities.insert(group.clone(), 1); - let mut indexer = RangeIndexer::new(&group, start, end, priorities, idle); + let mut indexer = RangeIndexer::new(&group, start, end, start, names::Rules::defaults(priorities), idle); let (mut inflate_s, mut index_s) = (0.0, 0.0); let (mut compressed, mut text_bytes) = (0usize, 0usize); @@ -212,6 +210,10 @@ fn main() { if let Some(d) = dump.as_mut() { dump_batch(&batch, d); } + if let Some(w) = wire_out.as_mut() { + w.write_all(&wire::to_json(&batch)).unwrap(); + w.write_all(b"\n").unwrap(); + } } let total = inflate_s + index_s; #[allow(static_mut_refs)] diff --git a/tools/edge-bench-rs/src/names.rs b/tools/edge-bench-rs/src/names.rs index e0a619f..ad2e411 100644 --- a/tools/edge-bench-rs/src/names.rs +++ b/tools/edge-bench-rs/src/names.rs @@ -806,12 +806,40 @@ pub enum Verdict { Drop { rule: &'static str, key: String, sample: bool }, } -/// RuleSet.judge with DEFAULT_RULES (no dropped signatures). -pub fn judge(release_key: &str, files: &[JudgedFile]) -> Verdict { - const DROPPED_TOOLS: [&str; 5] = ["ngPost", "hex32", "yuun", "JBinUp", "random"]; - let sample = explored(release_key, 1.0); +/// The rules a lease brings (lib/rules Rules): what the worker drops. +#[derive(Clone, Debug)] +pub struct Rules { + pub version: i64, + /// Indexed groups and their priority (owner rule). + pub priorities: rustc_hash::FxHashMap, + /// "tool\tgroup" + pub dropped_signatures: rustc_hash::FxHashSet, + pub dropped_tools: rustc_hash::FxHashSet, + pub per_article_max_parts: u32, + pub sample_percent: f64, + pub drop_executables: bool, +} + +impl Rules { + /// DEFAULT_RULES with these priorities and no dropped signatures (edge:bench's). + pub fn defaults(priorities: rustc_hash::FxHashMap) -> Self { + Rules { + version: 1, + priorities, + dropped_signatures: Default::default(), + dropped_tools: ["ngPost", "hex32", "yuun", "JBinUp", "random"].iter().map(|t| t.to_string()).collect(), + per_article_max_parts: 3, + sample_percent: 1.0, + drop_executables: true, + } + } +} + +/// RuleSet.judge. +pub fn judge(release_key: &str, files: &[JudgedFile], rules: &Rules) -> Verdict { + let sample = explored(release_key, rules.sample_percent); let content: Vec<&JudgedFile> = files.iter().filter(|f| !test(&SIDECAR, f.name)).collect(); - if !content.is_empty() && content.iter().all(|f| test(&EXECUTABLE, f.name)) { + if rules.drop_executables && !content.is_empty() && content.iter().all(|f| test(&EXECUTABLE, f.name)) { let c = EXECUTABLE.captures(content[0].name.as_bytes()).unwrap(); let key = std::str::from_utf8(&c[1]).unwrap().to_lowercase(); return Verdict::Drop { rule: "executable", key, sample }; @@ -822,10 +850,10 @@ pub fn judge(release_key: &str, files: &[JudgedFile]) -> Verdict { let first = &files[0]; let tool = upload_tool(first.first_message_id); let group = first.groups.first().map(String::as_str).unwrap_or(""); - if DROPPED_TOOLS.contains(&tool) { + if rules.dropped_tools.contains(tool) || rules.dropped_signatures.contains(&format!("{tool}\t{group}")) { return Verdict::Drop { rule: "garbage", key: format!("{tool} {group}"), sample }; } - if files.iter().all(|f| f.total_parts <= 3) { + if files.iter().all(|f| f.total_parts <= rules.per_article_max_parts) { return Verdict::Drop { rule: "per-article", key: tool.to_string(), sample }; } Verdict::Keep diff --git a/tools/edge-bench-rs/src/wire.rs b/tools/edge-bench-rs/src/wire.rs new file mode 100644 index 0000000..f593c99 --- /dev/null +++ b/tools/edge-bench-rs/src/wire.rs @@ -0,0 +1,144 @@ +//! A batch as the coordinator takes it (lib/edge/wire.ts toWire): JSON with +//! the TypeScript names, chunk payloads in base64. + +use crate::indexer::{Batch, Candidate, Fragment, ReleaseChunk}; +use base64::Engine; +use serde::Serialize; + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WireRelease<'a> { + key: &'a str, + upload: usize, + name: &'a str, + source: &'a str, + file_total: Option, + poster: &'a str, + groups: &'a [String], + posted_at: Option, + posted_last: Option, + files: usize, + segments: usize, + bytes: i64, + first_article: i64, + last_article: i64, + readable: bool, + has_par2: bool, + payload: String, + #[serde(skip_serializing_if = "Option::is_none")] + rule: Option<&'a str>, + #[serde(skip_serializing_if = "Option::is_none")] + rule_key: Option<&'a str>, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WireCandidate<'a> { + release_key: &'a str, + upload: usize, + kind: &'a str, + message_id: &'a str, + file: &'a str, +} + +#[derive(Serialize)] +struct WireCount<'a> { + rule: &'a str, + key: &'a str, + articles: u64, + files: u64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WireFragmentRelease<'a> { + key: &'a str, + name: &'a str, + source: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + file_total: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WireFragment<'a> { + file_key: &'a str, + name: &'a str, + poster: &'a str, + subject: &'a str, + total_parts: u32, + groups: &'a [String], + release: Option>, + posted_at: Option, + segments: Vec<(u32, &'a str, i64, i64)>, +} + +#[derive(Serialize)] +struct WireBatch<'a> { + from: i64, + next: i64, + headers: u64, + releases: Vec>, + samples: Vec>, + candidates: Vec>, + counts: Vec>, + fragments: Vec>, +} + +fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRelease<'a> { + WireRelease { + key: &r.key, + upload: r.upload, + name: &r.name, + source: r.source, + file_total: r.file_total, + poster: &r.poster, + groups: &r.groups, + posted_at: r.posted_at, + posted_last: r.posted_last, + files: r.files, + segments: r.segments, + bytes: r.bytes, + first_article: r.first_article, + last_article: r.last_article, + readable: r.readable, + has_par2: r.has_par2, + payload: base64::engine::general_purpose::STANDARD.encode(&r.payload), + rule: rule.map(|(rule, _)| rule), + rule_key: rule.map(|(_, key)| key), + } +} + +fn candidate(c: &Candidate) -> WireCandidate<'_> { + WireCandidate { release_key: &c.release_key, upload: c.upload, kind: c.kind, message_id: &c.message_id, file: &c.file } +} + +fn fragment(f: &Fragment) -> WireFragment<'_> { + WireFragment { + file_key: &f.file_key, + name: &f.name, + poster: &f.poster, + subject: &f.subject, + total_parts: f.total_parts, + groups: &f.groups, + // A named file always has a release; fileTotal only when truthy, as fragmentOf. + release: Some(WireFragmentRelease { key: &f.release.key, name: &f.release.name, source: f.release.source, file_total: f.release.file_total_truthy() }), + posted_at: f.posted_at, + segments: f.segments.iter().map(|(n, id, bytes, article)| (*n, id.as_str(), *bytes, *article)).collect(), + } +} + +/// toWire(batch) as JSON bytes. +pub fn to_json(batch: &Batch) -> Vec { + let wire = WireBatch { + from: batch.from, + next: batch.next, + headers: batch.headers, + releases: batch.releases.iter().map(|r| release(r, None)).collect(), + samples: batch.samples.iter().map(|(r, rule, key)| release(r, Some((rule, key.as_str())))).collect(), + candidates: batch.candidates.iter().map(candidate).collect(), + counts: batch.counts.iter().map(|(rule, key, articles, files)| WireCount { rule, key, articles: *articles, files: *files }).collect(), + fragments: batch.fragments.iter().map(fragment).collect(), + }; + serde_json::to_vec(&wire).unwrap() +}