feat: initial async generator support #1560

aearly · 2018-07-10T06:51:26Z

We have to create a special eachOfLimit implementation for this, but once fully tested, all collections methods should fall into place.

megawac · 2018-07-10T13:49:38Z

lib/internal/asyncEachOfLimit.js

+        running -= 1;
+        if (err) return handleError(err)
+
+        if (err === false) {


Can we avoid calling replenish int his case and go straight to calling the complete callback?

You're right, I should return in this block.

megawac · 2018-07-10T13:58:32Z

lib/internal/asyncEachOfLimit.js

+
+    function replenish() {
+        //console.log('replenish')
+        if (running >= limit || awaiting) return


How do you end up with more than one generator running asynchronously with this code? If I understand the code correctly, replenish only gets called

to spawn the generator when asyncEachOfLimit is called

whenever a generator completes

after iterateeCallback is called

Do we not need to loop limit - running inside of replenish to call several generators at once?

You'd only end up with more than one iteratee running if your iteratee takes longer to process than your generator takes to generate. For example, in the tests, I have the generator ticking every 1 ms, but have the iteratee take 5 ms.

Async generators are like streams -- you get a linear sequence of items. You can't await multiple items at the same time (and if you did, you'd get the same item multiple times, as you do when you call .then() on the same promise multiple times). This is why we can't loop in replenish -- we'd end up calling the iteratee with the same item multiple times.

I see thanks for explaining, I haven't used generators before.

So for my understanding, the replenish inside of the generator's then function is to spawn the asynchronous processes up until limit are running and the replenish inside of the iterateeCallback is to trigger additional items to run if the limit was reached?

You can't await multiple items at the same time (and if you did, you'd get the same item multiple times...)

Are you sure of this, as is right now we have the possibility of calling next multiple times before the promise resolves if iteratee is synchronous. In such a case will things break?

This is why we can't loop in replenish -- we'd end up calling the iteratee with the same item multiple times.

Sorry, I may be misunderstanding something, but I thought calls to next are queued and resolved separately. I'm looking at the Async iterators and async iterables section of the proposal.

That being said, I think we should still probably avoid looping in replenish as we still have to wait for earlier promises to resolve.

Ah, you're right, I was confusing .next() with .then(). We can call .next() as many times as we need to, perhaps up to the concurrency limit.

megawac · 2018-07-10T20:00:49Z

lib/internal/asyncEachOfLimit.js

