fs: write files in one thread pool round trip

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

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<void>}
+ */
+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';
 }

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<Value>& 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<Int32>()->Value(),
+                     args[2].As<Int32>()->Value(),
+                     args[3].As<ArrayBufferView>());
+  }
+
+  // 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<Value>& 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<Value> 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<unsigned int>(std::min<size_t>(
+                                     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<size_t>(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<WriteFileJob> 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<Value> 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> object,
+               std::string&& path,
+               int flags,
+               int mode,
+               Local<ArrayBufferView> 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<char*>(backing_store_->Data()) + view->ByteOffset(),
+             length_);
+      data_ = copy_.get();
+      backing_store_.reset();
+    } else {
+      buffer_.Reset(env->isolate(), view);
+      data_ = static_cast<char*>(backing_store_->Data()) + view->ByteOffset();
+    }
+    MakeWeak();
+  }
+
+  void Fail(const char* syscall, int error) {
+    syscall_ = syscall;
+    error_ = error;
+  }
+
+  const std::string path_;
+  v8::Global<v8::ArrayBufferView> buffer_;
+  std::shared_ptr<v8::BackingStore> backing_store_;
+  std::unique_ptr<char[]> 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<FunctionTemplate> 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<FunctionTemplate> 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);

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

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

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

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());