V3 unioniterator #79
V3 unioniterator #79
Conversation
asynciterator.ts
Outdated
| for (const source of this._sources) | ||
| for (const source of this._buffer) { | ||
| this._removeSource(source); | ||
| source.destroy(); |
There was a problem hiding this comment.
This one of 2 behaviors I am not happy about here is that we don't also destroy any sources that have not been produced by the this._sources iterator. But I can't think of a clean way of doing this here as this._sources may have a large amount of overhead to generate the rest of the sources to the end
| // Adds the given source to the internal sources array | ||
| protected _addSource(source: InternalSource<any>) { | ||
| source[DESTINATION] = this; | ||
| source.on('error', destinationEmitError); |
There was a problem hiding this comment.
This is the second of two behaviors I am not happy with. In particular; errors will only be propogated once we have pulled it from this._source. An error may occur before this.
| it('should pass errors', () => { | ||
| const callback = sinon.spy(); | ||
| const error = new Error('error'); | ||
| iterator.once('error', callback); | ||
| sourceIterator.emit('error', error); | ||
| callback.should.have.been.calledOnce; | ||
| callback.should.have.been.calledWith(error); | ||
| }); |
There was a problem hiding this comment.
This is an example of an error that can no longer be propogated with the changes we have made.
|
|
||
| sources[0].closed.should.be.true; | ||
| sources[1].closed.should.be.true; | ||
| sources[1].closed.should.be.false; |
There was a problem hiding this comment.
This is an example of where a source could not be destroyed and hence closed because it had not yet been pulled from this._source.
|
See #81 |
supercedes #65
The bolded performance regression needs to be ironed out first.
Before
For loop with 5x10^9 elems: 5.207s
UnionIterator 2x10^7 iterators: 30.297s
UnionIterator 1000x500 iterators: 1.897s
UnionIterator 1000x500 iterators - max parallelism of 1: 1.837s
After
For loop with 5x10^9 elems: 6.349s
UnionIterator 2x10^7 iterators: 3.513s
UnionIterator 1000x500 iterators: 7.219s
UnionIterator 1000x500 iterators - max parallelism of 1: 168.007ms