Breaking bug: Accessing the loop property while the loop is not running is not supported

[jacobtomlinson]

In the dask-kubernetes CI we've started seeing all of our test runs failing. It looks like a change on distributed main is causing it.

The last known good commit is 74a1bcd, we are seeing failures with 49437c2. I am bisecting now to try and see which PR caused this but it's a slow process. See dask/dask-kubernetes#738.

The error we are seeing is

==================================== ERRORS ====================================
______________________ ERROR at setup of test_kubecluster ______________________

kopf_runner = <kopf._kits.runner.KopfRunner object at 0x7f9444a7f1c0>
docker_image = 'dask-kubernetes:dev'

    @pytest.fixture
    def cluster(kopf_runner, docker_image):
        with dask.config.set({"kubernetes.name": "foo-{uuid}"}):
            with kopf_runner:
>               with KubeCluster(image=docker_image, n_workers=1) as cluster:

dask_kubernetes/operator/kubecluster/tests/conftest.py:13: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
dask_kubernetes/operator/kubecluster/kubecluster.py:265: in __init__
    if not self.asynchronous:
/opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/distributed/utils.py:337: in asynchronous
    return in_async_call(self.loop, default=getattr(self, "_asynchronous", False))
/opt/hostedtoolcache/Python/3.10.12/x64/lib/python3.10/site-packages/distributed/deploy/cluster.py:105: in loop
    self.__loop = loop = self._loop_runner.loop
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <distributed.utils.LoopRunner object at 0x7f9440533ac0>

    @property
    def loop(self):
        loop = self._loop
        if loop is None or not loop.asyncio_loop.is_running():
>           raise RuntimeError(
                "Accessing the loop property while the loop is not running is not supported"
            )
E           RuntimeError: Accessing the loop property while the loop is not running is not supported

[jacobtomlinson]

I suspect this is being caused by #7467 as that PR includes changes to distributed.utils.LoopRunner that may be related. cc @graingert

[fjetter]

Looks like KubeCluster is using a bit of a different pattern than other cluster implementations. The dask native cluster implementations seem to expect asynchronous to be provided as an argument. KubeCluster instead tries to infer this.

There should've been a deprecation warning about this. Accessing the loop property while the loop is not running is deprecate

However, I guess this case should be handled gracefully in SyncMethodMixin.asynchronous because I think this property should be safe to evaluate and should just catch this exception in this case.
cc @graingert

[jacobtomlinson]

It looks like KubeCluster just passes asynchronous along to Cluster as a part of **kwargs and then checks whether it needs to start the loop runner via the SyncMethodMixin.asynchronous property. This is likely because we often handle this in SpecCluster but KubeCluster doesn't inherit from that because the SpecCluster pattern isn't appropriate for the Kubernetes operator design pattern.

AFAIK dask-yarn and dask-gateway also do no use SpecCluster and may also run into this problem. I'm not sure how they handle starting the LoopRunner.

If fixing this in SyncMethodMixin.asynchronous makes sense then that's great. Otherwise we could explore starting the LoopRunner in Cluster, or we could handle this in KubeCluster and catch the exception (but we would probably need to do this in other projects too).

[graingert]

there's one edge case where it's possible to pass SpecCluster(asynchronous=False, loop=IOLoop.current()) This happens when called via LocalCluster(asynchronous=False) and SpecCluster does this to detect it:

        super().__init__(
            asynchronous=asynchronous,
            loop=loop,
            name=name,
            scheduler_sync_interval=scheduler_sync_interval,
        )

        try:
            called_from_running_loop = (
                getattr(loop, "asyncio_loop", None) is asyncio.get_running_loop()
            )
        except RuntimeError:
            called_from_running_loop = asynchronous

        if not called_from_running_loop:
            self._loop_runner.start()
            self.sync(self._start)

It's likely that KubeCluster will also want to behave the same way

[jacobtomlinson]

Sure that looks more complete than the check we current have. Although isn't that effectively what calling self.asynchronous did before?

if not self.asynchronous:
    self._loop_runner.start()
    self.sync(self._start)

This change seems to have broken things more deeply. It's not even possible to use a sync cluster in a context manager now.

diff --git a/distributed/deploy/tests/test_cluster.py b/distributed/deploy/tests/test_cluster.py
index bb468209..e9313f13 100644
--- a/distributed/deploy/tests/test_cluster.py
+++ b/distributed/deploy/tests/test_cluster.py
@@ -87,3 +87,8 @@ async def test_sync_context_manager_used_with_async_cluster():
             match=r"Used 'with' with asynchronous class; please use 'async with'",
         ), cluster:
             pass
+
+
+def test_cluster_context_manager():
+    with Cluster(name="S") as cluster:
+        pass

$ pytest distributed/deploy/tests/test_cluster.py::test_cluster_context_manager 
==================================================================== test session starts =====================================================================
platform linux -- Python 3.10.11, pytest-7.3.2, pluggy-1.0.0 -- /home/jtomlinson/miniconda3/envs/dask/bin/python
cachedir: .pytest_cache
rootdir: /home/jtomlinson/Projects/dask/distributed
configfile: pyproject.toml
plugins: rerunfailures-11.1.2, timeout-2.1.0, kind-22.11.1, cov-4.1.0
timeout: 300.0s
timeout method: thread
timeout func_only: False
collected 1 item                                                                                                                                             

distributed/deploy/tests/test_cluster.py::test_cluster_context_manager FAILED                                                                          [100%]

========================================================================== FAILURES ==========================================================================
________________________________________________________________ test_cluster_context_manager ________________________________________________________________

    def test_cluster_context_manager():
>       with Cluster(name="S") as cluster:

