Migrate from Orion

[ahuang11]

Move DaskTaskRunner from Orion

[ahuang11]

It stalls on this with 3.7

================================================================================================== ERRORS ==================================================================================================
___________________________________ ERROR at teardown of TestDaskTaskRunner.test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1] ___________________________________

self = <FixtureDef argname='event_loop' scope='session' baseid='tests/test_task_runners.py'>
request = <SubRequest 'event_loop' for <Function test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1]>>

    def finish(self, request: SubRequest) -> None:
        exc = None
        try:
            while self._finalizers:
                try:
                    func = self._finalizers.pop()
                    func()
                except BaseException as e:
                    # XXX Only first exception will be seen by user,
                    #     ideally all should be reported.
                    if exc is None:
                        exc = e
            if exc:
>               raise exc

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/_pytest/fixtures.py:1019:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <FixtureDef argname='event_loop' scope='session' baseid='tests/test_task_runners.py'>
request = <SubRequest 'event_loop' for <Function test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1]>>

    def finish(self, request: SubRequest) -> None:
        exc = None
        try:
            while self._finalizers:
                try:
                    func = self._finalizers.pop()
>                   func()

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/_pytest/fixtures.py:1012:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <FixtureDef argname='hosted_orion_api' scope='session' baseid='tests/test_task_runners.py'>
request = <SubRequest 'hosted_orion_api' for <Function test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_existing_cluster-exception1]>>

    def finish(self, request: SubRequest) -> None:
        exc = None
        try:
            while self._finalizers:
                try:
                    func = self._finalizers.pop()
                    func()
                except BaseException as e:
                    # XXX Only first exception will be seen by user,
                    #     ideally all should be reported.
                    if exc is None:
                        exc = e
            if exc:
>               raise exc

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/_pytest/fixtures.py:1019:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <FixtureDef argname='hosted_orion_api' scope='session' baseid='tests/test_task_runners.py'>
request = <SubRequest 'hosted_orion_api' for <Function test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_existing_cluster-exception1]>>

    def finish(self, request: SubRequest) -> None:
        exc = None
        try:
            while self._finalizers:
                try:
                    func = self._finalizers.pop()
>                   func()

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/_pytest/fixtures.py:1012:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    def finalizer() -> None:
        """Yield again, to finalize."""

        async def async_finalizer() -> None:
            try:
                await gen_obj.__anext__()
            except StopAsyncIteration:
                pass
            else:
                msg = "Async generator fixture didn't stop."
                msg += "Yield only once."
                raise ValueError(msg)

>       event_loop.run_until_complete(async_finalizer())

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/pytest_asyncio/plugin.py:291:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>
future = <Task pending coro=<_wrap_asyncgen.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer.<locals>.async_finalizer() ru...-packages/pytest_asyncio/plugin.py:283> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x127aa7588>()]>>

    def run_until_complete(self, future):
        """Run until the Future is done.

        If the argument is a coroutine, it is wrapped in a Task.

        WARNING: It would be disastrous to call run_until_complete()
        with the same coroutine twice -- it would wrap it in two
        different Tasks and that can't be good.

        Return the Future's result, or raise its exception.
        """
        self._check_closed()
        self._check_runnung()

        new_task = not futures.isfuture(future)
        future = tasks.ensure_future(future, loop=self)
        if new_task:
            # An exception is raised if the future didn't complete, so there
            # is no need to log the "destroy pending task" message
            future._log_destroy_pending = False

        future.add_done_callback(_run_until_complete_cb)
        try:
>           self.run_forever()

../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/base_events.py:574:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def run_forever(self):
        """Run until stop() is called."""
        self._check_closed()
        self._check_runnung()
        self._set_coroutine_origin_tracking(self._debug)
        self._thread_id = threading.get_ident()

        old_agen_hooks = sys.get_asyncgen_hooks()
        sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
                               finalizer=self._asyncgen_finalizer_hook)
        try:
            events._set_running_loop(self)
            while True:
>               self._run_once()

../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/base_events.py:541:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _run_once(self):
        """Run one full iteration of the event loop.

        This calls all currently ready callbacks, polls for I/O,
        schedules the resulting callbacks, and finally schedules
        'call_later' callbacks.
        """

        sched_count = len(self._scheduled)
        if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
            self._timer_cancelled_count / sched_count >
                _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
            # Remove delayed calls that were cancelled if their number
            # is too high
            new_scheduled = []
            for handle in self._scheduled:
                if handle._cancelled:
                    handle._scheduled = False
                else:
                    new_scheduled.append(handle)

            heapq.heapify(new_scheduled)
            self._scheduled = new_scheduled
            self._timer_cancelled_count = 0
        else:
            # Remove delayed calls that were cancelled from head of queue.
            while self._scheduled and self._scheduled[0]._cancelled:
                self._timer_cancelled_count -= 1
                handle = heapq.heappop(self._scheduled)
                handle._scheduled = False

        timeout = None
        if self._ready or self._stopping:
            timeout = 0
        elif self._scheduled:
            # Compute the desired timeout.
            when = self._scheduled[0]._when
            timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)

        if self._debug and timeout != 0:
            t0 = self.time()
            event_list = self._selector.select(timeout)
            dt = self.time() - t0
            if dt >= 1.0:
                level = logging.INFO
            else:
                level = logging.DEBUG
            nevent = len(event_list)
            if timeout is None:
                logger.log(level, 'poll took %.3f ms: %s events',
                           dt * 1e3, nevent)
            elif nevent:
                logger.log(level,
                           'poll %.3f ms took %.3f ms: %s events',
                           timeout * 1e3, dt * 1e3, nevent)
            elif dt >= 1.0:
                logger.log(level,
                           'poll %.3f ms took %.3f ms: timeout',
                           timeout * 1e3, dt * 1e3)
        else:
>           event_list = self._selector.select(timeout)

../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/base_events.py:1750:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <selectors.KqueueSelector object at 0x107141e48>, timeout = 0.39379980802880254

    def select(self, timeout=None):
        timeout = None if timeout is None else max(timeout, 0)
        max_ev = len(self._fd_to_key)
        ready = []
        try:
>           kev_list = self._selector.control(None, max_ev, timeout)
E           KeyboardInterrupt

../../../../.pyenv/versions/3.7.12/lib/python3.7/selectors.py:558: KeyboardInterrupt

During handling of the above exception, another exception occurred:

fixturedef = <FixtureDef argname='event_loop' scope='session' baseid='tests/test_task_runners.py'>
request = <SubRequest 'event_loop' for <Function test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1]>>

    @pytest.hookimpl(trylast=True)
    def pytest_fixture_post_finalizer(fixturedef: FixtureDef, request: SubRequest) -> None:
        """Called after fixture teardown"""
        if fixturedef.argname == "event_loop":
            policy = asyncio.get_event_loop_policy()
            try:
                loop = policy.get_event_loop()
            except RuntimeError:
                loop = None
            if loop is not None:
                # Clean up existing loop to avoid ResourceWarnings
                loop.close()
            new_loop = policy.new_event_loop()  # Replace existing event loop
            # Ensure subsequent calls to get_event_loop() succeed
>           policy.set_event_loop(new_loop)

../../../../.pyenv/versions/3.7.12/lib/python3.7/site-packages/pytest_asyncio/plugin.py:367:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:1134: in set_event_loop
    self._watcher.attach_loop(loop)
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:889: in attach_loop
    self._do_waitpid_all()
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:959: in _do_waitpid_all
    self._do_waitpid(pid)
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:993: in _do_waitpid
    callback(pid, returncode, *args)
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:204: in _child_watcher_callback
    self.call_soon_threadsafe(transp._process_exited, returncode)
../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/base_events.py:736: in call_soon_threadsafe
    self._check_closed()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _check_closed(self):
        if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

../../../../.pyenv/versions/3.7.12/lib/python3.7/asyncio/base_events.py:479: RuntimeError
----------------------------------------------------------------------------------------- Captured stderr teardown -----------------------------------------------------------------------------------------
INFO:     Shutting down
INFO:     Waiting for application shutdown.
INFO:     Application shutdown complete.
INFO:     Finished server process [84664]
============================================================================================= warnings summary =============================================================================================
tests/test_task_runners.py: 34 warnings
  /Users/andrew/.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:878: RuntimeWarning: A loop is being detached from a child watcher with pending handlers
    RuntimeWarning)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
===Flaky Test Report===

test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[default_dask_task_runner] passed 1 out of the required 1 times. Success!
test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_existing_cluster] passed 1 out of the required 1 times. Success!
test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_process_pool] passed 1 out of the required 1 times. Success!
test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_thread_pool] passed 1 out of the required 1 times. Success!

===End Flaky Test Report===
========================================================================================= short test summary info ==========================================================================================
ERROR test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1] - RuntimeError: Event loop is closed
===================================================================== 60 passed, 28 skipped, 34 warnings, 1 error in 222.38s (0:03:42) ===============================================================

[ahuang11]

Interestingly, if I run it alone without the other tasks running, it works...

TestDaskTaskRunner.test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1]

