mirror of
https://github.com/nodejs/node.git
synced 2026-10-28 10:15:07 +08:00
fs: serialize FileHandle writer() async writes
When writer() was created without a `start` offset, every write() and writev() call targeted the file descriptor's current position, and nothing prevented several of them from being in flight at once. Overlapping un-awaited writes then raced on the shared file offset (and partial writes were completed in follow-up syscalls), silently writing data at the wrong offsets while the reported byte count and the final file size still looked correct. Issue async writes one at a time, in call order. A write that is still queued when the writer fails, or whose signal aborts while it is queued, is no longer started. Assisted-by: OpenCode Signed-off-by: James M Snell <jasnell@gmail.com> PR-URL: https://github.com/nodejs/node/pull/66483 Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
This commit is contained in:
1 parent
daed8a3b7f
commit
cebc4337c4
2 files changed
+91
-3
No files matched your search
@@ -751,6 +751,11 @@ if (getOptionValue('--experimental-stream-iter')) {
|
||||
let asyncPending = 0;
|
||||
let released = false;
|
||||
const pendingWrites = new SafeSet();
|
||||
// Async writes are issued one at a time, in call order. Without this,
|
||||
// overlapping write() calls would race on the file position: when no
|
||||
// `start` is given every write targets the fd's current (shared)
|
||||
// position, and partial writes are completed in follow-up syscalls.
|
||||
let writeQueue = PromiseResolve();
|
||||
|
||||
validateBoolean(autoClose, 'options.autoClose');
|
||||
|
||||
@@ -866,6 +871,22 @@ if (getOptionValue('--experimental-stream-iter')) {
|
||||
});
|
||||
}
|
||||
|
||||
function ignoreWriteQueueResult() {}
|
||||
|
||||
// Run `start` once every previously queued write has settled.
|
||||
function enqueueWrite(start, signal) {
|
||||
const operation = PromisePrototypeThen(writeQueue, () => {
|
||||
// A failed writer must not touch the file again, and a write whose
|
||||
// signal aborted while it was queued must not be started.
|
||||
if (errored) throw error;
|
||||
signal?.throwIfAborted();
|
||||
return start();
|
||||
});
|
||||
writeQueue = PromisePrototypeThen(operation, ignoreWriteQueueResult,
|
||||
ignoreWriteQueueResult);
|
||||
return operation;
|
||||
}
|
||||
|
||||
function trackOperation(operation) {
|
||||
const { promise, resolve, reject } = PromiseWithResolvers();
|
||||
const pending = { __proto__: null, reject };
|
||||
@@ -950,8 +971,9 @@ if (getOptionValue('--experimental-stream-iter')) {
|
||||
if (bytesRemaining > 0) bytesRemaining -= chunk.byteLength;
|
||||
const position = pos;
|
||||
if (pos >= 0) pos += chunk.byteLength;
|
||||
return trackOperation(
|
||||
writeAll(chunk, 0, chunk.byteLength, position, signal));
|
||||
return trackOperation(enqueueWrite(
|
||||
() => writeAll(chunk, 0, chunk.byteLength, position, signal),
|
||||
signal));
|
||||
},
|
||||
|
||||
writev(chunks, options = kNullPrototo) {
|
||||
@@ -983,7 +1005,8 @@ if (getOptionValue('--experimental-stream-iter')) {
|
||||
if (bytesRemaining > 0) bytesRemaining -= totalSize;
|
||||
const position = pos;
|
||||
if (pos >= 0) pos += totalSize;
|
||||
return trackOperation(writevAll(chunks, position, signal));
|
||||
return trackOperation(enqueueWrite(
|
||||
() => writevAll(chunks, position, signal), signal));
|
||||
},
|
||||
|
||||
writeSync(chunk) {
|
||||
|
||||
@@ -1133,6 +1133,69 @@ async function testWriterWebIDLConversion() {
|
||||
// Run all tests
|
||||
// =============================================================================
|
||||
|
||||
// =============================================================================
|
||||
// Overlapping (un-awaited) writes must land in call order
|
||||
// =============================================================================
|
||||
|
||||
function makeTestData(size) {
|
||||
const data = Buffer.allocUnsafe(size);
|
||||
let seed = 0x9e3779b9;
|
||||
for (let i = 0; i < size; i++) {
|
||||
seed ^= seed << 13; seed ^= seed >>> 17; seed ^= seed << 5;
|
||||
data[i] = seed & 0xff;
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
async function testConcurrentWritesPreserveOrder() {
|
||||
const data = makeTestData(512 * 1024);
|
||||
for (const options of [{}, { start: 0 }]) {
|
||||
for (const useWritev of [false, true]) {
|
||||
const filePath = path.join(
|
||||
tmpDir,
|
||||
`writer-concurrent-${options.start ?? 'none'}-${useWritev}.bin`);
|
||||
const fh = await open(filePath, 'w');
|
||||
const w = fh.writer(options);
|
||||
const writes = [];
|
||||
let offset = 0;
|
||||
let i = 0;
|
||||
while (offset < data.length) {
|
||||
const size = 1 + ((i++ * 7919) % 9000);
|
||||
const end = Math.min(offset + size, data.length);
|
||||
const chunk = data.subarray(offset, end);
|
||||
writes.push(useWritev ?
|
||||
w.writev([chunk.subarray(0, 1), chunk.subarray(1)]) :
|
||||
w.write(chunk));
|
||||
offset = end;
|
||||
}
|
||||
await Promise.all(writes);
|
||||
assert.strictEqual(await w.end(), data.length);
|
||||
await fh.close();
|
||||
assert.deepStrictEqual(fs.readFileSync(filePath), data);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function testFailStopsQueuedWrites() {
|
||||
const filePath = path.join(tmpDir, 'writer-fail-queued.bin');
|
||||
const fh = await open(filePath, 'w');
|
||||
const w = fh.writer();
|
||||
const reason = new Error('stop');
|
||||
const writes = [];
|
||||
for (let i = 0; i < 10; i++) {
|
||||
writes.push(w.write(Buffer.alloc(1024, i)));
|
||||
}
|
||||
w.fail(reason);
|
||||
const results = await Promise.allSettled(writes);
|
||||
for (const result of results) {
|
||||
assert.strictEqual(result.status, 'rejected');
|
||||
assert.strictEqual(result.reason, reason);
|
||||
}
|
||||
await fh.close();
|
||||
// None of the queued writes may reach the file after fail().
|
||||
assert.strictEqual(fs.statSync(filePath).size, 0);
|
||||
}
|
||||
|
||||
Promise.all([
|
||||
testBasicWrite(),
|
||||
testBasicWritev(),
|
||||
@@ -1187,4 +1250,6 @@ Promise.all([
|
||||
testWriterLimitAndStart(),
|
||||
testWriterArgumentValidation(),
|
||||
testWriterWebIDLConversion(),
|
||||
testConcurrentWritesPreserveOrder(),
|
||||
testFailStopsQueuedWrites(),
|
||||
]).then(common.mustCall());
|
||||
Reference in new issue
Block a user