Concurrent task iteration support

[joshbeckman]

Over at https://github.com/shopify/flow we have been trying to adopt the maintenance task framework and have enjoyed the benefits for our small data migrations, but our main hangup is the loong runtimes of tasks that need to operate on large datasets (e.g. all records in one table table - tens of thousands now, will be much more the future). When we tried running a recent data migration via maintenance task recently, the total time to execute would have been months.

As such, our main desire with this library would be declarative concurrency support. Is #325 (comment) still the recommendation for concurrency in the future of this library?

No immediate need for action on this - we just wanted to provide feedback on our adoption!

[etiennebarrie]

Do you think batches enumerators could help? (see #409)
Depending on your tasks, being able to update 100/1000 records at a time could substantially speed them up.

Regarding actual parallelism when running tasks, it's something that we're thinking about but we haven't made any formal plans so we can't make any promise. We can keep this issue open to continue thinking about it, start fleshing out an API, behaviour, figure out the edge cases (e.g. it will require special handling for custom enumerators which may not have a way to start a cursor randomly, but only give out one item at a time), etc.

[joshbeckman]

Batches could help with some of our task types, yes!

But we have other types of tasks that require, for example, calling an external API with an individual record and then saving that value to our database, so the batching would remove some of the overhead of the job queue itself but wouldn't give us the speed up that we would get from concurrency.

[sle-c]

I recently ran a migration on flow which mainly involves making graphQL requests to core for certain things. Processing 874k rows would take about 7 days to complete. I think allowing parallelism really helps in these cases.

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[joshbeckman]

We would still like this!

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[joshbeckman]

We would still really like this

[segiddins]

This would be incredibly useful.

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[segiddins]

Still valid

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[segiddins]

still relevant

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[joshbeckman]

Still, I would love this.

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[joshbeckman]

Still want this.

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[segiddins]

Not stale

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[segiddins]

Not stale

[github-actions[bot]]

This issue has been marked as stale because it has not been commented on in two months.
Please reply in order to keep the issue open. Otherwise, it will close in 14 days.
Thank you for contributing!

[segiddins]

Still valid

[etiennebarrie]

Turned this into a discussion to avoid the stale bot problem.

---

There are few things to think about if we want to introduce parallelism.

The first part has already been done, which was to allow multiple Runs of a Task. Now we need a good way to split and coordinate the work. A naive approach is to take the ids of the first and the last rows, and then split evenly and start n Runs, however id distribution can be sparse so that can be problematic. There's also the issue of non Active Record collections, in particular custom enumerators as mentioned already above. CSV collections might actually be the easiest ones to handle.

We don't really have any bandwidth for this however, so any contribution would be useful, even if it's not working code. Scoping out the feature, what we need (e.g. should people choose the parallelism and that's it, or should it be automatic somehow?), and how it could work (e.g. the main Run coordinates multiple "sub-runs", they take a cursor and a "end cursor", and when they're done, the main Run starts a new one to keep the parallelism, or instead should the sub-runs stay running, but get a new cursor to another part of the table that needs processing?).

[larouxn]

Been thinking about this for a little bit now. Some thoughts.

Now we need a good way to split and coordinate the work. A naive approach is to take the ids of the first and the last rows, and then split evenly and start n Runs, however id distribution can be sparse so that can be problematic.

I have two ideas, other than using min + max ID, that I hope can be used to split data and mitigate the sparse data problem. Note, these ideas are focused on the ActiveRecord relation type of dataset, not on say CSV datasets. Though CSVs, as mentioned, should be quite a bit more straightforward and something like number 1 below would work for them too.

1. Count the number of rows and then grab quartile IDs (assuming 4 sub-runs) i.e. the ID of the record 1/4 of the way into the total list of records and use those IDs to create even ±1 record entirely within Ruby/Rails code.
2. Use raw SQL in order to access NTILE to cleanly and evenly (±1 record) divide up the records via a windowing function. Something roughly like the following though I'll note I've not used NTILE before. Typically prefer to stay in ActiveRecord but sometimes raw SQL is warranted in certain circumstances and NTILE looks promising to mitigate the sparse data problem.

SELECT
  COALESCE(LAG(max_id) OVER (ORDER BY bucket),
           (SELECT MIN(id) FROM records)) as start_id,
  max_id as end_id,
  bucket
FROM (
  SELECT
    MAX(id) as max_id,
    bucket
  FROM (
    SELECT id, NTILE(#{concurrency_count}) OVER (ORDER BY id) as bucket
    FROM records
  ) AS buckets
  GROUP BY bucket
) boundaries
ORDER BY bucket

should people choose the parallelism and that's it, or should it be automatic somehow?

I suppose people could choose concurrency i.e. number of sub-runs at the task definition level or even at the run level via custom task parameters within the web UI.

As for automatic parallelism I imagine there could be some kind of sane default, say 4 sub-runs, that one could enable by simply adding a concurrent class method from Task to a task definition. Could even allow it accept a value if one wants to use a value other than the default.

The "sane default" type approach stems from Rails philosophy as well Puma (3 threads, source) and Sidekiq (5 threads, source).

the main Run coordinates multiple "sub-runs", they take a cursor and a "end cursor", and when they're done, the main Run starts a new one to keep the parallelism, or instead should the sub-runs stay running, but get a new cursor to another part of the table that needs processing?

Was indeed looking at a single parent run (coordinator) + several child runs (processors with parent_run_id) architecture using existing cursor + new end_cursor fields to partition a given data set via the primary key. As for a parent run starting more runs when child runs finish, I was thinking more along the lines of having child runs have a subset of the total records, combined accounting for all records. Say there are 100 records, 4 sub-runs concurrency, each child run has 25 records to process.

Speaking of parent + child architecture, the parent run would also handle aggregating the status of its child runs. For example, only marking as success if all succeed, cancelling and marking run as errored if one errors out, cancelling all if run cancel is requested, and so on.

Anyway, just wanted share some ideas from my time looking into this all. 😁 🙇‍♂️