distributed/deploy/tests/test_cluster.py:93: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
distributed/deploy/cluster.py:527: in __enter__
    return self.sync(self.__aenter__)
distributed/utils.py:355: in sync
    self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
distributed/deploy/cluster.py:105: in loop
    self.__loop = loop = self._loop_runner.loop
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <distributed.utils.LoopRunner object at 0x7fac1c6adab0>

    @property
    def loop(self):
        loop = self._loop
        if loop is None or not loop.asyncio_loop.is_running():
>           raise RuntimeError(
                "Accessing the loop property while the loop is not running is not supported"
            )
E           RuntimeError: Accessing the loop property while the loop is not running is not supported

distributed/utils.py:647: RuntimeError
==================================================================== slowest 20 durations ====================================================================

(3 durations < 0.005s hidden.  Use -vv to show these durations.)
================================================================== short test summary info ===================================================================
FAILED distributed/deploy/tests/test_cluster.py::test_cluster_context_manager - RuntimeError: Accessing the loop property while the loop is not running is not supported
===================================================================== 1 failed in 0.19s ======================================================================

We aren't seeing this in KubeCluster because we override __enter__. We don't see this until later when we try and check self.asynchronous.

That actual test I wanted to add was the following, but we don't even get that far.

def test_cluster_not_asynchronous():
    with Cluster(name="S") as cluster:
        assert not cluster.asynchronous

[jacobtomlinson]

It looks like calling self.sync(...) when no event loop is running is broken. This is done in a ton of places, including Cluster.__enter__.

Here's another quick test that I would expect to pass.

def test_cluster_sync():
    class ExampleCluster(Cluster):
        async def _foo(self):
            pass

        def foo(self):
            return self.sync(self._foo)
        
        def __enter__(self):
            # Bypass failure here so that we can see foo() fail
            return self
        
    with ExampleCluster(name="S") as cluster:
        cluster.foo()

$ pytest distributed/deploy/tests/test_cluster.py::test_cluster_sync
==================================================================== test session starts =====================================================================
platform linux -- Python 3.10.11, pytest-7.3.2, pluggy-1.0.0 -- /home/jtomlinson/miniconda3/envs/dask/bin/python
cachedir: .pytest_cache
rootdir: /home/jtomlinson/Projects/dask/distributed
configfile: pyproject.toml
plugins: rerunfailures-11.1.2, timeout-2.1.0, kind-22.11.1, cov-4.1.0
timeout: 300.0s
timeout method: thread
timeout func_only: False
collected 1 item                                                                                                                                             

distributed/deploy/tests/test_cluster.py::test_cluster_sync FAILED                                                                                     [100%]

========================================================================== FAILURES ==========================================================================
_____________________________________________________________________ test_cluster_sync ______________________________________________________________________

    def test_cluster_sync():
        class ExampleCluster(Cluster):
            async def _foo(self):
                pass
    
            def foo(self):
                return self.sync(self._foo)
    
            def __enter__(self):
                # Bypass failure here so that we can see foo() fail
                return self
    
        with ExampleCluster(name="S") as cluster:
>           cluster.foo()

distributed/deploy/tests/test_cluster.py:110: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
distributed/deploy/tests/test_cluster.py:103: in foo
    return self.sync(self._foo)
distributed/utils.py:355: in sync
    self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
distributed/deploy/cluster.py:105: in loop
    self.__loop = loop = self._loop_runner.loop
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <distributed.utils.LoopRunner object at 0x7fdee4f836a0>

    @property
    def loop(self):
        loop = self._loop
        if loop is None or not loop.asyncio_loop.is_running():
>           raise RuntimeError(
                "Accessing the loop property while the loop is not running is not supported"
            )
E           RuntimeError: Accessing the loop property while the loop is not running is not supported

distributed/utils.py:647: RuntimeError
==================================================================== slowest 20 durations ====================================================================

(3 durations < 0.005s hidden.  Use -vv to show these durations.)
================================================================== short test summary info ===================================================================
FAILED distributed/deploy/tests/test_cluster.py::test_cluster_sync - RuntimeError: Accessing the loop property while the loop is not running is not supported
===================================================================== 1 failed in 0.18s ======================================================================

[graingert]

before this change (eg 2023.4.1) cluster.foo() would hang forever because the loop isn't running, subclasses of Cluster are required to start the loop_runner

Python 3.11.3 | packaged by conda-forge | (main, Apr  6 2023, 08:57:19) [GCC 11.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import distributed
>>> distributed.__version__
'2023.4.1+19.g2e249bca8'
>>> from distributed.deploy import Cluster
>>> class ExampleCluster(Cluster):
...     async def _foo(self):
...         pass
...     def foo(self):
...         return self.sync(self._foo())
... 
>>> with ExampleCluster() as cluster:
...     cluster.foo()
... 

[jacobtomlinson]

That makes sense. Ok I'm going to open a PR to make Cluster.asynchronous behave as @fjetter expected and also make it easier for subclasses to get things running. I'll ping you on it once it's up if you wouldn't mind reviewing it.

[jacobtomlinson]

@graingert it's always been kinda weird that Cluster is incomplete and needs extra stuff doing in subclasses in order to get it working. Perhaps we should explore moving things up from SpecCluster so we don't have to copy these things around other projects?

[graingert]

Perhaps we should explore moving things up from SpecCluster so we don't have to #7940 (comment) around other projects?

Yeah I'd like to see Cluster manage starting and stopping the loop_runner, but it's a bit of a chicken and egg problem because subclasses have been managing it instead.

Perhaps we could introduce a new subclass of Cluster between Cluster and SpecCluster that implements loop_runner start and stop

[jacobtomlinson]

Yeah I like that idea. Perhaps something like BaseCluster to show that it's the point that folks should inherit from?