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

src: update Blob implementation to use DataQueue / File-backed Blobs · nodejs/node@36f36b9 · GitHub

/ node Public

Commit 36f36b9

Browse files
authored and
targos
committed
src: update Blob implementation to use DataQueue / File-backed Blobs
Co-authored-by: flakey5 <73616808+flakey5@users.noreply.github.com> PR-URL: #45258 Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
1 parent 9b6270a commit 36f36b9

11 files changed

Lines changed: 989 additions & 323 deletions

File tree

‎doc/api/fs.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,15 @@ When operating on file handles, the mode cannot be changed from what it was set
202202
to with [`fsPromises.open()`][]. Therefore, this is equivalent to
203203
[`filehandle.writeFile()`][].
204204

205+
#### `filehandle.blob()`
206+
<!-- YAML
207+
added: REPLACEME
208+
-->
209+
210+
> Stability: 1 - Experimental
211+
212+
Returns a {Blob} whose data is backed by this file.
213+
205214
#### `filehandle.chmod(mode)`
206215

207216
<!-- YAML

‎lib/internal/blob.js‎

Lines changed: 78 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,8 @@ const {
66
MathMin,
77
ObjectDefineProperties,
88
ObjectDefineProperty,
9-
PromiseResolve,
109
PromiseReject,
11-
SafePromisePrototypeFinally,
10+
PromiseResolve,
1211
ReflectConstruct,
1312
RegExpPrototypeExec,
1413
RegExpPrototypeSymbolReplace,
@@ -22,7 +21,8 @@ const {
2221

2322
const {
2423
createBlob: _createBlob,
25-
FixedSizeBlobCopyJob,
24+
createBlobFromFileHandle: _createBlobFromFileHandle,
25+
concat,
2626
getDataObject,
2727
} = internalBinding('blob');
2828

@@ -52,13 +52,13 @@ const {
5252
const { inspect } = require('internal/util/inspect');
5353

5454
const {
55-
AbortError,
5655
codes: {
5756
ERR_INVALID_ARG_TYPE,
5857
ERR_INVALID_ARG_VALUE,
5958
ERR_INVALID_THIS,
6059
ERR_BUFFER_TOO_LARGE,
61-
}
60+
},
61+
errnoException,
6262
} = require('internal/errors');
6363

6464
const {
@@ -67,13 +67,8 @@ const {
6767
} = require('internal/validators');
6868

6969
const kHandle = Symbol('kHandle');
70-
const kState = Symbol('kState');
71-
const kIndex = Symbol('kIndex');
7270
const kType = Symbol('kType');
7371
const kLength = Symbol('kLength');
74-
const kArrayBufferPromise = Symbol('kArrayBufferPromise');
75-
76-
const kMaxChunkSize = 65536;
7772

7873
const disallowedTypeCharacters = /[^\u{0020}-\u{007E}]/u;
7974

@@ -266,40 +261,28 @@ class Blob {
266261
if (!isBlob(this))
267262
return PromiseReject(new ERR_INVALID_THIS('Blob'));
268263

269-
// If there's already a promise in flight for the content,
270-
// reuse it, but only while it's in flight. After the cached
271-
// promise resolves it will be cleared, allowing it to be
272-
// garbage collected as soon as possible.
273-
if (this[kArrayBufferPromise])
274-
return this[kArrayBufferPromise];
275-
276-
const job = new FixedSizeBlobCopyJob(this[kHandle]);
277-
278-
const ret = job.run();
279-
280-
// If the job returns a value immediately, the ArrayBuffer
281-
// was generated synchronously and should just be returned
282-
// directly.
283-
if (ret !== undefined)
284-
return PromiseResolve(ret);
264+
if (this.size === 0) {
265+
return PromiseResolve(new ArrayBuffer(0));
266+
}
285267

286-
const {
287-
promise,
288-
resolve,
289-
reject,
290-
} = createDeferredPromise();
291-
292-
job.ondone = (err, ab) => {
293-
if (err !== undefined)
294-
return reject(new AbortError(undefined, { cause: err }));
295-
resolve(ab);
268+
const { promise, resolve } = createDeferredPromise();
269+
const reader = this[kHandle].getReader();
270+
const buffers = [];
271+
const readNext = () => {
272+
reader.pull((status, buffer) => {
273+
if (status === -1) {
274+
// EOS, concat & resolve
275+
// buffer should be undefined here
276+
resolve(concat(buffers));
277+
return;
278+
}
279+
if (buffer !== undefined)
280+
buffers.push(buffer);
281+
readNext();
282+
});
296283
};
297-
this[kArrayBufferPromise] =
298-
SafePromisePrototypeFinally(
299-
promise,
300-
() => this[kArrayBufferPromise] = undefined);
301-
302-
return this[kArrayBufferPromise];
284+
readNext();
285+
return promise;
303286
}
304287

305288
/**
@@ -321,24 +304,57 @@ class Blob {
321304
if (!isBlob(this))
322305
throw new ERR_INVALID_THIS('Blob');
323306

324-
const self = this;
307+
if (this.size === 0) {
308+
return new lazyReadableStream({
309+
start(c) { c.close(); }
310+
});
311+
}
312+
313+
const reader = this[kHandle].getReader();
325314
return new lazyReadableStream({
326-
async start() {
327-
this[kState] = await self.arrayBuffer();
328-
this[kIndex] = 0;
315+
start(c) {
316+
// There really should only be one read at a time so using an
317+
// array here is purely defensive.
318+
this.pendingPulls = [];
329319
},
330-
331-
pull(controller) {
332-
if (this[kState].byteLength - this[kIndex] <= kMaxChunkSize) {
333-
controller.enqueue(new Uint8Array(this[kState], this[kIndex]));
334-
controller.close();
335-
this[kState] = undefined;
336-
} else {
337-
controller.enqueue(new Uint8Array(this[kState], this[kIndex], kMaxChunkSize));
338-
this[kIndex] += kMaxChunkSize;
320+
pull(c) {
321+
const { promise, resolve, reject } = createDeferredPromise();
322+
this.pendingPulls.push({resolve, reject});
323+
reader.pull((status, buffer) => {
324+
// If pendingPulls is empty here, the stream had to have
325+
// been canceled, and we don't really care about the result.
326+
// we can simply exit.
327+
if (this.pendingPulls.length === 0) {
328+
return;
329+
}
330+
const pending = this.pendingPulls.shift();
331+
if (status === -1 || (status === 0 && buffer === undefined)) {
332+
// EOS
333+
c.close();
334+
pending.resolve();
335+
return;
336+
} else if (status < 0) {
337+
const error = errnoException(status, 'read');
338+
c.error(error);
339+
pending.reject(error);
340+
return;
341+
}
342+
c.enqueue(new Uint8Array(buffer));
343+
pending.resolve();
344+
});
345+
return promise;
346+
},
347+
cancel(reason) {
348+
// Reject any currently pending pulls here.
349+
for (const pending of this.pendingPulls) {
350+
pending.reject(reason);
339351
}
352+
this.pendingPulls = [];
340353
}
341-
});
354+
// We set the highWaterMark to 0 because we do not want the stream to
355+
// start reading immediately on creation. We want it to wait until read
356+
// is called.
357+
}, new CountQueuingStrategy({ highWaterMark: 0 }));
342358
}
343359
}
344360

@@ -406,10 +422,16 @@ function resolveObjectURL(url) {
406422
}
407423
}
408424

425+
function createBlobFromFileHandle(handle) {
426+
const [blob, length] = _createBlobFromFileHandle(handle);
427+
return createBlob(blob, length);
428+
}
429+
409430
module.exports = {
410431
Blob,
411432
ClonedBlob,
412433
createBlob,
434+
createBlobFromFileHandle,
413435
isBlob,
414436
kHandle,
415437
resolveObjectURL,

‎lib/internal/fs/promises.js‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ const {
2525
S_IFREG
2626
} = constants;
2727

28+
const { createBlobFromFileHandle } = require('internal/blob');
29+
2830
const binding = internalBinding('fs');
2931
const { Buffer } = require('buffer');
3032

@@ -310,6 +312,14 @@ class FileHandle extends EventEmitterMixin(JSTransferable) {
310312
return new WriteStream(undefined, { ...options, fd: this });
311313
}
312314

315+
/**
316+
* @typedef {import('../blob').Blob} Blob
317+
* @returns {Blob}
318+
*/
319+
blob() {
320+
return createBlobFromFileHandle(this[kHandle]);
321+
}
322+
313323
[kTransfer]() {
314324
if (this[kClosePromise] || this[kRefs] > 1) {
315325
throw lazyDOMException('Cannot transfer FileHandle while in use',

‎src/async_wrap.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ namespace node {
3838
V(ELDHISTOGRAM) \
3939
V(FILEHANDLE) \
4040
V(FILEHANDLECLOSEREQ) \
41-
V(FIXEDSIZEBLOBCOPY) \
41+
V(BLOBREADER) \
4242
V(FSEVENTWRAP) \
4343
V(FSREQCALLBACK) \
4444
V(FSREQPROMISE) \

‎src/env_properties.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -330,6 +330,7 @@
330330
V(base_object_ctor_template, v8::FunctionTemplate) \
331331
V(binding_data_ctor_template, v8::FunctionTemplate) \
332332
V(blob_constructor_template, v8::FunctionTemplate) \
333+
V(blob_reader_constructor_template, v8::FunctionTemplate) \
333334
V(blocklist_constructor_template, v8::FunctionTemplate) \
334335
V(contextify_global_template, v8::ObjectTemplate) \
335336
V(contextify_wrapper_template, v8::ObjectTemplate) \
@@ -340,6 +341,7 @@
340341
V(dir_instance_template, v8::ObjectTemplate) \
341342
V(fd_constructor_template, v8::ObjectTemplate) \
342343
V(fdclose_constructor_template, v8::ObjectTemplate) \
344+
V(fdentry_constructor_template, v8::FunctionTemplate) \
343345
V(filehandlereadwrap_template, v8::ObjectTemplate) \
344346
V(fsreqpromise_constructor_template, v8::ObjectTemplate) \
345347
V(handle_wrap_ctor_template, v8::FunctionTemplate) \
@@ -359,14 +361,18 @@
359361
V(secure_context_constructor_template, v8::FunctionTemplate) \
360362
V(shutdown_wrap_template, v8::ObjectTemplate) \
361363
V(socketaddress_constructor_template, v8::FunctionTemplate) \
364+
V(streambaseentry_ctor_template, v8::FunctionTemplate) \
362365
V(streambaseoutputstream_constructor_template, v8::ObjectTemplate) \
366+
V(streamentry_ctor_template, v8::FunctionTemplate) \
367+
V(streamentry_opaque_ctor_template, v8::FunctionTemplate) \
363368
V(qlogoutputstream_constructor_template, v8::ObjectTemplate) \
364369
V(tcp_constructor_template, v8::FunctionTemplate) \
365370
V(tty_constructor_template, v8::FunctionTemplate) \
366371
V(write_wrap_template, v8::ObjectTemplate) \
367372
V(worker_heap_snapshot_taker_template, v8::ObjectTemplate) \
368373
V(x509_constructor_template, v8::FunctionTemplate)
369374

375+
370376
#define PER_REALM_STRONG_PERSISTENT_VALUES(V) \
371377
V(async_hooks_after_function, v8::Function) \
372378
V(async_hooks_before_function, v8::Function) \

0 commit comments

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