FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

stream: fix pipeToSync byte accounting · nodejs/node@24b7831 · GitHub

/ node Public

Commit 24b7831

Browse files
trivikr
authored and
aduh95
committed
stream: fix pipeToSync byte accounting
Stop pipeToSync() from counting chunks when writeSync() or writevSync() returns false without accepting the data. Preserve writers that use false as an accepted backpressure signal. Fixes: #63562 Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com> Assisted-by: openai:gpt-5.5 PR-URL: #63564 Backport-PR-URL: #64675 Fixes: #63562 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 3b5f1b5 commit 24b7831

2 files changed

Lines changed: 78 additions & 16 deletions

File tree

‎lib/internal/streams/iter/pull.js‎

Lines changed: 39 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -863,6 +863,18 @@ async function* createAsyncPipeline(source, transforms, signal) {
863863
}
864864
}
865865

866+
/**
867+
* Check if a false sync write result means accepted backpressure.
868+
* @param {object} writer - The writer whose sync method returned.
869+
* @param {*} result - The return value from writeSync() or writevSync().
870+
* @returns {boolean}
871+
*/
872+
function isAcceptedSyncWriteBackpressure(writer, result) {
873+
return result === false &&
874+
writer[kSyncWriteAcceptedOnFalse] === true &&
875+
writer.desiredSize === 0;
876+
}
877+
866878
// =============================================================================
867879
// Public API: pull() and pullSync()
868880
// =============================================================================
@@ -950,16 +962,29 @@ function pipeToSync(source, ...args) {
950962
const hasEndSync = typeof writer.endSync === 'function';
951963

952964
try {
965+
let canContinue = true;
953966
for (const batch of pipeline) {
967+
if (!canContinue) {
968+
break;
969+
}
954970
if (hasWritevSync && batch.length > 1) {
955-
writer.writevSync(batch);
971+
const result = writer.writevSync(batch);
972+
if (result === false &&
973+
!isAcceptedSyncWriteBackpressure(writer, result)) {
974+
break;
975+
}
956976
for (let i = 0; i < batch.length; i++) {
957977
totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
958978
}
959979
} else {
960980
for (let i = 0; i < batch.length; i++) {
961981
const chunk = batch[i];
962-
writer.writeSync(chunk);
982+
const result = writer.writeSync(chunk);
983+
if (result === false &&
984+
!isAcceptedSyncWriteBackpressure(writer, result)) {
985+
canContinue = false;
986+
break;
987+
}
963988
totalBytes += TypedArrayPrototypeGetByteLength(chunk);
964989
}
965990
}
@@ -1011,11 +1036,6 @@ async function pipeTo(source, ...args) {
10111036
const hasWritev = typeof writer.writev === 'function';
10121037
const hasWritevSync = typeof writer.writevSync === 'function';
10131038
const hasEndSync = typeof writer.endSync === 'function';
1014-
const syncFalseCanBeAccepted = writer[kSyncWriteAcceptedOnFalse] === true;
1015-
1016-
function syncFalseWasAccepted() {
1017-
return syncFalseCanBeAccepted && writer.desiredSize === 0;
1018-
}
10191039

10201040
function waitForSyncBackpressure() {
10211041
const ondrain = writer[drainableProtocol];
@@ -1032,9 +1052,10 @@ async function pipeTo(source, ...args) {
10321052
async function writeBatchAsyncFallback(batch, startIndex) {
10331053
for (let i = startIndex; i < batch.length; i++) {
10341054
const chunk = batch[i];
1035-
if (hasWriteSync && writer.writeSync(chunk)) {
1055+
const result = hasWriteSync && writer.writeSync(chunk);
1056+
if (result) {
10361057
// Sync retry succeeded
1037-
} else if (syncFalseWasAccepted()) {
1058+
} else if (isAcceptedSyncWriteBackpressure(writer, result)) {
10381059
totalBytes += TypedArrayPrototypeGetByteLength(chunk);
10391060
await waitForSyncBackpressure();
10401061
continue;
@@ -1054,22 +1075,23 @@ async function pipeTo(source, ...args) {
10541075
// is required. Callers must check: const p = writeBatch(b); if (p) await p;
10551076
function writeBatch(batch) {
10561077
if (hasWritev && batch.length > 1) {
1057-
if (!hasWritevSync || !writer.writevSync(batch)) {
1058-
if (hasWritevSync && syncFalseWasAccepted()) {
1078+
const result = hasWritevSync && writer.writevSync(batch);
1079+
if (!result) {
1080+
if (isAcceptedSyncWriteBackpressure(writer, result)) {
10591081
for (let i = 0; i < batch.length; i++) {
10601082
totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
10611083
}
10621084
return waitForSyncBackpressure();
10631085
}
10641086
const opts = signal ? { __proto__: null, signal } : undefined;
1065-
const result = writer.writev(batch, opts);
1066-
if (result === undefined) {
1087+
const writevResult = writer.writev(batch, opts);
1088+
if (writevResult === undefined) {
10671089
for (let i = 0; i < batch.length; i++) {
10681090
totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
10691091
}
10701092
return;
10711093
}
1072-
return PromisePrototypeThen(PromiseResolve(result), () => {
1094+
return PromisePrototypeThen(PromiseResolve(writevResult), () => {
10731095
for (let i = 0; i < batch.length; i++) {
10741096
totalBytes += TypedArrayPrototypeGetByteLength(batch[i]);
10751097
}
@@ -1082,8 +1104,9 @@ async function pipeTo(source, ...args) {
10821104
}
10831105
for (let i = 0; i < batch.length; i++) {
10841106
const chunk = batch[i];
1085-
if (!hasWriteSync || !writer.writeSync(chunk)) {
1086-
if (hasWriteSync && syncFalseWasAccepted()) {
1107+
const result = hasWriteSync && writer.writeSync(chunk);
1108+
if (!result) {
1109+
if (isAcceptedSyncWriteBackpressure(writer, result)) {
10871110
totalBytes += TypedArrayPrototypeGetByteLength(chunk);
10881111
return writeBatchAfterAcceptedBackpressure(batch, i + 1);
10891112
}

‎test/parallel/test-stream-iter-pipeto-writev.js‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,43 @@ async function testPushWriterBlockSyncFalseAccepted() {
154154
})(), 'abcd', 4);
155155
}
156156

157+
async function testPipeToSyncPushWriterStrictFalseRejected() {
158+
const decoder = new TextDecoder();
159+
const { writer, readable } = push({ highWaterMark: 1 });
160+
161+
const total = pipeToSync(['a', 'b'], writer, { preventClose: true });
162+
assert.strictEqual(total, 1);
163+
164+
const iter = readable[Symbol.asyncIterator]();
165+
const first = await iter.next();
166+
assert.strictEqual(first.done, false);
167+
assert.strictEqual(decoder.decode(first.value[0]), 'a');
168+
169+
const second = await Promise.race([
170+
iter.next().then((result) => {
171+
return result.done ? '<done>' : decoder.decode(result.value[0]);
172+
}),
173+
setImmediatePromise().then(() => '<no second chunk>'),
174+
]);
175+
assert.strictEqual(second, '<no second chunk>');
176+
177+
await iter.return?.();
178+
}
179+
180+
async function testPipeToSyncWritevFalseNotCounted() {
181+
const writer = {
182+
writevSync() { return false; },
183+
writeSync: common.mustNotCall(),
184+
endSync() { return 0; },
185+
};
186+
function* source() {
187+
yield [new Uint8Array([1]), new Uint8Array([2])];
188+
}
189+
190+
const total = pipeToSync(source(), writer);
191+
assert.strictEqual(total, 0);
192+
}
193+
157194
// pipeToSync with writevSync
158195
async function testPipeToSyncWritev() {
159196
const batches = [];
@@ -215,6 +252,8 @@ Promise.all([
215252
testWriteSyncFailsMidBatch(),
216253
testWriteSyncAlwaysFails(),
217254
testPushWriterBlockSyncFalseAccepted(),
255+
testPipeToSyncPushWriterStrictFalseRejected(),
256+
testPipeToSyncWritevFalseNotCounted(),
218257
testPipeToSyncWritev(),
219258
testPipeToSyncPlainChunksWritev(),
220259
testPipeToSyncWriteFallback(),

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL