caolan / caolan/highland

parallel does not respect back-pressure

Open
#571 5 comments 0 reactions 0 assignees View on GitHub
Dominant language
JavaScript
Stars
3.4k
Forks
145
PR merge metrics
No merged PRs in 30d

Description

Because of some ORM limitations I want to split fetching of a huge # of database records into batches, doing lookups based on primary key. I have a function that takes an array of said primary keys, and then my goal is to use highland to:

- split the array into smaller batches of primary keys, each of one which results in a database query (I want some batching of keys so don't want to issue a query for each key, but don't want to issue one query for 50k records either)

- since talking to the database is async I'm okay having some parallel # of queries ongoing to the database (but want a cap on this so as to not take all of the database's resources for just this function call)

- then I want to rejoin into a single array

I've come up with the following which is working great. However AFAICT it does not respect backpressure from the node stream it is piped to. In my use case I would pipe it through a JSON stringifier and then pipe to node's HTTP(S) response stream. If the consuming web client is slow, the memory consumption of parallel not respecting backpressure & continuing to buffer makes this undesirable.

I tested for backpressure using https://github.com/dominictarr/stream-tester#createpausestream-prob-delay plus it looks like from the Highland docs that parallel will buffer infinitely. I appreciate & understand that it needs to buffer at least some when there are parallelized calls but don't want it to buffer to infinity.

```js
let primaryKeys = ['uuid1', 'uuid2', 'uuid3'] // imagine there are 50k of these

function getBatchCb(ids, cb) {
getFromDatabase(ids, cb)
}
let getBatch = _.wrapCallback(getBatchCb)

let stream = _(primaryKeys)
.batch(5)
.map(getBatch)
.parallel(4)
.sequence()

stream.pipe(nodeWriteableStream)
```

Highland's `mergeWithLimit` seems to work in the way I'd like but ideally I would like to preserve order of the resulting stream (imagine if the database primary keys are sorted in some way).

Contributor guide

Open the contributing guide

Research direction

Start with Highland's parallel and mergeWithLimit entry points, then reproduce the behavior using stream-tester's pause stream and the shown batch/map/parallel/sequence pipeline. Compare buffering and ordering under a slow Node writable stream. Done means parallel applies bounded back-pressure without losing the requested result order.

Written by the indexing model from the issue text.

Assessment

Tech stack
javascript, node.js
Domain
backend
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.