+        if (running >= limit || awaiting) return
+        //console.log('replenish awaiting')
+        awaiting = true
+        generator.next().then(({value, done: iterDone}) => {


Can you have a generator that has no items?

Nevermind dumb question, I see how this works

megawac · 2018-07-10T20:06:53Z

lib/internal/asyncEachOfLimit.js

+            //console.log('done')
+            return callback(null);
+        }
+        replenish()


I think you have to avoid calling replenish when done === true or have replenish check if done === true.

Lets say hypothetically we have a limit of 2 and the final 2 iteratees are running. When one of these iteratees resolves, running = 1 and done = true but we'll still replenish and call the generator.next().then(). This has the possibility of entering a race condition, if the other iteratee resolves in the time between the then callback is called, you can actually end up calling callback twice

hargasinski · 2018-07-10T23:59:04Z

lib/internal/asyncEachOfLimit.js

+        if (running >= limit || awaiting) return
+        //console.log('replenish awaiting')
+        awaiting = true
+        generator.next().then(({value, done: iterDone}) => {


Similar to what @megawac was saying, I think we need a if (done) return; line at the very top of this callback. If one of the previous iteratees cancelled or errored, we shouldn't be invoking the iteratee with successive items.

For example, if the first iteratee starts and we start waiting for the second item. If the first iteratee cancels before the then callback is called with the second item, when it is called, we would still invoke the iteratee with the second item.

hargasinski · 2018-07-11T00:04:54Z

lib/internal/eachOfLimit.js

@@ -15,6 +17,9 @@ export default (limit) => {
        if (!obj) {
            return callback(null);
        }
+        if (isAsyncGenerator(obj)) {


This is a minor thing, but should this be expanded to support async iterators in general? Something similar to how we currently have getIterator except with Symbol.asyncIterator.

Ah, good point, didnt know about that detail of the spec.

lib/internal/asyncEachOfLimit.js

+        replenish()
+    }
+
+    function handleError(err) {


aearly · 2018-07-11T04:26:46Z

lib/internal/asyncEachOfLimit.js

+
+    function replenish() {
+        //console.log('replenish')
+        if (running >= limit || awaiting || done) return


BTW, I experimented with calling replenish() multiple times, awaiting mutiple .next()s at the same time, but there's some grey area in how it could work. Since you don't know the length of the iterator up front, you have to set some arbitrary limits in the pure parallel case, e.g. 10. Whenever you call .next() N times synchronously, you end up calling next() and receiving {done: true} N times spuriously at the end. We don't need to await multiple .next()s at the same time, because as far as I know, there will be a strict linear sequence of yields in the generator, meaning it's pointless to await more than one. It leaves the job of backpressure to the generator implementation too.

megawac · 2018-07-22T04:21:43Z

@aearly can we update our eslint config to be consistent on semi colons please

aearly · 2018-07-30T00:36:26Z

Okay, I've added support for async iterables. Implementation was pretty simple, hard part is getting all the various babel stuff to be happy.

aearly · 2018-07-30T02:08:43Z

Also, I'd want to do a linter change in a separate PR. Going to be a big ugly diff if we switch.

[wip] initial async generator support

8567a7f

aearly requested review from beaugunderson and megawac July 10, 2018 06:51

aearly added this to the 3.0 milestone Jul 10, 2018

aearly added the feature label Jul 10, 2018

megawac reviewed Jul 10, 2018

View reviewed changes

hargasinski reviewed Jul 10, 2018

View reviewed changes

hargasinski reviewed Jul 11, 2018

View reviewed changes

lib/internal/asyncEachOfLimit.js

replenish()

}

function handleError(err) {

This comment was marked as resolved.

Sign in to view

tighten up implementation

aa1ec94

aearly commented Jul 11, 2018

View reviewed changes

aearly added 3 commits July 29, 2018 16:41

fix linter

82cef4f

add suport for async iterables

59fea53

docs

2cdaa45

aearly added 3 commits July 29, 2018 18:08

bump timeout

e491bd1

switch to babel-minify

0b5c27e

fix rollup build

c99905a

aearly changed the title ~~[wip] initial async generator support~~ feat: initial async generator support Aug 5, 2018

aearly merged commit 61268c5 into master Aug 5, 2018

aearly deleted the async-generator-support branch August 5, 2018 20:36

dependabot bot mentioned this pull request Mar 16, 2021

Bump async from 2.6.3 to 3.2.0 dudleycarr/nsqjs#314

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: initial async generator support #1560

feat: initial async generator support #1560

aearly commented Jul 10, 2018

megawac Jul 10, 2018

aearly Jul 10, 2018

megawac Jul 10, 2018

aearly Jul 10, 2018

megawac Jul 10, 2018

megawac Jul 10, 2018

hargasinski Jul 10, 2018

aearly Jul 11, 2018

megawac Jul 10, 2018

megawac Jul 10, 2018

megawac Jul 10, 2018

hargasinski Jul 10, 2018

hargasinski Jul 11, 2018

aearly Jul 11, 2018

This comment was marked as resolved.

aearly Jul 11, 2018

megawac commented Jul 22, 2018

aearly commented Jul 30, 2018

aearly commented Jul 30, 2018

feat: initial async generator support #1560

feat: initial async generator support #1560

Conversation

aearly commented Jul 10, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This comment was marked as resolved.

Choose a reason for hiding this comment

megawac commented Jul 22, 2018

aearly commented Jul 30, 2018

aearly commented Jul 30, 2018