python37 -m pytest . -vv -k exception
========================================================================================== test session starts ===========================================================================================
platform darwin -- Python 3.7.12, pytest-7.1.2, pluggy-1.0.0 -- /Users/andrew/.pyenv/versions/3.7.12/bin/python
cachedir: .pytest_cache
rootdir: /Users/andrew/Applications/python/prefect-dask, configfile: setup.cfg
plugins: anyio-3.6.1, asyncio-0.18.3, flaky-3.7.0
asyncio: mode=auto
collected 88 items / 80 deselected / 8 selected

tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[default_dask_task_runner-exception0] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 12%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[default_dask_task_runner-exception1] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 25%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_existing_cluster-exception0] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 37%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_existing_cluster-exception1] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 50%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_process_pool-exception0] PASSED                                               [ 62%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_process_pool-exception1] PASSED                                               [ 75%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception0] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 87%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1] SKIPPED (This will abort the run for CONCURRENT task runners.) [100%]

============================================================================== 2 passed, 6 skipped, 80 deselected in 7.14s ===============================================================================

[ahuang11]

Actually it does pass, but the teardown is messed up.

tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_process_pool-exception0] PASSED                                               [ 96%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_process_pool-exception1] PASSED                                               [ 97%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception0] SKIPPED (This will abort the run for CONCURRENT task runners.) [ 98%]
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1] SKIPPED (This will abort the run for CONCURRENT task runners.) [100%]^C
tests/test_task_runners.py::TestDaskTaskRunner::test_wait_captures_exceptions_as_crashed_state[dask_task_runner_with_thread_pool-exception1] ERROR

[ahuang11]

Okay it's related to existing cluster; when I comment out dask_task_runner_with_existing_cluster, it works.

class TestDaskTaskRunner(TaskRunnerStandardTestSuite):
    @pytest.fixture(
        params=[
            default_dask_task_runner,
            # dask_task_runner_with_existing_cluster,
            dask_task_runner_with_process_pool,
            dask_task_runner_with_thread_pool,
        ]
    )

[ahuang11]

Pretty stumped on this; tests pass 100%, but hangs

python 3.7

python37 -m pytest . -vv -k test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type
========================================================================================== test session starts ===========================================================================================
platform darwin -- Python 3.7.12, pytest-7.1.2, pluggy-1.0.0 -- /Users/andrew/.pyenv/versions/3.7.12/bin/python
cachedir: .pytest_cache
rootdir: /Users/andrew/Applications/python/prefect-dask, configfile: setup.cfg
plugins: anyio-3.5.0, env-0.6.2, flaky-3.7.0, timeout-2.1.0, asyncio-0.18.2
asyncio: mode=auto
collected 22 items / 21 deselected / 1 selected

tests/test_task_runners.py::TestDaskTaskRunner::test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_existing_cluster] <- ../orion/src/prefect/testing/standard_test_suites.py PASSED [100%]^C

============================================================================================ warnings summary ============================================================================================
tests/test_task_runners.py::TestDaskTaskRunner::test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_existing_cluster]
  /Users/andrew/.pyenv/versions/3.7.12/lib/python3.7/asyncio/unix_events.py:878: RuntimeWarning: A loop is being detached from a child watcher with pending handlers
    RuntimeWarning)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! KeyboardInterrupt !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
/Users/andrew/.pyenv/versions/3.7.12/lib/python3.7/selectors.py:558: KeyboardInterrupt
(to show a full traceback on KeyboardInterrupt use --full-trace)
============================================================================== 1 passed, 21 deselected, 1 warning in 15.42s ==============================================================================

python 3.9

python -m pytest . -vv -k test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type
========================================================================================== test session starts ===========================================================================================
platform darwin -- Python 3.9.7, pytest-7.0.1, pluggy-1.0.0 -- /Users/andrew/mambaforge/bin/python
cachedir: .pytest_cache
rootdir: /Users/andrew/Applications/python/prefect-dask, configfile: setup.cfg
plugins: anyio-3.5.0, env-0.6.2, flaky-3.7.0, timeout-2.1.0, asyncio-0.18.2, respx-0.19.2, cookies-0.6.1
asyncio: mode=auto
collected 22 items / 21 deselected / 1 selected

tests/test_task_runners.py::TestDaskTaskRunner::test_sync_tasks_run_concurrently_with_nonsequential_concurrency_type[dask_task_runner_with_existing_cluster] <- ../orion/src/prefect/testing/standard_test_suites.py PASSED [100%]

==================================================================================== 1 passed, 21 deselected in 7.37s ====================================================================================

[ahuang11]

Okay it was the event loop; I had to use the event loop from https://github.com/PrefectHQ/orion/blob/main/tests/conftest.py

[desertaxle]

Looking good! Few comments from me around commenting/documentation.