Not that great for implementing transducers?

[jedwards1211]

Since the API is so similar to Observable, I find it great for consuming observable-like things but I'm not as sold for using it on non-observable-like things, like transducers or other async iterator combiners.

It seems almost simpler to implement transducers like map correctly using a raw async iterator protocol than using Repeater.

The naive assumption (in repeaterMap below) is to use a for await loop, but this keeps consuming input after the caller returns, which would be especially bad if the input is infinite. Whereas a raw async iterator can immediately forward return and throw calls to the input.

A less naive attempt (awkwardRepeaterMap) is to do the for await in a separate function and break once stop is detected. However more calls to next() still go through than with the raw async iterator.

Thoughts?

Code

const { Repeater } = require('@repeaterjs/repeater')

const range = n => ({
  i: 0,
  n,
  async next() {
    console.log('next()')
    const i = ++this.i
    if (i >= this.n) return { done: true }
    return { value: i, done: false }
  },
  async return() {
    console.log('return()')
    this.i = this.n
    return { done: true }
  },
  async throw(error) {
    console.log('throw()')
    this.i = this.n
    return { done: true }
  },
  [Symbol.asyncIterator]() {
    return this
  },
})

const repeaterMap = (iterable, iteratee) =>
  new Repeater(async (push, stop) => {
    for await (const value of iterable) push(iteratee(value))
    await stop
  })

const rawMap = (iterable, iteratee) => ({
  iterator: iterable[Symbol.asyncIterator](),
  async next() {
    const { value, done } = await this.iterator.next()
    return { value: iteratee(value), done }
  },
  return() {
    return this.iterator.return()
  },
  throw(error) {
    return this.iterator.throw(error)
  },
  [Symbol.asyncIterator]() {
    return this
  },
})


const awkwardRepeaterMap = (iterable, iteratee) =>
  new Repeater(async (push, stop) => {
    let stopped = false
    async function next() {
      for await (const value of iterable) {
        if (stopped) break
        push(iteratee(value))
      }
    }
    next()
    await stop
    stopped = true
  })

async function go() {
  console.log('rawMap')
  for await (const i of rawMap(range(10), i => i * 2)) {
    console.log(i)
    if (i >= 5) break
  }
  console.log('\nrepeaterMap')
  for await (const i of repeaterMap(range(10), i => i * 2)) {
    console.log(i)
    if (i >= 5) break
  }
  console.log('\nawkwardRepeaterMap')
  for await (const i of awkwardRepeaterMap(range(10), i => i * 2)) {
    console.log(i)
    if (i >= 5) break
  }
}

go()

Output

rawMap
next()
2
next()
4
next()
6
return()

repeaterMap
next()
next()
next()
2
next()
next()
next()
4
next()
next()
6
next()
next()

awkwardRepeaterMap
next()
next()
next()
2
next()
next()
next()
4
next()
next()
6
next()
return()

[jedwards1211]

