Vine: add wait and as_completed for futures

[BarrySlyDelgado]

Proposed Changes

This adds module functions wait and as_completed to the vine futures. in regard to #3835

Merge Checklist

The following items must be completed before PRs can be merge.
Check these off to verify you have completed all steps.

• make test Run local tests prior to pushing.
• make format Format source code to comply with lint policies. Note that some lint errors can only be resolved manually (e.g., Python)
• make lint Run lint on source code prior to pushing.
• Manual Update Update the manual to reflect user-visible changes.
• Type Labels Select a github label for the type: bugfix, enhancement, etc.
• Product Labels Select a github label for the product: TaskVine, Makeflow, etc.
• PR RTM Mark your PR as ready to merge.

[btovar]

@BarrySlyDelgado Is there an easy way to add a test for this one?

[BarrySlyDelgado]

I'll create a test that tests these methods.

[gpauloski]

Thanks, @BarrySlyDelgado! Is it also possible to either call super().__init__() in VineFuture or have VineFuture not inherit from Future?

I'm not sure which option is best, but I think both repr(future) and future.exception() will fail because both those methods are getting inherited. I've only briefly looked so do correct me if I'm wrong.

[BarrySlyDelgado]

I think i'll have it call super().__init() since we're leaning into having our executor conform more to concurrent.futures. I'll do some testing to incorporate it.

[dthain]

@BarrySlyDelgado what's the status on this one?

[gpauloski]

Hi, is this branch stable or is there still work to be done?

[BarrySlyDelgado]

The branch has been tested and should work.

[gpauloski]

Hi @BarrySlyDelgado, I've been testing out the branch and noticed a couple of issues.

Invoking future callbacks in VineFuture.result() causes infinite recursion if you access future.result() in the callback. (Maybe this is why the result, rather than the future, was passed to the callback originally?) It's safe to access future.result() in the callback of a concurrent.futures.Future because the callbacks are invoked in Future.set_result().

import time
import ndcctools.taskvine as vine

def foo() -> int:
    time.sleep(1)
    return 42

def check(future: vine.VineFuture) -> None:
    assert future.result() == 42

m = vine.FuturesExecutor(manager_name='my_manager')

t = m.submit(foo)
t.add_done_callback(check)

print(t.result())

Exceptions don't get set on nor raised by VineFuture.

import ndcctools.taskvine as vine

def foo():
    raise RuntimeError('oops')

m = vine.FuturesExecutor(manager_name='my_manager')

t = m.submit(foo)

try:
    t.result()
except RuntimeError:
    print('Correctly raised exception')
else:
    raise AssertionError('Did not raise exception')

[BarrySlyDelgado]

Thanks for letting me know, after looking at the source code the infinite recursion is due to the order in which callback functions are executed. Essentially, what is happening is that a call to f.result() executes callback functions if self._ran_functions is False and it dose not get set to True until the callbacks execute. So, a call back that also calls f.result() will also end up running callbacks. This can be fixed by setting Self._ran_functions before running the callbacks. I'll add to this PR with that change and one that raises exceptions from futures.

[BarrySlyDelgado]

@gpauloski The recursion bug should be fixed. Tasks that raise an exception now raise an exception when f.result()is called.

[gpauloski]

Thanks, @BarrySlyDelgado. The recursion is fixed for me, but I think this only raises exceptions if loading the output fails. I had to check if the output was an instance of Exception in VineFuture.result() and raise it there.

Also, I opened issue #3892 about poor performance. But it's possible the problem is not specific to the FuturesExecutor as I also observe it with Parsl's TaskVineExecutor.