[transducers-async] New async packages and how they work

[dennemark]

Hi,
some questions came up on Mastodon and for documentation purposes, we shift it to GH.

I have been wondering about use cases for transducers-async. They internally use generators

When a value is consumed by calling the generator's next method, the Generator function executes until it encounters the yield keyword.

On mastodon I mistakenly expected these generators to work like co-routines, so the generator would be synchronous code. But looking at the code it seems they can run promises and only after the promise finishes, the yield is being called. Therefore transducers-async can now map/reduce... promises.

I guess what i am mainly trying to understand is, how the calls of the functions will happen. I am not fully sure how this actually works. If the promises are called synchronously or asynchronously and when the next() calls happen. A search shows the use of next()within zip (very synchronous, since await is called in a for loop) or within sync (quite async with a Promise.all call).
But how do the evaluators call these functions?

I might miss some general concepts, but I hope my questions are useful...

[postspectacular]

@dennemark You're correct that generators are playing a role here (and they are co-routines), but just as ther're two kinds of functions in JS/TS, there're two kinds of generators: synchronous & asynchronous. The latter is a much more recent addition to the language, and as I wrote on Mastodon, I already started working on this package in 2018, but then stopped again after realising that there's no actual support for these async generators (and even async fns themselves were not widely supported, only Promises (without polyfillss)...). The main differences between the sync & async generators are that yield x; in the latter will yield a promise of the yielded value, rather than the x itself. Also, there're currently no built-in datatypes which implement the AsyncIterable or AsyncIterator interfaces (meaning only async generators can be used as data source). In contrast, arrays, sets, maps, strings all implement the synchronous Iteratable/Iterator interfaces...

Both the the older thi.ng/transducers and the new thi.ng/transducers-async each provide 3 groups of functions: generators/iterators, transducers and reducers. You can use most (all?) of the synchronous versions in the async context, but not vice versa. I hope the example below (without any libraries) helps to better understand how these 3 groups work together:

0. Some helper types (see package docs for details):

type MaybePromise<T> = T | Promise<T>;

type AsyncReducer<A,B> = [
  // init
  () => MaybePromise<B>,
  // completion
  (acc: B) => MaybePromise<B>,
  // reduction
  (acc: B, x: A) => MaybePromise<B>
];

type AsyncTransducer<A, B> = (r: AsyncReducer<B, any>) => AsyncReducer<A, any>;

1. A simple async generator as data source:

async function* countdown(n: number, delay = 1000) {
  while(n >= 0) {
    yield n--;
    await wait(delay);
  }
}

const wait = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));

2. Some helper types and a simple map transducer for transforming async values:

const map = <A, B>(fn: (x: A) => MaybePromise<B>): AsyncTransducer<A, B> =>
  ([init, complete, reduce]: AsyncReducer<B, any>) => [
    init,
    complete,
    async (acc, x) => reduce(acc, await fn(x))
  ];

3. An async reducer to collect values in an array:

const push = <A, B>(): AsyncReducer<A, B> => [
  async () => [],
  async (acc) => acc,
  async (acc, x) => (acc.push(x), acc)
];

4. Pulling it all together:

Let's start with the async generator only:

// note the new `for await` loop form, which can ONLY be used with async iterables (and only in async contexts)
for await (let x of countdown(3)) console.log(x);

// will loutput (once per second)
// 3
// 2
// 1
// 0

Now use the reducer to collect these values:

const [init, complete, reduce] = push();
const acc = await init();

for await (let x of countdown(3)) {
  acc = await reduce(acc, x);
}

console.log("final result:", await complete(acc));

// after a few seconds:
// final result: [3, 2, 1, 0]

5. Finally, use the transducer to augment the simple push() with a map() transform:

// actual mapping function
const xform = map(async (x: number) => x * 10);

// compose xform with reducer
const [init, complete, reduce] = xform(push());
const acc = await init();

// rest is unchanged...
for await (let x of countdown(3)) {
  acc = await reduce(acc, x);
}

console.log("final result:", await complete(acc));
// final result: [ 30, 20, 10, 0 ]

5a. Use the library for the same result:

import { map, push, range, transduce } "@thi.ng/transducers-async";

console.log(await transduce(map(async (x: number)=> x * 10), push(), range(3,-1, 1000)))
// [ 30, 20, 10, 0 ]

Hope that helps... :)

[dennemark]

This does help! But excuse the nitpicking, I try to understand a bit more.

I could also write await Promise.all(countdown(3).map((x)=>x*10)) at least if countdown is not a generator function and I would get the same result with not much boilerplate.

But the goals of the package is to bring this logic into transducers, so that you can use async components within rdom rstream etc. I guess. And I am wondering if the generators have the advantage that we could also directly render some components while others are still loading. So this would be a great advantage over the simple Promise.all call.

Probably the above example might be interesting with a random delay. - Something I should try.

Maybe some parts of this discussion can be used for a readme later.

[postspectacular]

Maybe it is my mistake/laziness to give a too simplistic example, but your await Promise.all(countdown(3).map((x)=>x*10)) would only be useful in a very small subset of cases where you want wait for all the results to arrive at once (and as you said yourself, the countdown() itself couldn't be async either). The whole point of async functions and generators is to have more flexibility with running concurrent processes (incl. co-routines) and the async versions of transducers are simply an architectural pattern to functionally compose powerful, multi-stage processing pipelines from those, where each processing step can be async itself. The above examples also didn't demonstrate the idea of early termination or composing multiple transducers, so sorry, I just tried to keep it super simple. If you haven't done so yet, maybe you find my blog post about (synchronous) transducers helpful too, which goes into a lot more detail... Though again, the existing synchronous transducers can't be used (on their own) for async pipelines, since the intermediate promises need to be resolved/awaited. This is the main addition to this new package. Also, rdom & rstream usages are just some (future) use cases for these async transducers[1]. Just as with synchronous transducers, the range of applications for this package is much more general...

As for updating the readme, yes totally, I just haven't had bandwidth yet to write proper documentation for this package... I recommend looking at the various tests, though... maybe they help 🤷‍♂️

[1] In fact, I have a feeling many aspects of rstream can eventually be superceded by just using async transducers, but it will take me a while to map out all the overlaps... There're already versions of sync(), merge(), sidechain() which behave very similar to the rstream constructs, but they're just using async iterables...

[dennemark]

I think your remarks are on point. I have read your blog posts some years ago and might need to revisit them. I sometimes use some functions like range etc., but often fallback to classic javascript functions, since they are most time enough for my use cases and less overhead to learn for other coders. But I enjoy it when using i.e. hiccup-svg.

I would love to know on my first post already, what I am trying to figure out... But reflecting from our conversation, I think my main point is to know when to use this package and what advantages it gives me over other existing functions. And now I think I can imagine it much better. Correct me if i am wrong, or if it is helpful, add those examples to the readme:

• Fetch a list of remote data asynchronously and render each item as soon as its data is processed.
• Query and process map tiles: fetch data per tile, process geometry data per tile on a worker, render on canvas.
  (i am currently looking into this..., so it might be interesting. caching would need to get into this process too.)

React has this pattern of rendering a fallback like a skeleton while data is being fetched (Suspense), I guess if one would have a reference on the fallback i.e. in the init() function, this fallback could be replaced once the actual component is ready.