To be fair, my rawMap implementation is probably a bit flawed as far as return waiting for calls to next to finish. (Nevermind, it's fine because it just defers return to its input). But in any case, it seems like an ideal map implementation shouldn't eagerly pull values from its input.

[brainkim]

@jedwards1211 have you tried awaiting the push call in your repeater examples? Without any sort of pausing mechanism within the for await loop, you’re simply unspooling the range iterator as fast as you can. Your raw iterator runs step-wise through the range iterator because it only calls next once per iteration. In other words do something like this:

const repeaterMap = (iterable, iteratee) =>
  new Repeater(async (push, stop) => {
    for await (const value of iterable) await push(iteratee(value))
    await stop
  })

I write about what the promise returned from push does in the overview:

In addition, the executor API exposes promises which resolve according to the state of the repeater. The push function returns a promise which resolves when next is called, and the stop function doubles as a promise which resolves when the repeater is stopped. As a promise, stop can be awaited to defer event listener cleanup.

const repeater = new Repeater(async (push, stop) => {
  console.log("repeater started!");
  await push(1);
  console.log("pushed 1");
  await push(2);
  console.log("pushed 2");
  await stop;
  console.log("done");
});
(async () => {
  console.log(await repeater.next());
  // repeater started!
  // { value: 1, done: false }
  console.log(await repeater.next());
  // "pushed 1"
  // { value: 2, done: false }
  console.log(await repeater.return());
  // "pushed 2"
  // "done"
  // { done: true }
})();

[brainkim]

Also some additional thoughts:

1. I don’t think you need repeaters to express a map combinator, async generators are more than sufficient:

async function *map(iterable, iteratee) {
  for await (const value of iterable) yield iteratee(value)
}

I also don’t export any combinator methods directly on the async iterator interface, cause this would widen the surface API, most combinators are expressible using async generators, and the iterator helper proposal would obsolete/clash with any instance methods I defined on the repeater class.

2. If you don’t have any cleanup code, you don’t need to await stop. This is usually an indication that you don’t need a repeater in the first place. Literally every single case where I’ve needed a repeater and not an async generator, I’ve used both the push and stop functions/promises.

[jedwards1211]

Using an async generator suffers the secure stop issue though, and is especially a problem with GraphQL subs at the moment. I ended up using a raw iterator for map cases to solve the memory leaks I was having, because it's the only way I can ensure return/throw get forwarded to the input and stop the underlying redis subs right away.

I didn't realize I can await push, that makes sense!

[jedwards1211]

Cool so this works well:

const repeaterMap = (iterable, iteratee) =>
  new Repeater(async (push, stop) => {
    let stopped = false
    stop.then(() => stopped = true)
    for await (const value of iterable) {
      if (stopped) break
      await push(iteratee(value))
    }
    await stop
  })

I didn't make it very clear but even though this was a toy example that doesn't need cleanup, I was trying to test whether cleanup would occur based upon either the input being return()ed or iterated to completion. (I didn't realize until doing this testing that for await only calls return() if iteration was stopped prematurely)

repeaterMap
next()
2
next()
4
next()
6
next()
return()

[brainkim]

Using an async generator suffers the secure stop issue though, and is especially a problem with GraphQL subs at the moment.

I think you said it best when you said:

Also FWIW, I don't know if I would say that async generators are the main problem here, if your async iterator creator function registers stuff before the first call to next(), the danger exists whether you're consuming the iterator with an async generator, for await loop, or even using the low level async iterator API.

tc39/proposal-async-iteration#126 (comment)

Unless I’m misunderstanding your secure stop problem, repeaters wouldn’t help you here anyways because the repeater executor won’t execute until the first time next is called, so there‘s always the possibility that even when wrapping your iterator in a repeaterMap call, you still never reach the break statement in the for await loop. The answer is to go directly to the source, and fix them so they work lazily. If you can’t, you either have to manually thread some kind of return signal or “prime” the downstream repeater so that its executor is executed at least once.

Lemme know if you have any other questions or concerns or if anything is unclear.

[jedwards1211]

Here's some test code I wrote to illustrate the secure stop problem you're talking about when using my revised repeaterMap function.

The subscribe function simulates a GraphQL subscription. Basically, we want to unsubscribe from Redis ASAP after calling return() on the async iterator. This works when using the redis async iterator directly or when wrapping it in a raw async iterator map. But when I wrap with the revised repeaterMap implementation, it doesn't unsubscribe from Redis until the next event comes through -- same problem as an async generator would have.

Theoretically I could implement map with a Repeater in such a way that it can pass return() calls to the input iterator immediately, but at that point I might as well just use a raw async iterator.

Code

const { RedisPubSub } = require('graphql-redis-subscriptions')
const { Repeater } = require('@repeaterjs/repeater')

const debug = require('debug')('pubsub')

class LoggingPubSub extends RedisPubSub {
  publish(topic, ...args) {
    debug('  publish', topic)
    return super.publish(topic, ...args)
  }
  subscribe(trigger, ...args) {
    debug('  subscribe', trigger)
    return super.subscribe(trigger, ...args)
  }
  unsubscribe(subId) {
    const [triggerName = null] = this.subscriptionMap[subId] || []
    debug('  unsubscribe', triggerName)
    return super.unsubscribe(subId)
  }
}

const pubsub = new LoggingPubSub()

const repeaterMap = (iterable, iteratee) =>
  new Repeater(async (push, stop) => {
    let stopped = false
    stop.then(() => (stopped = true))
    for await (const value of iterable) {
      if (stopped) break
      await push(iteratee(value))
    }
    await stop
  })

const rawMap = (iterable, iteratee) => ({
  iterator: iterable[Symbol.asyncIterator](),
  next() {
    return this.iterator
      .next()
      .then(i => (i.done ? i : { value: iteratee(i.value), done: i.done }))
  },
  return() {
    return this.iterator.return()
  },
  throw(error) {
    return this.iterator.throw(error)
  },
  [Symbol.asyncIterator]() {
    return this
  },
})

async function subscribe(iterable, time) {
  const iterator = iterable[Symbol.asyncIterator]()
  async function iterate() {
    iterator.next().then(({ value, done }) => {
      if (!done) iterate()
    })
  }
  iterate()
  await new Promise(resolve =>
    setTimeout(() => {
      debug('  calling return()')
      iterator.return()
      resolve()
    }, time)
  )
}

async function go() {
  let i = 0
  setInterval(() => pubsub.publish('foo', i++), 10000)
  console.log('without async generator')
  await subscribe(pubsub.asyncIterator('foo'), 2000)

  console.log('\nwith raw map')
  await subscribe(rawMap(pubsub.asyncIterator('foo'), i => i * 2), 2000)

  console.log('\nwith repeaterMap')
  await subscribe(repeaterMap(pubsub.asyncIterator('foo'), i => i * 2), 2000)
}

go()

Output

$ DEBUG=pubsub node index.js 
without async generator
  pubsub   subscribe foo +0ms
  pubsub   calling return() +2s
  pubsub   unsubscribe foo +1ms

with raw map
  pubsub   subscribe foo +2ms
  pubsub   calling return() +2s
  pubsub   unsubscribe foo +0ms

with repeaterMap
  pubsub   subscribe foo +2ms
  pubsub   calling return() +2s
  pubsub   publish foo +4s
  pubsub   unsubscribe foo +3ms

[brainkim]

Hmmmm so I read through the code and went on a walk to think about it and here’s my interpretation:

The problem isn’t repeaters or async generators but for await. As you’ve noted elsewhere, the next and return calls happen sequentially in a desugared for await loop, so you have to call the next and return fns manually if you want combinators to transparently forward return calls to the source concurrently.

In addition, however, this partly works because the raw map async iterator has potentially surprising behavior where calls to next/return don’t settle in order. In other words, in the following code:

(async () => {
  const iter = rawMap(sourceIter, (i) => i * 2);
  const nextP = iter.next();
  const returnP = iter.return();
  console.log(await Promise.race([returnP, nextP]));
})();

the race might resolve to returnP, depending on the behavior of sourceIter. For the most part, you’ll never notice this because sourceIter will usually settle iterations in order, but you’re noticing it here because the source also does not settle iterations in order:

https://github.com/davidyaha/graphql-redis-subscriptions/blob/master/src/pubsub-async-iterator.ts

If my reading of the code is correct, next and return for PubsubAsyncIterator both await subscriptionIds, but next additionally awaits this.pullValue, and while return will resolve all values in the pullQueue, thereby causing all next calls to resolve, it does not wait for all the handlers of next to fire, so it fulfills before the next. You can test this assumption by doing something like:

import { RedisPubSub } from 'graphql-redis-subscriptions';
const pubsub = new RedisPubSub();
const iter = pubsub.asyncIterator("SOMETHING_CHANGED");
const p1 = iter.next();
const p2 = iter.return("I WON");
Promise.race([p2, p1]).then((value) => console.log(value));
// {value: "I WON", done: true}

If the above code doesn’t print what I expect then I’ve misunderstood the graphql-redis-subscriptions code (sorry I don’t feel like spinning up a redis instance right now to test 😛).

[brainkim]

I would probably go inform the async-iterator-helper peeps of this possible complication with using for await for combinators but I have no desire to read through the spec, and the source for core-js is labyrinthian and github search is terrible. Maybe they call next/return manually.

[brainkim]

Theoretically I could implement map with a Repeater in such a way that it can pass return() calls to the input iterator immediately, but at that point I might as well just use a raw async iterator.

Even if you call next/return manually, repeaters can still be helpful here because they make iterations settle in order and implement the next/return/throw methods in a way that is indistinguishable from async generators. I, for instance, implement theRepeater.race, Repeater.merge, Repeater.zip, and Repeater.latest combinators as static methods which call next/return manually. I guarantee you will not find a clearer/bug-free implementation of these methods elsewhere, and they’re virtually indistinguishable from async generators to boot.

https://github.com/repeaterjs/repeater/blob/master/packages/repeater/src/repeater.ts#L502-L724

[jedwards1211]

So you're right that that code could probably be improved, but in PubSubAsyncIterator, this.subscribeAll() memoizes itself into this.subscriptionIds, and emptyQueue awaits this.subscriptionIds, so return basically ends up awaiting the same promise. I'm 99% sure looking at the code that this.subscriptionIds gets sets synchronously before the call to return(). And I assume when this.subscriptionIds resolves, next() will get resumed and call pullValue() before emptyQueue gets resumed?

[brainkim]

Ahh I see this.listening is set to false before emptyQueue awaits this.subscriptionIds, so it causes next calls which were stuck on await this.subscribeAll to continue with a call to this.return instead of this.pullValue according to the conditional expression. However I feel like this is still 1 microtick slower than the original iter.return call so the original iter.return call should still win the race, right?

So you're right that that code could probably be improved

To be clear, I wasn’t criticizing the code, I don’t think it’s of poor quality, and it seems pretty readable as far as async code usually goes. My secret wish is that someone takes the repeater codebase and rewrites it, keeping only the unit tests, because I feel like the code has gotten pretty hairy to handle a bunch of important edge-cases.

[brainkim]

Here’s how you could implement a map combinator with concurrent returns with repeaters:

function map(iterable, fn) {
  return new Repeater(async (push, stop) => {
    const iter = iterable[Symbol.asyncIterator]();
    let finalIteration;
    stop.then(() => {
      finalIteration = finalIteration || iter.return();
    });
    while (!finalIteration) {
      const iteration = await iter.next();
      if (iteration.done) {
        finalIteration = finalIteration || iteration;
        break;
      }
      await push(fn(iteration.value));
    }
    // TODO: only await finalIteration if it’s a promise
    finalIteration = await finalIteration;
    return finalIteration.value;
  });
}

Probably can be cleaned up a bit, but as you can see, repeaters are still valuable here somewhat.