diff --git a/agent/cafs-writer/.gitignore b/agent/cafs-writer/.gitignore new file mode 100644 index 0000000000..7fb5043224 --- /dev/null +++ b/agent/cafs-writer/.gitignore @@ -0,0 +1,3 @@ +target/ +Cargo.lock +pnpm-cafs-writer.*.node diff --git a/agent/cafs-writer/Cargo.toml b/agent/cafs-writer/Cargo.toml new file mode 100644 index 0000000000..e3fa8b845f --- /dev/null +++ b/agent/cafs-writer/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "pnpm-cafs-writer" +version = "0.1.0" +edition = "2021" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +napi = { version = "2", default-features = false, features = ["napi4"] } +napi-derive = "2" +rayon = "1" + +[target.'cfg(target_os = "linux")'.dependencies] +tokio-uring = "0.5" + +[build-dependencies] +napi-build = "2" + +[profile.release] +lto = true +codegen-units = 1 +strip = "symbols" diff --git a/agent/cafs-writer/bench/compare.js b/agent/cafs-writer/bench/compare.js new file mode 100644 index 0000000000..a62420d9ff --- /dev/null +++ b/agent/cafs-writer/bench/compare.js @@ -0,0 +1,96 @@ +#!/usr/bin/env node +'use strict' + +// Microbenchmark comparing the native Rust writer against the JS parse+write +// path used by worker/src/start.ts fetchAndWriteCafs (minus the HTTP/gunzip +// framing, so this measures parse + write only). +// +// Run: node bench/compare.js [FILE_COUNT] [FILE_SIZE_BYTES] [ITERATIONS] + +const crypto = require('node:crypto') +const fs = require('node:fs') +const os = require('node:os') +const path = require('node:path') + +const { writeFiles: nativeWriteFiles } = require('..') + +const FILE_COUNT = parseInt(process.argv[2] ?? '5000', 10) +const FILE_SIZE = parseInt(process.argv[3] ?? '4096', 10) +const ITERATIONS = parseInt(process.argv[4] ?? '3', 10) + +function buildPayload (fileCount, fileSize) { + const parts = [] + const lenBuf = Buffer.alloc(4) + lenBuf.writeUInt32BE(2, 0) + parts.push(lenBuf, Buffer.from('{}')) + + for (let i = 0; i < fileCount; i++) { + // Unique content per file so digests are unique + const content = crypto.randomBytes(fileSize) + const digest = crypto.hash('sha512', content, 'hex') + parts.push(Buffer.from(digest, 'hex')) + const sizeBuf = Buffer.alloc(4) + sizeBuf.writeUInt32BE(fileSize, 0) + parts.push(sizeBuf, Buffer.from([i & 1 ? 0x01 : 0x00]), content) + } + parts.push(Buffer.alloc(64, 0)) + return Buffer.concat(parts) +} + +function jsWriteFiles (storeDir, payload) { + let pos = 0 + const jsonLen = payload.readUInt32BE(pos); pos += 4 + pos += jsonLen + + const END_MARKER = Buffer.alloc(64, 0) + const createdDirs = new Set() + let written = 0 + + while (pos < payload.length) { + const digestBuf = payload.subarray(pos, pos + 64) + if (digestBuf.equals(END_MARKER)) break + pos += 64 + const size = payload.readUInt32BE(pos); pos += 4 + const executable = (payload[pos] & 0x01) !== 0; pos += 1 + const content = payload.subarray(pos, pos + size); pos += size + + const digest = digestBuf.toString('hex') + const dir = path.join(storeDir, 'files', digest.slice(0, 2)) + const file = path.join(dir, executable ? `${digest.slice(2)}-exec` : digest.slice(2)) + if (!createdDirs.has(dir)) { + fs.mkdirSync(dir, { recursive: true }) + createdDirs.add(dir) + } + try { + fs.writeFileSync(file, content, { flag: 'wx', mode: executable ? 0o755 : 0o644 }) + written++ + } catch (err) { + if (err.code !== 'EEXIST') throw err + } + } + return written +} + +function bench (label, fn) { + const tmpBase = fs.mkdtempSync(path.join(os.tmpdir(), 'cafs-bench-')) + try { + const start = process.hrtime.bigint() + const written = fn(tmpBase) + const ms = Number(process.hrtime.bigint() - start) / 1_000_000 + console.log(` ${label.padEnd(10)} ${ms.toFixed(1)}ms (${written} files written)`) + return ms + } finally { + fs.rmSync(tmpBase, { recursive: true, force: true }) + } +} + +console.log(`Files: ${FILE_COUNT} size: ${FILE_SIZE} bytes iterations: ${ITERATIONS}`) +const payload = buildPayload(FILE_COUNT, FILE_SIZE) +console.log(`Payload size: ${(payload.length / 1024 / 1024).toFixed(1)} MB\n`) + +for (let i = 0; i < ITERATIONS; i++) { + console.log(`Iteration ${i + 1}:`) + bench('native', (dir) => nativeWriteFiles(dir, payload)) + bench('js', (dir) => jsWriteFiles(dir, payload)) + console.log() +} diff --git a/agent/cafs-writer/build.rs b/agent/cafs-writer/build.rs new file mode 100644 index 0000000000..bbfc9e4b9e --- /dev/null +++ b/agent/cafs-writer/build.rs @@ -0,0 +1,3 @@ +fn main() { + napi_build::setup(); +} diff --git a/agent/cafs-writer/index.d.ts b/agent/cafs-writer/index.d.ts new file mode 100644 index 0000000000..4c70ff041e --- /dev/null +++ b/agent/cafs-writer/index.d.ts @@ -0,0 +1,5 @@ +/** + * Parse an uncompressed /v1/files payload and write each file to the CAFS. + * Returns the number of files newly written (EEXIST is not counted). + */ +export function writeFiles (storeDir: string, payload: Buffer): number diff --git a/agent/cafs-writer/index.js b/agent/cafs-writer/index.js new file mode 100644 index 0000000000..3f5d1137b2 --- /dev/null +++ b/agent/cafs-writer/index.js @@ -0,0 +1,21 @@ +'use strict' + +// Prototype loader: locates the compiled native addon for the current +// platform/arch and re-exports its functions. Real napi-rs packages ship +// prebuilds per triple and have a more elaborate loader; that's out of +// scope for a prototype. + +const { existsSync } = require('node:fs') +const { join } = require('node:path') + +const triple = `${process.platform}-${process.arch}` +const candidate = join(__dirname, `pnpm-cafs-writer.${triple}.node`) + +if (!existsSync(candidate)) { + throw new Error( + `@pnpm/agent.cafs-writer: native addon not built for ${triple}. ` + + `Run \`pnpm --filter @pnpm/agent.cafs-writer run build\` first.` + ) +} + +module.exports = require(candidate) diff --git a/agent/cafs-writer/package.json b/agent/cafs-writer/package.json new file mode 100644 index 0000000000..336538c4aa --- /dev/null +++ b/agent/cafs-writer/package.json @@ -0,0 +1,22 @@ +{ + "name": "@pnpm/agent.cafs-writer", + "version": "0.1.0", + "description": "Native (Rust + napi-rs) batch writer for files in the pnpm CAFS. Used by the agent client path to cut syscall overhead on large installs.", + "private": true, + "license": "MIT", + "main": "index.js", + "types": "index.d.ts", + "files": [ + "index.js", + "index.d.ts", + "pnpm-cafs-writer.*.node" + ], + "scripts": { + "build": "cargo build --release && node scripts/copy-artifact.js", + "build:debug": "cargo build && node scripts/copy-artifact.js --debug", + "test": "echo 'prototype — no tests'" + }, + "engines": { + "node": ">=22.13" + } +} diff --git a/agent/cafs-writer/scripts/copy-artifact.js b/agent/cafs-writer/scripts/copy-artifact.js new file mode 100644 index 0000000000..fab8a3f9cd --- /dev/null +++ b/agent/cafs-writer/scripts/copy-artifact.js @@ -0,0 +1,36 @@ +#!/usr/bin/env node +'use strict' + +// Copy the cargo-built cdylib to a predictable name next to index.js. +// napi-rs crates produce libpnpm_cafs_writer.{dylib,so,dll}; we rename to +// pnpm-cafs-writer.-.node which is what index.js looks for. + +const { copyFileSync, existsSync } = require('node:fs') +const { join } = require('node:path') + +const debug = process.argv.includes('--debug') +const profile = debug ? 'debug' : 'release' +const root = join(__dirname, '..') + +const sourceByPlatform = { + darwin: 'libpnpm_cafs_writer.dylib', + linux: 'libpnpm_cafs_writer.so', + win32: 'pnpm_cafs_writer.dll', +} + +const sourceName = sourceByPlatform[process.platform] +if (!sourceName) { + console.error(`Unsupported platform: ${process.platform}`) + process.exit(1) +} + +const source = join(root, 'target', profile, sourceName) +if (!existsSync(source)) { + console.error(`Expected cargo artifact not found: ${source}`) + process.exit(1) +} + +const triple = `${process.platform}-${process.arch}` +const dest = join(root, `pnpm-cafs-writer.${triple}.node`) +copyFileSync(source, dest) +console.log(`Wrote ${dest}`) diff --git a/agent/cafs-writer/src/lib.rs b/agent/cafs-writer/src/lib.rs new file mode 100644 index 0000000000..930f165eaf --- /dev/null +++ b/agent/cafs-writer/src/lib.rs @@ -0,0 +1,249 @@ +#![deny(clippy::all)] + +use napi::bindgen_prelude::*; +use napi_derive::napi; +use std::path::PathBuf; + +#[cfg(not(target_os = "linux"))] +use std::{ + fs::OpenOptions, + io::Write, +}; + +#[cfg(unix)] +use std::os::unix::fs::OpenOptionsExt; + +/// Parse an uncompressed /v1/files payload and write each file to the CAFS. +/// +/// Payload format (same as the Node-side parser in worker/src/start.ts): +/// [4-byte BE u32: JSON header length] +/// [N bytes: JSON header (ignored — reserved for future use)] +/// [entries...] +/// [64 zero bytes: end marker] +/// +/// Each entry: +/// [64 bytes: SHA-512 digest, raw binary] +/// [4-byte BE u32: content length] +/// [1 byte: 0x00 = regular, 0x01 = executable] +/// [N bytes: content] +/// +/// Returns the number of files newly written. Files already present (another +/// worker won the race) are silently skipped via O_EXCL and not counted. +#[napi] +pub fn write_files(store_dir: String, payload: Buffer) -> Result { + let bytes: &[u8] = &payload; + let entries = parse_payload(bytes) + .map_err(|e| Error::from_reason(format!("cafs-writer parse error: {e}")))?; + + let files_dir = PathBuf::from(store_dir).join("files"); + pre_create_parent_dirs(&files_dir, &entries) + .map_err(|e| Error::from_reason(format!("cafs-writer mkdir error: {e}")))?; + + #[cfg(target_os = "linux")] + { + linux_uring::write_all(&files_dir, &entries) + .map_err(|e| Error::from_reason(format!("cafs-writer (io_uring) write error: {e}"))) + } + #[cfg(not(target_os = "linux"))] + { + write_all_std(&files_dir, &entries) + .map_err(|e| Error::from_reason(format!("cafs-writer write error: {e}"))) + } +} + +pub(crate) struct Entry<'a> { + pub digest_hex: String, + pub executable: bool, + pub content: &'a [u8], +} + +impl Entry<'_> { + pub fn filename(&self) -> String { + let suffix = &self.digest_hex[2..]; + if self.executable { + format!("{suffix}-exec") + } else { + suffix.to_string() + } + } + + pub fn prefix(&self) -> &str { + &self.digest_hex[..2] + } + + pub fn mode(&self) -> u32 { + if self.executable { 0o755 } else { 0o644 } + } +} + +fn parse_payload(bytes: &[u8]) -> std::result::Result>, String> { + if bytes.len() < 4 { + return Err("payload too small for header length prefix".into()); + } + let json_len = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize; + let mut pos = 4 + json_len; + if pos > bytes.len() { + return Err("payload truncated inside JSON header".into()); + } + + let mut entries = Vec::new(); + loop { + if pos + 64 > bytes.len() { + return Err("payload truncated at digest boundary".into()); + } + let digest = &bytes[pos..pos + 64]; + if digest.iter().all(|&b| b == 0) { + break; + } + pos += 64; + + if pos + 5 > bytes.len() { + return Err("payload truncated at size/mode header".into()); + } + let size = u32::from_be_bytes([bytes[pos], bytes[pos + 1], bytes[pos + 2], bytes[pos + 3]]) + as usize; + let executable = (bytes[pos + 4] & 0x01) != 0; + pos += 5; + + if pos + size > bytes.len() { + return Err("payload truncated inside file content".into()); + } + let content = &bytes[pos..pos + size]; + pos += size; + + entries.push(Entry { + digest_hex: hex_encode(digest), + executable, + content, + }); + } + Ok(entries) +} + +// The store layout normally pre-creates files/XX/ subdirectories (see +// worker init-store), but the CAFS may be empty on first agent-client use. +// Create each needed prefix dir once up front — both the std and io_uring +// paths benefit from not having to handle ENOENT on every file open. +fn pre_create_parent_dirs( + files_dir: &std::path::Path, + entries: &[Entry<'_>], +) -> std::io::Result<()> { + let mut seen = std::collections::HashSet::<&str>::new(); + for e in entries { + if seen.insert(e.prefix()) { + std::fs::create_dir_all(files_dir.join(e.prefix()))?; + } + } + Ok(()) +} + +#[cfg(not(target_os = "linux"))] +fn write_all_std( + files_dir: &std::path::Path, + entries: &[Entry<'_>], +) -> std::result::Result { + use rayon::prelude::*; + entries + .par_iter() + .map(|entry| write_one_std(files_dir, entry)) + .try_reduce(|| 0u32, |a, b| Ok(a + b)) +} + +#[cfg(not(target_os = "linux"))] +fn write_one_std( + files_dir: &std::path::Path, + entry: &Entry, +) -> std::result::Result { + let path = files_dir.join(entry.prefix()).join(entry.filename()); + let mut opts = OpenOptions::new(); + opts.write(true).create_new(true); + #[cfg(unix)] + opts.mode(entry.mode()); + match opts.open(&path) { + Ok(mut f) => { + f.write_all(entry.content) + .map_err(|e| format!("write {}: {e}", path.display()))?; + Ok(1) + } + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Ok(0), + Err(e) => Err(format!("open {}: {e}", path.display())), + } +} + +#[cfg(target_os = "linux")] +mod linux_uring { + //! io_uring write backend. + //! + //! Opens + writes + closes each file via tokio-uring SQEs. The colleague's + //! benchmark data (1000 small files: rayon+std 39s vs tokio_uring 27s) + //! suggests the overlap-syscalls pattern dominates once file count is high. + //! + //! Not yet tuned: we spawn one task per file. The faster variant in the + //! colleague's gist used a `tokio_uring::builder` with SQPOLL; worth trying + //! next if this doesn't meet targets. + use super::Entry; + use std::path::Path; + + pub(super) fn write_all( + files_dir: &Path, + entries: &[Entry<'_>], + ) -> std::result::Result { + tokio_uring::start(async { + let mut tasks = Vec::with_capacity(entries.len()); + for entry in entries { + let path = files_dir.join(entry.prefix()).join(entry.filename()); + let content = entry.content.to_vec(); + let executable = entry.executable; + tasks.push(tokio_uring::spawn(async move { + // tokio-uring 0.5's OpenOptions doesn't expose a .mode() setter, + // so we take the umask default (0o666 & ~umask ≈ 0o644) for + // non-exec files and explicitly chmod executables after writing. + // For the CAFS's purposes, non-exec=0o644 / exec=0o755 is all + // that matters. + let file = tokio_uring::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .await; + let file = match file { + Ok(f) => f, + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => { + return Ok::(0) + } + Err(e) => return Err(format!("open {}: {e}", path.display())), + }; + let (res, _buf) = file.write_all_at(content, 0).await; + res.map_err(|e| format!("write {}: {e}", path.display()))?; + file + .close() + .await + .map_err(|e| format!("close {}: {e}", path.display()))?; + if executable { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)) + .map_err(|e| format!("chmod {}: {e}", path.display()))?; + } + Ok(1) + })); + } + let mut total = 0u32; + for task in tasks { + total += task + .await + .map_err(|e| format!("tokio-uring task join: {e}"))??; + } + Ok(total) + }) + } +} + +// Small, stable hex encoder — avoids pulling in the `hex` crate. +fn hex_encode(bytes: &[u8]) -> String { + const HEX: &[u8; 16] = b"0123456789abcdef"; + let mut out = String::with_capacity(bytes.len() * 2); + for &b in bytes { + out.push(HEX[(b >> 4) as usize] as char); + out.push(HEX[(b & 0x0f) as usize] as char); + } + out +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6a75208858..f06d58194a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1410,6 +1410,8 @@ importers: specifier: workspace:* version: 'link:' + agent/cafs-writer: {} + agent/client: dependencies: '@pnpm/lockfile.types': @@ -1482,7 +1484,7 @@ importers: version: link:../../core/logger '@pnpm/registry-mock': specifier: 'catalog:' - version: 5.2.4(verdaccio@6.3.2(encoding@0.1.13)(typanion@3.14.0)) + version: 6.0.0(encoding@0.1.13)(verdaccio@6.3.2(encoding@0.1.13)(typanion@3.14.0)) cross-env: specifier: 'catalog:' version: 10.1.0 @@ -7552,6 +7554,10 @@ importers: write-yaml-file: specifier: 'catalog:' version: 6.0.0 + optionalDependencies: + '@pnpm/agent.cafs-writer': + specifier: workspace:* + version: link:../agent/cafs-writer pnpm/artifacts/exe: dependencies: @@ -8957,6 +8963,10 @@ importers: '@types/semver': specifier: 'catalog:' version: 7.7.1 + optionalDependencies: + '@pnpm/agent.cafs-writer': + specifier: workspace:* + version: link:../agent/cafs-writer workspace/commands: dependencies: diff --git a/pnpm/package.json b/pnpm/package.json index 77a164ad78..0a5c5bb861 100644 --- a/pnpm/package.json +++ b/pnpm/package.json @@ -75,6 +75,9 @@ "node-gyp": "^12.2.0", "v8-compile-cache": "2.4.0" }, + "optionalDependencies": { + "@pnpm/agent.cafs-writer": "workspace:*" + }, "devDependencies": { "@jest/globals": "catalog:", "@pnpm/assert-project": "workspace:*", diff --git a/worker/package.json b/worker/package.json index e2118caca3..0b29430cce 100644 --- a/worker/package.json +++ b/worker/package.json @@ -32,6 +32,9 @@ "compile": "tsgo --build && pn lint --fix", ".test": "cross-env NODE_OPTIONS=\"$NODE_OPTIONS --experimental-vm-modules --disable-warning=ExperimentalWarning --disable-warning=DEP0169\" jest" }, + "optionalDependencies": { + "@pnpm/agent.cafs-writer": "workspace:*" + }, "dependencies": { "@pnpm/building.pkg-requires-build": "workspace:*", "@pnpm/crypto.integrity": "workspace:*", diff --git a/worker/src/start.ts b/worker/src/start.ts index 2fc1921051..91916506fb 100644 --- a/worker/src/start.ts +++ b/worker/src/start.ts @@ -1,5 +1,6 @@ import crypto from 'node:crypto' import fs from 'node:fs' +import { createRequire } from 'node:module' import path from 'node:path' import { parentPort } from 'node:worker_threads' @@ -500,20 +501,70 @@ function writeCafsFiles (message: WriteCafsFilesMessage): { status: string, file return { status: 'success', filesWritten } } +// Try to load the native CAFS writer. Falls back to the pure-JS streaming +// parser below when the addon hasn't been built for this platform. +let nativeWriteFiles: ((storeDir: string, payload: Buffer) => number) | undefined +if (!process.env.PNPM_CAFS_WRITER_DISABLE_NATIVE) { + try { + const require = createRequire(import.meta.url) + const native = require('@pnpm/agent.cafs-writer') as { writeFiles: typeof nativeWriteFiles } + nativeWriteFiles = native.writeFiles + } catch { + nativeWriteFiles = undefined + } +} + async function fetchAndWriteCafs (message: FetchAndWriteCafsMessage): Promise<{ status: string, filesWritten: number }> { const http = await import('node:http') const https = await import('node:https') const { URL } = await import('node:url') const { createGunzip } = await import('node:zlib') - const { contentPathFromHex } = await import('@pnpm/store.cafs') const url = new URL('/v1/files', message.registryUrl) const requestFn = url.protocol === 'https:' ? https.request : http.request const body = JSON.stringify({ digests: message.digests }) + + if (nativeWriteFiles) { + // Native path: gunzip into a single buffer, hand the whole payload to + // Rust which parses and writes all files in parallel (rayon). + return new Promise<{ status: string, filesWritten: number }>((resolve, reject) => { + const req = requestFn(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Content-Length': Buffer.byteLength(body), + 'Accept-Encoding': 'gzip', + }, + }, (res: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any + let stream: NodeJS.ReadableStream = res + if (res.headers['content-encoding'] === 'gzip') { + const gunzip = createGunzip() + res.pipe(gunzip) + stream = gunzip + } + const chunks: Buffer[] = [] + stream.on('data', (chunk: Buffer) => chunks.push(chunk)) + stream.on('end', () => { + try { + const filesWritten = nativeWriteFiles!(message.storeDir, Buffer.concat(chunks)) + resolve({ status: 'success', filesWritten }) + } catch (err) { + reject(err as Error) + } + }) + stream.on('error', reject) + }) + req.on('error', reject) + req.write(body) + req.end() + }) + } + + // Pure-JS fallback: parse the binary stream incrementally and write each + // file with fs.writeFileSync as soon as its bytes arrive. + const { contentPathFromHex } = await import('@pnpm/store.cafs') const createdDirs = new Set() - // Stream: HTTP response → gunzip → parse entries → write to CAFS. - // No buffering — files are written as data arrives. return new Promise<{ status: string, filesWritten: number }>((resolve, reject) => { let filesWritten = 0 let buf = Buffer.alloc(0) @@ -521,7 +572,6 @@ async function fetchAndWriteCafs (message: FetchAndWriteCafsMessage): Promise<{ const END_MARKER = Buffer.alloc(64, 0) const processBuffer = () => { - // Skip JSON header on first chunk if (!headerSkipped && buf.length >= 4) { const jsonLen = buf.readUInt32BE(0) if (buf.length >= 4 + jsonLen) { @@ -532,7 +582,6 @@ async function fetchAndWriteCafs (message: FetchAndWriteCafsMessage): Promise<{ } } - // Parse complete file entries from the buffer while (headerSkipped) { if (buf.length < 64) break if (buf.subarray(0, 64).equals(END_MARKER)) {