-
Notifications
You must be signed in to change notification settings - Fork 29.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
lib: performance improvement on readline async iterator
Using a direct approach to create the readline async iterator allowed an iteration over 20 to 58% faster. **BREAKING CHANGE**: With that change, the async iteterator obtained from the readline interface doesn't have the property "stream" any longer. This happened because it's no longer created through a Readable, instead, the async iterator is created directly from the events of the readline interface instance, so, if anyone is using that property, this change will break their code. Also, the Readable added a backpressure control that is fairly compensated by the use of FixedQueue + monitoring its size. This control wasn't really precise with readline before, though, because it only pauses the reading of the original stream, but the lines generated from the last message received from it was still emitted. For example: if the readable was paused at 1000 messages but the last one received generated 10k lines, but no further messages were emitted again until the queue was lower than the readable highWaterMark. A similar behavior still happens with the new implementation, but the highWaterMark used is fixed: 1024, and the original stream is resumed again only after the queue is cleared. Before making that change, I created a package implementing the same concept used here to validate it. You can find it [here](https://github.com/Farenheith/faster-readline-iterator) if this helps anyhow.
- Loading branch information
1 parent
cf69964
commit c163dc1
Showing
6 changed files
with
387 additions
and
135 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,110 @@ | ||
| const { | ||
| Map, | ||
| SymbolIterator, | ||
| } = primordials; | ||
|
|
||
| function push(root, value) { | ||
| const node = { value }; | ||
| if (root.last) { | ||
| node.previous = root.last; | ||
| root.last.next = root.last = node; | ||
| } else { | ||
| root.last = root.first = node; | ||
| } | ||
| root.length++; | ||
|
|
||
| return node; | ||
| } | ||
|
|
||
| function unshift(value) { | ||
| const node = { value }; | ||
| if (root.first) { | ||
| node.next = root.first; | ||
| root.first.previous = root.first = node; | ||
| } else { | ||
| root.last = root.first = node; | ||
| } | ||
| root.length++; | ||
|
|
||
| return node; | ||
| } | ||
|
|
||
| function shift(root) { | ||
| if (root.first) { | ||
| const { value } = root.first; | ||
| root.first = root.first.next; | ||
| root.length--; | ||
| return value; | ||
| } | ||
| } | ||
|
|
||
|
|
||
| function getLinkedMap() { | ||
| const map = new Map(); | ||
| function addToMap(key, node, operation) { | ||
| let refs = map.get(key); | ||
| if (!refs) { | ||
| map.set(key, refs = { length: 0 }); | ||
| } | ||
| operation(refs, node); | ||
| } | ||
| const root = { length: 0 }; | ||
|
|
||
| return { | ||
| get length() { | ||
| return root.length; | ||
| }, | ||
| push(value) { | ||
| const node = push(root, value); | ||
| addToMap(value, node, push); | ||
|
|
||
| return this; | ||
| }, | ||
| unshift(value) { | ||
| const node = unshift(root, value); | ||
| addToMap(key, node, unshift); | ||
|
|
||
| return this; | ||
| }, | ||
| remove(value) { | ||
| const refs = map.get(value); | ||
| if (refs) { | ||
| const result = shift(refs); | ||
| if (result.previous) | ||
| result.previous.next = result.next; | ||
| if (result.next) | ||
| result.next.previous = result.previous; | ||
| if (refs.length === 0) { | ||
| map.delete(value); | ||
| } | ||
| root.length--; | ||
| return 1; | ||
| } | ||
| return 0; | ||
| }, | ||
| [SymbolIterator]() { | ||
| let node = root.first; | ||
|
|
||
| const iterator = { | ||
| next() { | ||
| if (!node) { | ||
| return { done: true }; | ||
| } | ||
| const result = { | ||
| done: false, | ||
| value: node.value, | ||
| }; | ||
| node = node.next; | ||
| return result; | ||
| }, | ||
| [SymbolIterator]() { | ||
| return iterator; | ||
| } | ||
| }; | ||
|
|
||
| return iterator; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| module.exports = getLinkedMap; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| 'use strict'; | ||
| const { | ||
| Promise, | ||
| SymbolAsyncIterator, | ||
| ArrayPrototypeConcat, | ||
| } = primordials; | ||
| const FixedQueue = require('internal/fixed_queue'); | ||
|
|
||
| const PAUSE_THRESHOLD = 1024; | ||
| const RESUME_THRESHOLD = 1; | ||
| const ITEM_EVENTS = ['data']; | ||
| const CLOSE_EVENTS = ['close', 'end']; | ||
| const ERROR_EVENTS = ['error']; | ||
|
|
||
|
|
||
| function waitNext(emitter, next, events) { | ||
| return new Promise((resolve, reject) => { | ||
| const resolveNext = () => { | ||
| for (let i = 0; i < events.length; i++) | ||
| emitter.off(events[i], resolveNext); | ||
| try { | ||
| resolve(next()); | ||
| } catch (promiseError) { | ||
| reject(promiseError); | ||
| } | ||
| }; | ||
| for (let i = 0; i < events.length; i++) | ||
| emitter.once(events[i], resolveNext); | ||
| }); | ||
| } | ||
|
|
||
| module.exports = function eventsToAsyncIteratorFactory(readable, { | ||
| pauseThreshold = PAUSE_THRESHOLD, | ||
| resumeThreshold = RESUME_THRESHOLD, | ||
| closeEvents = CLOSE_EVENTS, | ||
| itemEvents = ITEM_EVENTS, | ||
| errorEvents = ERROR_EVENTS, | ||
| }) { | ||
| const events = ArrayPrototypeConcat(itemEvents, errorEvents, closeEvents); | ||
| const highWaterMark = RESUME_THRESHOLD; | ||
|
|
||
| const queue = new FixedQueue(); | ||
| let done = false; | ||
| let error; | ||
| let queueSize = 0; | ||
| let paused = false; | ||
| const onError = (value) => { | ||
| turn('off'); | ||
| error = value; | ||
| }; | ||
| const onClose = () => { | ||
| turn('off'); | ||
| done = true; | ||
| }; | ||
| const onItem = (value) => { | ||
| queue.push(value); | ||
| queueSize++; | ||
| if (queueSize >= pauseThreshold) { | ||
| paused = true; | ||
| readable.pause(); | ||
| } | ||
| }; | ||
| function turn(onOff) { | ||
| for (let i = 0; i < closeEvents.length; i++) | ||
| readable[onOff](closeEvents[i], onClose); | ||
| for (let i = 0; i < itemEvents.length; i++) | ||
| readable[onOff](itemEvents[i], onItem); | ||
| for (let i = 0; i < itemEvents.length; i++) | ||
| readable[onOff](errorEvents[i], onError); | ||
| } | ||
|
|
||
| turn('on'); | ||
|
|
||
| function next() { | ||
| if (!queue.isEmpty()) { | ||
| const value = queue.shift(); | ||
| queueSize--; | ||
| if (queueSize < resumeThreshold) { | ||
| paused = false; | ||
| readable.resume(); | ||
| } | ||
| return { | ||
| done: false, | ||
| value, | ||
| }; | ||
| } | ||
| if (error) { | ||
| throw error; | ||
| } | ||
| if (done) { | ||
| return { done }; | ||
| } | ||
| return waitNext(readable, next, events); | ||
| } | ||
|
|
||
| return { | ||
| next, | ||
| highWaterMark, | ||
| get isPaused() { | ||
| return paused; | ||
| }, | ||
| get queueSize() { | ||
| return queueSize; | ||
| }, | ||
| [SymbolAsyncIterator]() { return result; }, | ||
| }; | ||
| }; |
Oops, something went wrong.