From 4c3fcd6de7b19cc7f31fb6efdb0eed5c9f3bab6c Mon Sep 17 00:00:00 2001 From: Shelley Vohr Date: Sat, 22 Aug 2026 13:42:06 +0000 Subject: [PATCH] fs: write files in one thread pool round trip fs.writeFile(path, data) took three libuv thread pool round trips (open, write, close), each its own request with its own queue wait, completion callback and JS/C++ crossing, and fs.promises.writeFile() did the same through a FileHandle. For the small files applications write most, the round trips are the cost, and each occupies a pool slot that concurrent fs, dns.lookup() and crypto work is also queueing for. Add WriteFileJob next to ReadFileJob: an AsyncWrap + ThreadPoolWork that opens, writes the whole buffer (looping on short writes) and closes as one pool task, keeping the buffer alive until it is done. fs.writeFile() uses it for path arguments without flush; fs.promises.writeFile() additionally keeps data above one write chunk (and iterables) on the FileHandle path, so large writes stay abortable between chunks as before. File descriptors, FileHandles, flush: true and an active VFS keep their existing paths. Behavior is otherwise kept: open failures report syscall 'open' with the path, write failures 'write'; permission errors are delivered through the callback/promise; an abort signalled while the write is in flight is still reported as an AbortError; the job is an FSREQCALLBACK resource for async_hooks and emits the 'write' fs trace event. Tests that used fs.writeFile() as a proxy for open/close trace events, or injected FileHandle faults for path-based writes, are adjusted to keep testing what they test. Signed-off-by: Shelley Vohr --- lib/fs.js | 18 +++ lib/internal/fs/promises.js | 35 ++++ src/node_file.cc | 152 ++++++++++++++++++ ...s-promises-file-handle-aggregate-errors.js | 4 +- ...st-fs-promises-file-handle-close-errors.js | 4 +- .../test-fs-promises-file-handle-op-errors.js | 4 +- .../test-fs-writefile-one-roundtrip.js | 66 ++++++++ test/parallel/test-trace-events-fs-async.js | 9 +- 8 files changed, 286 insertions(+), 6 deletions(-) create mode 100644 test/parallel/test-fs-writefile-one-roundtrip.js diff --git a/lib/fs.js b/lib/fs.js index cfed38d8127..7049fd3c9e9 100644 --- a/lib/fs.js +++ b/lib/fs.js @@ -82,6 +82,7 @@ const { const { FSReqCallback, ReadFileJob, + WriteFileJob, } = binding; const { toPathIfFileURL } = require('internal/url'); const { @@ -2924,6 +2925,23 @@ function writeFile(path, data, options, callback) { if (checkAborted(options.signal, callback)) return; + if (!flush) { + // Open + write + close in one thread pool round trip. + const signal = options.signal; + path = getValidatedPath(path); + const job = new WriteFileJob(path, stringToFlags(flag, 'options.flag'), + parseFileMode(options.mode, 'mode', 0o666), data); + job.ondone = signal == null ? callback : (err) => { + // An abort that arrived while the write was in flight still wins. + callback(signal.aborted && !err ? new AbortError(undefined, { cause: signal.reason }) : err); + }; + const accessError = job.run(path); + if (accessError !== undefined) { + callback(accessError); + } + return; + } + fs.open(path, flag, options.mode, (openErr, fd) => { if (openErr) { callback(openErr); diff --git a/lib/internal/fs/promises.js b/lib/internal/fs/promises.js index d571de820ce..166712f90c1 100644 --- a/lib/internal/fs/promises.js +++ b/lib/internal/fs/promises.js @@ -2107,6 +2107,14 @@ async function writeFile(path, data, options) { checkAborted(options.signal); + if (!flush && !isCustomIterable(data) && data.byteLength <= kWriteFileMaxChunkSize) { + path = getValidatedPath(path); + await writeFileInOneRoundTrip(path, stringToFlags(flag, 'options.flag'), + parseFileMode(options.mode, 'mode', 0o666), data); + checkAborted(options.signal); // An abort during the write still wins. + return; + } + const fd = await open(path, flag, options.mode); let writeOp = writeFileHandle(fd, data, options.signal, options.encoding); @@ -2117,6 +2125,33 @@ async function writeFile(path, data, options) { return handleFdClose(writeOp, fd.close); } +/** + * Open + write + close as one thread pool round trip. + * @param {string|Buffer} path Validated path + * @param {number} flagsNumber + * @param {number} mode + * @param {ArrayBufferView} data + * @returns {Promise} + */ +function writeFileInOneRoundTrip(path, flagsNumber, mode, data) { + return new Promise((resolve, reject) => { + const job = new binding.WriteFileJob(path, flagsNumber, mode, data); + job.ondone = (err) => { + if (err != null) { + ErrorCaptureStackTrace(err, writeFileInOneRoundTrip); + reject(err); + } else { + resolve(); + } + }; + const accessError = job.run(path); + if (accessError !== undefined) { + ErrorCaptureStackTrace(accessError, writeFileInOneRoundTrip); + reject(accessError); + } + }); +} + function isCustomIterable(obj) { return isIterable(obj) && !isArrayBufferView(obj) && typeof obj !== 'string'; } diff --git a/src/node_file.cc b/src/node_file.cc index 871d8f16bd3..63f50aa3b5e 100644 --- a/src/node_file.cc +++ b/src/node_file.cc @@ -64,6 +64,7 @@ namespace node { namespace fs { using v8::Array; +using v8::ArrayBufferView; using v8::BigInt; using v8::Context; using v8::EscapableHandleScope; @@ -3154,6 +3155,7 @@ class ReadFileJob final : public AsyncWrap, public ThreadPoolWork { SET_SELF_SIZE(ReadFileJob) private: + friend class WriteFileJob; static constexpr size_t kUnknownSizeChunk = 64 * 1024; static constexpr size_t kMaxReadChunk = 256 * 1024 * 1024; @@ -3237,6 +3239,147 @@ class ReadFileJob final : public AsyncWrap, public ThreadPoolWork { int fd_ = -1; }; +// Writes a whole buffer to a file in ONE thread pool round trip -- open + +// write (until everything is written) + close -- for fs.writeFile() and +// fs.promises.writeFile() with a path, which otherwise pay one round trip per +// step. +// +// JS: const job = new WriteFileJob(path, flags, mode, buffer); +// job.ondone = (err) => {...}; job.run(path); +// `err` carries the syscall that failed ('open', 'write' or 'close'); the file +// descriptor opened here is always closed. +class WriteFileJob final : public AsyncWrap, public ThreadPoolWork { + public: + static void New(const FunctionCallbackInfo& args) { + CHECK(args.IsConstructCall()); + Environment* env = Environment::GetCurrent(args); + CHECK_GE(args.Length(), 4); + BufferValue path(env->isolate(), args[0]); + CHECK_NOT_NULL(*path); + ToNamespacedPath(env, &path); + CHECK(args[1]->IsInt32()); + CHECK(args[2]->IsInt32()); + CHECK(args[3]->IsArrayBufferView()); + new WriteFileJob(env, + args.This(), + path.ToString(), + args[1].As()->Value(), + args[2].As()->Value(), + args[3].As()); + } + + // Returns undefined when the job was scheduled, or the ERR_ACCESS_DENIED + // error the asynchronous open() would have delivered (nothing is scheduled). + static void Run(const FunctionCallbackInfo& args) { + WriteFileJob* job; + ASSIGN_OR_RETURN_UNWRAP(&job, args.This()); + Environment* env = job->AsyncWrap::env(); + CHECK(!job->scheduled_); + BufferValue path(env->isolate(), args[0]); + CHECK_NOT_NULL(*path); + ToNamespacedPath(env, &path); + Local access_error; + if (ReadFileJob::OpenPermissionError(env, path, job->flags_) + .ToLocal(&access_error)) { + args.GetReturnValue().Set(access_error); + return; + } + job->scheduled_ = true; + job->ClearWeak(); + FS_ASYNC_TRACE_BEGIN0(UV_FS_WRITE, job) + job->ScheduleWork(); + } + + void DoThreadPoolWork() override { + uv_fs_t req; + int fd = uv_fs_open(nullptr, &req, path_.c_str(), flags_, mode_, nullptr); + uv_fs_req_cleanup(&req); + if (fd < 0) return Fail("open", fd); + + size_t written = 0; + while (written < length_) { + uv_buf_t buf = uv_buf_init(data_ + written, + static_cast(std::min( + length_ - written, kMaxWriteChunk))); + int r = uv_fs_write(nullptr, &req, fd, &buf, 1, -1, nullptr); + uv_fs_req_cleanup(&req); + if (r < 0) { + Fail("write", r); + break; + } + written += static_cast(r); + } + + int rc = uv_fs_close(nullptr, &req, fd, nullptr); + uv_fs_req_cleanup(&req); + if (rc < 0 && error_ == 0) Fail("close", rc); + } + + void AfterThreadPoolWork(int status) override { + Environment* env = AsyncWrap::env(); + std::unique_ptr self(this); + CHECK(status == 0 || status == UV_ECANCELED); + FS_ASYNC_TRACE_END0(UV_FS_WRITE, this) + if (status == UV_ECANCELED || !env->can_call_into_js()) return; + HandleScope handle_scope(env->isolate()); + Context::Scope context_scope(env->context()); + Isolate* isolate = env->isolate(); + Local argv[1] = {Null(isolate)}; + if (error_ != 0) { + argv[0] = UVException(isolate, + error_, + syscall_, + nullptr, + syscall_ == kOpen ? path_.c_str() : nullptr); + } + MakeCallback(env->ondone_string(), arraysize(argv), argv); + } + + bool IsNotIndicativeOfMemoryLeakAtExit() const override { return true; } + SET_NO_MEMORY_INFO() + SET_MEMORY_INFO_NAME(WriteFileJob) + SET_SELF_SIZE(WriteFileJob) + + private: + static constexpr size_t kMaxWriteChunk = 256 * 1024 * 1024; + static constexpr const char* kOpen = "open"; + + WriteFileJob(Environment* env, + Local object, + std::string&& path, + int flags, + int mode, + Local view) + : AsyncWrap(env, object, AsyncWrap::PROVIDER_FSREQCALLBACK), + ThreadPoolWork(env, "fs.writefile"), + path_(std::move(path)), + flags_(flags), + mode_(mode) { + // Keep the buffer alive (and its backing store in place) until done. + buffer_.Reset(env->isolate(), view); + backing_store_ = view->Buffer()->GetBackingStore(); + data_ = static_cast(backing_store_->Data()) + view->ByteOffset(); + length_ = view->ByteLength(); + MakeWeak(); + } + + void Fail(const char* syscall, int error) { + syscall_ = syscall; + error_ = error; + } + + const std::string path_; + v8::Global buffer_; + std::shared_ptr backing_store_; + char* data_ = nullptr; + size_t length_ = 0; + const int flags_; + const int mode_; + bool scheduled_ = false; + int error_ = 0; + const char* syscall_ = nullptr; +}; + // Wrapper for readv(2). // // bytesRead = fs.readv(fd, buffers[, position], callback) @@ -4553,6 +4696,13 @@ static void CreatePerIsolateProperties(IsolateData* isolate_data, SetProtoMethod(isolate, rfj, "run", ReadFileJob::Run); SetConstructorFunction(isolate, target, "ReadFileJob", rfj); + Local wfj = NewFunctionTemplate(isolate, WriteFileJob::New); + wfj->InstanceTemplate()->SetInternalFieldCount( + WriteFileJob::kInternalFieldCount); + wfj->Inherit(AsyncWrap::GetConstructorTemplate(isolate_data)); + SetProtoMethod(isolate, wfj, "run", WriteFileJob::Run); + SetConstructorFunction(isolate, target, "WriteFileJob", wfj); + // Create FunctionTemplate for FSReqCallback Local fst = NewFunctionTemplate(isolate, NewFSReqCallback); fst->InstanceTemplate()->SetInternalFieldCount( @@ -4626,6 +4776,8 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) { registry->Register(Open); registry->Register(ReadFileJob::New); registry->Register(ReadFileJob::Run); + registry->Register(WriteFileJob::New); + registry->Register(WriteFileJob::Run); registry->Register(OpenFileHandle); registry->Register(Read); registry->Register(ReadFileUtf8); diff --git a/test/parallel/test-fs-promises-file-handle-aggregate-errors.js b/test/parallel/test-fs-promises-file-handle-aggregate-errors.js index 36ebc23491c..b299a7d5d57 100644 --- a/test/parallel/test-fs-promises-file-handle-aggregate-errors.js +++ b/test/parallel/test-fs-promises-file-handle-aggregate-errors.js @@ -67,7 +67,9 @@ async function checkAggregateError(op) { tmpdir.refresh(); await checkAggregateError((filePath) => truncate(filePath)); await checkAggregateError((filePath) => readFile(filePath)); - await checkAggregateError((filePath) => writeFile(filePath, '123')); + // More than one write chunk (512 KiB), so that writeFile(path) goes through + // a FileHandle as well. + await checkAggregateError((filePath) => writeFile(filePath, '123'.repeat(200_000))); if (common.isMacOS) { await checkAggregateError((filePath) => lchmod(filePath, 0o777)); } diff --git a/test/parallel/test-fs-promises-file-handle-close-errors.js b/test/parallel/test-fs-promises-file-handle-close-errors.js index 901e71c15ac..c368f3acb93 100644 --- a/test/parallel/test-fs-promises-file-handle-close-errors.js +++ b/test/parallel/test-fs-promises-file-handle-close-errors.js @@ -62,7 +62,9 @@ async function checkCloseError(op) { tmpdir.refresh(); await checkCloseError((filePath) => truncate(filePath)); await checkCloseError((filePath) => readFile(filePath)); - await checkCloseError((filePath) => writeFile(filePath, '123')); + // More than one write chunk (512 KiB), so that writeFile(path) goes through + // a FileHandle as well. + await checkCloseError((filePath) => writeFile(filePath, '123'.repeat(200_000))); if (common.isMacOS) { await checkCloseError((filePath) => lchmod(filePath, 0o777)); } diff --git a/test/parallel/test-fs-promises-file-handle-op-errors.js b/test/parallel/test-fs-promises-file-handle-op-errors.js index 46b4acd0b8f..d3605622784 100644 --- a/test/parallel/test-fs-promises-file-handle-op-errors.js +++ b/test/parallel/test-fs-promises-file-handle-op-errors.js @@ -56,7 +56,9 @@ async function checkOperationError(op) { tmpdir.refresh(); await checkOperationError((filePath) => truncate(filePath)); await checkOperationError((filePath) => readFile(filePath)); - await checkOperationError((filePath) => writeFile(filePath, '123')); + // More than one write chunk (512 KiB), so that writeFile(path) goes through + // a FileHandle as well. + await checkOperationError((filePath) => writeFile(filePath, '123'.repeat(200_000))); if (common.isMacOS) { await checkOperationError((filePath) => lchmod(filePath, 0o777)); } diff --git a/test/parallel/test-fs-writefile-one-roundtrip.js b/test/parallel/test-fs-writefile-one-roundtrip.js new file mode 100644 index 00000000000..6b5c5d14bbb --- /dev/null +++ b/test/parallel/test-fs-writefile-one-roundtrip.js @@ -0,0 +1,66 @@ +'use strict'; +// fs.writeFile() and fs.promises.writeFile() with a path perform +// open + write + close as one thread pool request. This covers what that +// request must keep doing: honor flags and mode, append, report the +// failing syscall, accept every ArrayBufferView, and write buffers larger +// than one write() call in full. +const common = require('../common'); +const tmpdir = require('../common/tmpdir'); +const assert = require('assert'); +const fs = require('fs'); +const path = require('path'); + +tmpdir.refresh(); +let counter = 0; +const next = () => tmpdir.resolve(`file-${counter++}`); + +async function check(write) { + { + const file = next(); + await write(file, 'hello'); + await write(file, ' world', { flag: 'a' }); + assert.strictEqual(fs.readFileSync(file, 'utf8'), 'hello world'); + await assert.rejects(write(file, 'again', { flag: 'wx' }), { code: 'EEXIST', syscall: 'open', path: file }); + } + { + const file = path.join(next(), 'missing-dir', 'file'); + await assert.rejects(write(file, 'x'), { code: 'ENOENT', syscall: 'open', path: file }); + } + { + const file = next(); + await write(file, ''); + assert.strictEqual(fs.statSync(file).size, 0); + } + if (!common.isWindows) { + const file = next(); + const mask = process.umask(0o022); + await write(file, 'x', { mode: 0o640 }); + process.umask(mask); + assert.strictEqual(fs.statSync(file).mode & 0o777, 0o640); + } + { + const file = next(); + const units = new Uint16Array([0x6968]); + await write(file, units); + await write(file, new DataView(new TextEncoder().encode('!?').buffer, 1, 1), { flag: 'a' }); + assert.deepStrictEqual(fs.readFileSync(file), + Buffer.concat([Buffer.from(units.buffer), Buffer.from('?')])); + } + if (fs.existsSync('/dev/full')) { + // open() succeeds, write() fails. + await assert.rejects(write('/dev/full', 'x'), { code: 'ENOSPC', syscall: 'write' }); + } + { + const file = next(); + const big = Buffer.alloc(3 * 1024 * 1024 + 7, 'z'); + await write(file, big); + assert.deepStrictEqual(fs.readFileSync(file), big); + } +} + +(async () => { + await check((file, data, options) => new Promise((resolve, reject) => { + fs.writeFile(file, data, options, (err) => (err ? reject(err) : resolve())); + })); + await check(fs.promises.writeFile); +})().then(common.mustCall()); diff --git a/test/parallel/test-trace-events-fs-async.js b/test/parallel/test-trace-events-fs-async.js index 9b8e21b3d56..2a658838b39 100644 --- a/test/parallel/test-trace-events-fs-async.js +++ b/test/parallel/test-trace-events-fs-async.js @@ -47,8 +47,10 @@ function chown({ uid, gid }) { function close() { const fs = require('fs'); - fs.writeFile('fs3.txt', '123', 'utf8', () => { - fs.unlinkSync('fs3.txt'); + fs.open('fs3.txt', 'w', (err, fd) => { + fs.close(fd, () => { + fs.unlinkSync('fs3.txt'); + }); }); } @@ -173,7 +175,8 @@ function mktmp() { function open() { const fs = require('fs'); - fs.writeFile('fs16.txt', '123', 'utf8', () => { + fs.open('fs16.txt', 'w', (err, fd) => { + fs.closeSync(fd); fs.unlinkSync('fs16.txt'); }); }