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'); }); }