Remove python 3.6 support [2/n] (#34416)

Removed some dead code for 3.6

LICENSE

@@ -289,7 +289,7 @@ Code in python/ray/_private/prometheus_exporter.py is adapted from https://githu
 # limitations under the License.
 
 --------------------------------------------------------------------------------
-Code in python/ray/tests/modin/test_modin and 
+Code in python/ray/tests/modin/test_modin and
 python/ray/tests/modin/modin_test_utils adapted from:
 - http://github.com/modin-project/modin/master/modin/pandas/test/test_general.py
 - http://github.com/modin-project/modin/master/modin/pandas/test/utils.py
@@ -309,7 +309,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 
 --------------------------------------------------------------------------------
-Code in src/ray/util/logging.h is adapted from 
+Code in src/ray/util/logging.h is adapted from
 https://github.com/google/glog/blob/master/src/glog/logging.h.in
 
 Copyright (c) 2008, Google Inc.
@@ -342,7 +342,7 @@ THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 --------------------------------------------------------------------------------
-Code in python/ray/_private/runtime_env/conda_utils.py is adapted from 
+Code in python/ray/_private/runtime_env/conda_utils.py is adapted from
 https://github.com/mlflow/mlflow/blob/master/mlflow/utils/conda.py
 
 Copyright (c) 2018, Databricks, Inc.
@@ -384,24 +384,6 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
---------------------------------------------------------------------------------
-Code in python/ray/_private/async_compat.py is adapted from
-https://github.com/python-trio/async_generator/blob/master/async_generator/_util.py
-
-Copyright (c) 2022, Nathaniel J. Smith
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
 ---------------------------------------------------------------------------------------------------------------
 Code in python/ray/_private/thirdparty/tabulate/tabulate.py is adapted from https://github.com/astanin/python-tabulate/blob/4892c6e9a79638c7897ccea68b602040da9cc7a7/tabulate.py
 

dashboard/modules/runtime_env/runtime_env_agent.py

@@ -14,7 +14,6 @@
 import ray.dashboard.consts as dashboard_consts
 import ray.dashboard.modules.runtime_env.runtime_env_consts as runtime_env_consts
 import ray.dashboard.utils as dashboard_utils
-from ray._private.async_compat import create_task
 from ray._private.ray_logging import setup_component_logger
 from ray._private.runtime_env.conda import CondaPlugin
 from ray._private.runtime_env.container import ContainerManager
@@ -344,15 +343,10 @@ async def _create_runtime_env_with_retry(
             error_message = None
             for _ in range(runtime_env_consts.RUNTIME_ENV_RETRY_TIMES):
                 try:
-                    # python 3.6 requires the type of input is `Future`,
-                    # python 3.7+ only requires the type of input is `Awaitable`
-                    # TODO(Catch-Bull): remove create_task when ray drop python 3.6
-                    runtime_env_setup_task = create_task(
-                        _setup_runtime_env(
-                            runtime_env,
-                            serialized_env,
-                            request.serialized_allocated_resource_instances,
-                        )
+                    runtime_env_setup_task = _setup_runtime_env(
+                        runtime_env,
+                        serialized_env,
+                        request.serialized_allocated_resource_instances,
                     )
                     runtime_env_context = await asyncio.wait_for(
                         runtime_env_setup_task, timeout=setup_timeout_seconds

python/ray/_private/async_compat.py

@@ -11,20 +11,6 @@
     uvloop = None
 
 
-try:
-    # This function has been added in Python 3.7. Prior to Python 3.7,
-    # the low-level asyncio.ensure_future() function can be used instead.
-    from asyncio import create_task  # noqa: F401
-except ImportError:
-    from asyncio import ensure_future as create_task  # noqa: F401
-
-
-try:
-    from asyncio import get_running_loop  # noqa: F401
-except ImportError:
-    from asyncio import _get_running_loop as get_running_loop  # noqa: F401
-
-
 def get_new_event_loop():
     """Construct a new event loop. Ray will use uvloop if it exists"""
     if uvloop:
@@ -43,116 +29,3 @@ async def wrapper(*args, **kwargs):
         return func(*args, **kwargs)
 
     return wrapper
-
-
-try:
-    from contextlib import asynccontextmanager
-except ImportError:
-    # Copy from https://github.com/python-trio/async_generator
-    # for compatible with Python 3.6
-    import sys
-    from functools import wraps
-    from inspect import isasyncgenfunction
-
-    class _aclosing:
-        def __init__(self, aiter):
-            self._aiter = aiter
-
-        async def __aenter__(self):
-            return self._aiter
-
-        async def __aexit__(self, *args):
-            await self._aiter.aclose()
-
-    # Very much derived from the one in contextlib, by copy/pasting and then
-    # asyncifying everything. (Also I dropped the obscure support for using
-    # context managers as function decorators. It could be re-added; I just
-    # couldn't be bothered.)
-    # So this is a derivative work licensed under the PSF License, which requires
-    # the following notice:
-    #
-    # Copyright © 2001-2017 Python Software Foundation; All Rights Reserved
-    class _AsyncGeneratorContextManager:
-        def __init__(self, func, args, kwds):
-            self._func_name = func.__name__
-            self._agen = func(*args, **kwds).__aiter__()
-
-        async def __aenter__(self):
-            try:
-                return await self._agen.asend(None)
-            except StopAsyncIteration:
-                raise RuntimeError("async generator didn't yield") from None
-
-        async def __aexit__(self, type, value, traceback):
-            async with _aclosing(self._agen):
-                if type is None:
-                    try:
-                        await self._agen.asend(None)
-                    except StopAsyncIteration:
-                        return False
-                    else:
-                        raise RuntimeError("async generator didn't stop")
-                else:
-                    # It used to be possible to have type != None, value == None:
-                    #    https://bugs.python.org/issue1705170
-                    # but AFAICT this can't happen anymore.
-                    assert value is not None
-                    try:
-                        await self._agen.athrow(type, value, traceback)
-                        raise RuntimeError("async generator didn't stop after athrow()")
-                    except StopAsyncIteration as exc:
-                        # Suppress StopIteration *unless* it's the same exception
-                        # that was passed to throw(). This prevents a
-                        # StopIteration raised inside the "with" statement from
-                        # being suppressed.
-                        return exc is not value
-                    except RuntimeError as exc:
-                        # Don't re-raise the passed in exception. (issue27112)
-                        if exc is value:
-                            return False
-                        # Likewise, avoid suppressing if a StopIteration exception
-                        # was passed to throw() and later wrapped into a
-                        # RuntimeError (see PEP 479).
-                        if (
-                            isinstance(value, (StopIteration, StopAsyncIteration))
-                            and exc.__cause__ is value
-                        ):
-                            return False
-                        raise
-                    except:  # noqa: E722
-                        # only re-raise if it's *not* the exception that was
-                        # passed to throw(), because __exit__() must not raise an
-                        # exception unless __exit__() itself failed. But throw()
-                        # has to raise the exception to signal propagation, so
-                        # this fixes the impedance mismatch between the throw()
-                        # protocol and the __exit__() protocol.
-                        #
-                        if sys.exc_info()[1] is value:
-                            return False
-                        raise
-
-        def __enter__(self):
-            raise RuntimeError(
-                "use 'async with {func_name}(...)', not 'with {func_name}(...)'".format(
-                    func_name=self._func_name
-                )
-            )
-
-        def __exit__(self):  # pragma: no cover
-            assert False, """Never called, but should be defined"""
-
-    def asynccontextmanager(func):
-        """Like @contextmanager, but async."""
-        if not isasyncgenfunction(func):
-            raise TypeError(
-                "must be an async generator (native or from async_generator; "
-                "if using @async_generator then @acontextmanager must be on top."
-            )
-
-        @wraps(func)
-        def helper(*args, **kwds):
-            return _AsyncGeneratorContextManager(func, args, kwds)
-
-        # A hint for sphinxcontrib-trio:
-        helper.__returns_acontextmanager__ = True
-        return helper

python/ray/_private/function_manager.py

@@ -5,7 +5,6 @@
 import json
 import logging
 import os
-import sys
 import threading
 import time
 import traceback
@@ -128,10 +127,7 @@ def compute_collision_identifier(self, function_or_class):
         import io
 
         string_file = io.StringIO()
-        if sys.version_info[1] >= 7:
-            dis.dis(function_or_class, file=string_file, depth=2)
-        else:
-            dis.dis(function_or_class, file=string_file)
+        dis.dis(function_or_class, file=string_file, depth=2)
         collision_identifier = function_or_class.__name__ + ":" + string_file.getvalue()
 
         # Return a hash of the identifier in case it is too large.

python/ray/_private/runtime_env/pip.py

@@ -7,8 +7,9 @@
 import sys
 import tempfile
 from typing import Dict, List, Optional, Tuple
+from contextlib import asynccontextmanager
+from asyncio import create_task, get_running_loop
 
-from ray._private.async_compat import asynccontextmanager, create_task, get_running_loop
 from ray._private.runtime_env.context import RuntimeEnvContext
 from ray._private.runtime_env.packaging import Protocol, parse_uri
 from ray._private.runtime_env.plugin import RuntimeEnvPlugin

python/ray/_private/utils.py

@@ -1800,9 +1800,7 @@ def create_if_main_thread(cls) -> contextlib.AbstractContextManager:
         if threading.current_thread() == threading.main_thread():
             return cls()
         else:
-            # TODO(Clark): Use contextlib.nullcontext() once Python 3.6 support is
-            # dropped.
-            return contextlib.suppress()
+            return contextlib.nullcontext()
 
     def _set_task_cancelled(self, signum, frame):
         """SIGINT handler that defers the signal."""

python/ray/data/__init__.py

@@ -1,10 +1,6 @@
-import sys
-
 # Short term workaround for https://github.com/ray-project/ray/issues/32435
 # Datastream has a hard dependency on pandas, so it doesn't need to be delayed.
-# ray.data import is still eager for all ray imports for Python 3.6:
-if sys.version_info >= (3, 7):
-    import pandas  # noqa
+import pandas  # noqa
 
 from ray.data._internal.compute import ActorPoolStrategy
 from ray.data._internal.progress_bar import set_progress_bars

python/ray/data/_internal/block_batching/block_batching.py

@@ -1,7 +1,7 @@
 import collections
 import itertools
-import sys
 from typing import Any, Callable, Iterator, Optional, TypeVar, Union
+from contextlib import nullcontext
 
 import ray
 from ray.data._internal.block_batching.interfaces import BlockPrefetcher
@@ -22,15 +22,6 @@
 
 T = TypeVar("T")
 
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
-
 
 def batch_block_refs(
     block_refs: Iterator[ObjectRef[Block]],

python/ray/data/_internal/block_batching/iter_batches.py

@@ -1,5 +1,4 @@
 import collections
-import sys
 from typing import Any, Callable, Dict, Iterator, Optional, Tuple
 
 import ray
@@ -22,15 +21,7 @@
 from ray.data._internal.memory_tracing import trace_deallocation
 from ray.data._internal.stats import DatastreamStats
 from ray.data.context import DataContext
-
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
+from contextlib import nullcontext
 
 
 def iter_batches(

python/ray/data/_internal/block_batching/util.py

@@ -1,8 +1,8 @@
 import logging
 import queue
 import threading
-import sys
 from typing import Any, Callable, Iterator, List, Optional, Tuple, TypeVar, Union
+from contextlib import nullcontext
 
 import ray
 from ray.types import ObjectRef
@@ -22,15 +22,6 @@
 
 logger = logging.getLogger(__name__)
 
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
-
 
 def _calculate_ref_hits(refs: List[ObjectRef[Any]]) -> Tuple[int, int, int]:
     """Given a list of object references, returns how many are already on the local

python/ray/serve/benchmarks/handle.py

@@ -94,5 +94,4 @@ async def main():
                 await run_test(num_replicas, num_forwarders, sync)
 
 
-# TODO(rickyx): use asyncio.run after deprecating 3.6
-asyncio.new_event_loop().run_until_complete(main())
+asyncio.run(main())

python/ray/tests/modin/test_modin.py

@@ -24,16 +24,14 @@
 import ray
 from ray.tests.conftest import start_cluster  # noqa F401
 
-modin_compatible_version = sys.version_info >= (3, 7, 0)
 modin_installed = True
 
-if modin_compatible_version:
-    try:
-        import modin  # noqa: F401
-    except ModuleNotFoundError:
-        modin_installed = False
+try:
+    import modin  # noqa: F401
+except ModuleNotFoundError:
+    modin_installed = False
 
-skip = not modin_compatible_version or not modin_installed
+skip = not modin_installed
 
 # These tests are written for versions of Modin that require python 3.7+
 pytestmark = pytest.mark.skipif(skip, reason="Outdated or missing Modin dependency")

python/ray/tests/test_gcs_pubsub.py

@@ -142,9 +142,6 @@ async def test_aio_publish_and_subscribe_resource_usage(ray_start_regular):
     await subscriber.close()
 
 
-@pytest.mark.skipif(
-    sys.version_info < (3, 7, 0), reason="no asyncio.all_tasks in py3.6"
-)
 @pytest.mark.asyncio
 async def test_aio_poll_no_leaks(ray_start_regular):
     """Test that polling doesn't leak memory."""

python/ray/tests/test_output.py

@@ -362,7 +362,6 @@ def test_core_worker_error_message():
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows.")
-@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7+")
 def test_disable_driver_logs_breakpoint():
     script = """
 import time