From a1d02de789eb2ef676ac425829146519a4e5c70b 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. The job holds the buffer's backing store, so the memory stays valid if the buffer is detached or collected before the write finishes; a resizable ArrayBuffer could still have its pages decommitted by a shrink, so its contents are copied when the job is created. Signed-off-by: Shelley Vohr --- lib/fs.js | 18 ++ lib/internal/fs/promises.js | 35 ++++ src/node_file.cc | 166 ++++++++++++++++++ ...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 | 84 +++++++++ test/parallel/test-trace-events-fs-async.js | 9 +- 8 files changed, 318 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 cfed38d8127e..7049fd3c9e95 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 d571de820ced..166712f90c1d 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 871d8f16bd34..b53c7d27ae5f 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,161 @@ 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; } + void MemoryInfo(MemoryTracker* tracker) const override { + if (copy_) tracker->TrackFieldWithSize("copy", length_); + } + 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) { + // Holding the backing store keeps the memory valid even if the buffer is + // detached or collected meanwhile; a resizable buffer can still have its + // pages decommitted by a shrink, so its contents are copied instead. + length_ = view->ByteLength(); + backing_store_ = view->Buffer()->GetBackingStore(); + if (backing_store_->IsResizableByUserJavaScript()) { + copy_.reset(new char[length_]); + memcpy(copy_.get(), + static_cast(backing_store_->Data()) + view->ByteOffset(), + length_); + data_ = copy_.get(); + backing_store_.reset(); + } else { + buffer_.Reset(env->isolate(), view); + data_ = static_cast(backing_store_->Data()) + view->ByteOffset(); + } + MakeWeak(); + } + + void Fail(const char* syscall, int error) { + syscall_ = syscall; + error_ = error; + } + + const std::string path_; + v8::Global buffer_; + std::shared_ptr backing_store_; + std::unique_ptr copy_; + 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 +4710,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 +4790,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 36ebc23491ca..b299a7d5d572 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 901e71c15ac5..c368f3acb931 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 46b4acd0b8ff..d36056227841 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 000000000000..7f6c58ed93c8 --- /dev/null +++ b/test/parallel/test-fs-writefile-one-roundtrip.js @@ -0,0 +1,84 @@ +'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, write the data as it was at +// the call even if the buffer is resized or detached right after, 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 resizable = new ArrayBuffer(64 * 1024, { maxByteLength: 128 * 1024 }); + const view = new Uint8Array(resizable).fill(97); + const done = write(file, view); + resizable.resize(0); + await done; + assert.strictEqual(fs.readFileSync(file, 'utf8'), 'a'.repeat(64 * 1024)); + } + { + const file = next(); + const detached = new ArrayBuffer(1024); + const done = write(file, new Uint8Array(detached).fill(98)); + detached.transfer(8); + await done; + assert.strictEqual(fs.readFileSync(file, 'utf8'), 'b'.repeat(1024)); + } + { + 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 9b8e21b3d560..2a658838b39e 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'); }); }