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 <shelley.vohr@gmail.com>

Author: codebytere

lib/fs.js

@@ -83,6 +83,7 @@ const {
 const {
   FSReqCallback,
   ReadFileJob,
+  WriteFileJob,
 } = binding;
 const { toPathIfFileURL } = require('internal/url');
 const {
@@ -2946,6 +2947,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

@@ -2113,6 +2113,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);
 
@@ -2123,6 +2131,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;
@@ -3214,6 +3215,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;
 
@@ -3297,6 +3299,162 @@ 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 {
+    tracker->TrackField("buffer", buffer_);
+    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)
@@ -4613,6 +4771,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(
@@ -4686,6 +4851,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));
   }