Adds as_completed utility for PrefectFuture

[jeanluciano]

Adds a utility that yields completed futures in order they are completed. Closes #14241

Example

        @task
        def sleep_task(seconds):
            sleep(seconds)
            return 42

        @flow
        def flow():
            futures = random_task.map(range(10))
            for future in as_completed(futures):
                print(future.result())

Checklist

• [ x ] This pull request includes a label categorizing the change e.g. maintenance, fix, feature, enhancement, docs.
• [ x ] This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• [ x ] If this pull request adds new functionality, it includes unit tests that cover the changes
• [ x ] If this pull request removes docs files, it includes redirect settings in mint.json.
• [ x ] If this pull request adds functions or classes, it includes helpful docstrings.

[codspeed-hq]

CodSpeed Performance Report

Merging #14641 will not alter performance

_{Comparing ascompleted-future-utility (6f0cc71) with main (530521d)}

Summary

✅ 5 untouched benchmarks

[desertaxle]

I think this implementation will be good for now, but it looks like futures early in the list will block futures later in the list from being yielded when they are complete until the futures earlier in the list are finished. Could you see if an implementation that would allow the behavior of this utility to not be dependent on the ordering of the passed in futures? I think a callback structure on PrefectFuture might make that possible.

[cicdw]

a few stray debug logs that I think need to be removed, and a question

[desertaxle]

FYI, I think we'll want to support this for the Dask and Ray flavors of PrefectFuture also

[zzstoatzz]

nice!

a couple nits and:

maybe we should add a test or two for cases where we throw exceptions / timeout? just to be explicit on what we expect

[zzstoatzz]

seems pretty elegant to me!

[cicdw]

some small nitpicks about tests and duplicated code, the core implementation makes sense to me

[cicdw]

Really weird that Ray is behaving strangely; @jeanluciano I don't think we need to optimize for a perfectly reversed output, as there is probably some latency in spinup that is allowing the 5 second sleep to execute faster. Maybe just assert that 10 is the last one returned, which will prove that as_completed affected the return order (ultimately all we need for this specific test).

[cicdw]

I would keep the Dask test as-is and only change the Ray test, between the two it's decent coverage

[jeanluciano]

Yeah I thought it had to do with the way I was converting the object ref into a future, but setting the callback on the object ref directly behaves the same way.