WINDOWS: enable and fix failures in test_runtime_env_complicated (#22449)

.buildkite/pipeline.windows.yml

@@ -29,6 +29,8 @@ steps:
   parallelism: 6
   commands:
     - *prelude_commands
+    # conda init should be moved to the docker image setup
+    - conda init
     - . ./ci/travis/ci.sh init
     - . ./ci/travis/ci.sh build
     - if [ "${BUILDKITE_PARALLEL_JOB}" = "0" ]; then . ./ci/travis/ci.sh test_core; fi

ci/travis/ci.sh

@@ -156,7 +156,6 @@ test_python() {
       -python/ray/tests:test_multi_node_3
       -python/ray/tests:test_object_manager # OOM on test_object_directory_basic
       -python/ray/tests:test_resource_demand_scheduler
-      -python/ray/tests:test_runtime_env_complicated # requires conda
       -python/ray/tests:test_stress  # timeout
       -python/ray/tests:test_stress_sharded  # timeout
       -python/ray/tests:test_k8s_operator_unit_tests
@@ -180,6 +179,10 @@ test_python() {
     bazel test --config=ci \
       --build_tests_only $(./scripts/bazel_export_options) \
       --test_env=PYTHONPATH="${PYTHONPATH-}${pathsep}${WORKSPACE_DIR}/python/ray/pickle5_files" \
+      --test_env=USERPROFILE="${USERPROFILE}" \
+      --test_env=CI=1 \
+      --test_env=RAY_CI_POST_WHEEL_TESTS=1 \
+      --test_output=streamed \
       -- \
       ${test_shard_selection};
   fi

dashboard/modules/reporter/reporter_agent.py

@@ -275,7 +275,10 @@ class ReporterAgent(
             return {
                 "/": psutil._common.sdiskusage(total=1, used=0, free=1, percent=0.0)
             }
-        root = os.environ["USERPROFILE"] if sys.platform == "win32" else os.sep
+        if sys.platform == "win32":
+            root = psutil.disk_partitions()[0].mountpoint
+        else:
+            root = os.sep
         tmp = ray._private.utils.get_user_temp_dir()
         return {
             "/": psutil.disk_usage(root),

doc/source/ray-core/handling-dependencies.rst

@@ -52,7 +52,7 @@ Runtime environments
 
 .. note::
 
-    This feature requires a full installation of Ray using ``pip install "ray[default]"``. This feature is available starting with Ray 1.4.0 and is currently only supported on macOS and Linux.
+    This feature requires a full installation of Ray using ``pip install "ray[default]"``. This feature is available starting with Ray 1.4.0 and is currently supported on macOS and Linux. It is experimentally supported on Windows.
 
 The second way to set up dependencies is to install them dynamically while Ray is running.
 

python/ray/_private/runtime_env/_clonevirtualenv.py

@@ -21,6 +21,9 @@ logger = logging.getLogger()
 env_bin_dir = "bin"
 if sys.platform == "win32":
     env_bin_dir = "Scripts"
+    _WIN32 = True
+else:
+    _WIN32 = False
 
 
 class UserError(Exception):
@@ -49,6 +52,10 @@ def _dirmatch(path, matchwith):
 def _virtualenv_sys(venv_path):
     """obtain version and path info from a virtualenv."""
     executable = os.path.join(venv_path, env_bin_dir, "python")
+    if _WIN32:
+        env = os.environ.copy()
+    else:
+        env = {}
     # Must use "executable" as the first argument rather than as the
     # keyword argument "executable" to get correct value from sys.path
     p = subprocess.Popen(
@@ -59,7 +66,7 @@ def _virtualenv_sys(venv_path):
             'print ("%d.%d" % (sys.version_info.major, sys.version_info.minor));'
             'print ("\\n".join(sys.path));',
         ],
-        env={},
+        env=env,
         stdout=subprocess.PIPE,
     )
     stdout, err = p.communicate()

python/ray/_private/runtime_env/conda.py

@@ -33,6 +33,8 @@ from ray._private.runtime_env.packaging import Protocol, parse_uri
 
 default_logger = logging.getLogger(__name__)
 
+_WIN32 = os.name == "nt"
+
 
 def _resolve_current_ray_path() -> str:
     # When ray is built from source with pip install -e,
@@ -63,7 +65,10 @@ def _inject_ray_to_conda_site(
     conda_path, logger: Optional[logging.Logger] = default_logger
 ):
     """Write the current Ray site package directory to a new site"""
-    python_binary = os.path.join(conda_path, "bin/python")
+    if _WIN32:
+        python_binary = os.path.join(conda_path, "python")
+    else:
+        python_binary = os.path.join(conda_path, "bin/python")
     site_packages_path = (
         subprocess.check_output(
             [python_binary, "-c", "import site; print(site.getsitepackages()[0])"]

python/ray/_private/runtime_env/conda_utils.py

@@ -12,13 +12,15 @@ from typing import Optional, List, Union, Tuple
 # will default to running "conda" if unset.
 RAY_CONDA_HOME = "RAY_CONDA_HOME"
 
+_WIN32 = os.name == "nt"
+
 
 def get_conda_activate_commands(conda_env_name: str) -> List[str]:
     """
     Get a list of commands to run to silently activate the given conda env.
     """
     #  Checking for newer conda versions
-    if os.name != "nt" and ("CONDA_EXE" in os.environ or RAY_CONDA_HOME in os.environ):
+    if not _WIN32 and ("CONDA_EXE" in os.environ or RAY_CONDA_HOME in os.environ):
         conda_path = get_conda_bin_executable("conda")
         activate_conda_env = [
             ". {}/../etc/profile.d/conda.sh".format(os.path.dirname(conda_path))
@@ -27,9 +29,8 @@ def get_conda_activate_commands(conda_env_name: str) -> List[str]:
 
     else:
         activate_path = get_conda_bin_executable("activate")
-        # in case os name is not 'nt', we are not running on windows. Introduce
+        if not _WIN32:
-        # bash command otherwise.
+            # Use bash command syntax
-        if os.name != "nt":
             return ["source %s %s 1>&2" % (activate_path, conda_env_name)]
         else:
             return ["conda activate %s" % (conda_env_name)]
@@ -39,20 +40,41 @@ def get_conda_activate_commands(conda_env_name: str) -> List[str]:
 def get_conda_bin_executable(executable_name: str) -> str:
     """
     Return path to the specified executable, assumed to be discoverable within
-    the 'bin' subdirectory of a conda installation.
+    a conda installation.
 
-    The conda home directory (expected to contain a 'bin' subdirectory) is
+    The conda home directory (expected to contain a 'bin' subdirectory on
-    configurable via the ``RAY_CONDA_HOME`` environment variable. If
+    linux) is configurable via the ``RAY_CONDA_HOME`` environment variable. If
-    ``RAY_CONDA_HOME`` is unspecified, this method simply returns the passed-in
+    ``RAY_CONDA_HOME`` is unspecified, try the ``CONDA_EXE`` environment
-    executable name.
+    variable set by activating conda. If neither is specified, this method
+    returns `executable_name`.
     """
     conda_home = os.environ.get(RAY_CONDA_HOME)
     if conda_home:
-        return os.path.join(conda_home, "bin/%s" % executable_name)
+        if _WIN32:
+            candidate = os.path.join(conda_home, "%s.exe" % executable_name)
+            if os.path.exists(candidate):
+                return candidate
+            candidate = os.path.join(conda_home, "%s.bat" % executable_name)
+            if os.path.exists(candidate):
+                return candidate
+        else:
+            return os.path.join(conda_home, "bin/%s" % executable_name)
+    else:
+        conda_home = "."
     # Use CONDA_EXE as per https://github.com/conda/conda/issues/7126
     if "CONDA_EXE" in os.environ:
         conda_bin_dir = os.path.dirname(os.environ["CONDA_EXE"])
-        return os.path.join(conda_bin_dir, executable_name)
+        if _WIN32:
+            candidate = os.path.join(conda_home, "%s.exe" % executable_name)
+            if os.path.exists(candidate):
+                return candidate
+            candidate = os.path.join(conda_home, "%s.bat" % executable_name)
+            if os.path.exists(candidate):
+                return candidate
+        else:
+            return os.path.join(conda_bin_dir, executable_name)
+    if _WIN32:
+        return executable_name + ".bat"
     return executable_name
 
 
@@ -79,9 +101,9 @@ def create_conda_env_if_needed(
     conda_path = get_conda_bin_executable("conda")
     try:
         exec_cmd([conda_path, "--help"], throw_on_error=False)
-    except EnvironmentError:
+    except (EnvironmentError, FileNotFoundError):
         raise ValueError(
-            f"Could not find Conda executable at {conda_path}. "
+            f"Could not find Conda executable at '{conda_path}'. "
             "Ensure Conda is installed as per the instructions at "
             "https://conda.io/projects/conda/en/latest/"
             "user-guide/install/index.html. "
@@ -192,6 +214,8 @@ def exec_cmd_stream_to_logger(
 
     The last n_lines lines of output are also returned (stdout and stderr).
     """
+    if "env" in kwargs and _WIN32 and "PATH" not in [x.upper() for x in kwargs.keys]:
+        raise ValueError("On windows, Popen requires 'PATH' in 'env'")
     child = subprocess.Popen(
         cmd,
         universal_newlines=True,

python/ray/_private/runtime_env/pip.py

@@ -18,6 +18,8 @@ from ray._private.utils import (
 
 default_logger = logging.getLogger(__name__)
 
+_WIN32 = os.name == "nt"
+
 
 def _get_pip_hash(pip_dict: Dict) -> str:
     serialized_pip_spec = json.dumps(pip_dict, sort_keys=True)
@@ -51,7 +53,10 @@ class _PathHelper:
     @classmethod
     def get_virtualenv_python(cls, target_dir: str) -> str:
         virtualenv_path = cls.get_virtualenv_path(target_dir)
-        return os.path.join(virtualenv_path, "bin/python")
+        if _WIN32:
+            return os.path.join(virtualenv_path, "Scripts", "python.exe")
+        else:
+            return os.path.join(virtualenv_path, "bin", "python")
 
     @classmethod
     def get_virtualenv_activate_command(cls, target_dir: str) -> str:
@@ -167,8 +172,12 @@ class PipProcessor:
                 "-c",
                 "import ray; print(ray.__version__, ray.__path__[0])",
             ]
+            if _WIN32:
+                env = os.environ.copy()
+            else:
+                env = {}
             output = await check_output_cmd(
-                check_ray_cmd, logger=logger, cwd=cwd, env={}
+                check_ray_cmd, logger=logger, cwd=cwd, env=env
             )
             # print after import ray may have  endings, so we strip them by *_
             ray_version, ray_path, *_ = [s.strip() for s in output.split()]
@@ -196,9 +205,15 @@ class PipProcessor:
         python = sys.executable
         virtualenv_path = os.path.join(path, "virtualenv")
         virtualenv_app_data_path = os.path.join(path, "virtualenv_app_data")
-        current_python_dir = os.path.abspath(
+
-            os.path.join(os.path.dirname(python), "..")
+        if _WIN32:
-        )
+            current_python_dir = sys.prefix
+            env = os.environ.copy()
+        else:
+            current_python_dir = os.path.abspath(
+                os.path.join(os.path.dirname(python), "..")
+            )
+            env = {}
 
         if cls._is_in_virtualenv():
             # virtualenv-clone homepage:
@@ -251,9 +266,9 @@ class PipProcessor:
             logger.info(
                 "Creating virtualenv at %s, current python dir %s",
                 virtualenv_path,
-                current_python_dir,
+                virtualenv_path,
             )
-        await check_output_cmd(create_venv_cmd, logger=logger, cwd=cwd, env={})
+        await check_output_cmd(create_venv_cmd, logger=logger, cwd=cwd, env=env)
 
     @classmethod
     async def _install_pip_packages(

python/ray/_private/runtime_env/validation.py

@@ -73,14 +73,15 @@ def parse_and_validate_conda(conda: Union[str, dict]) -> Union[str, dict]:
     """
     assert conda is not None
 
-    result = None
     if sys.platform == "win32":
-        raise NotImplementedError(
+        logger.warning(
-            "The 'conda' field in runtime_env "
+            "runtime environment support is experimental on Windows. "
-            "is not currently supported on "
+            "If you run into issues please file a report at "
-            "Windows."
+            "https://github.com/ray-project/ray/issues."
         )
-    elif isinstance(conda, str):
+
+    result = None
+    if isinstance(conda, str):
         yaml_file = Path(conda)
         if yaml_file.suffix in (".yaml", ".yml"):
             if not yaml_file.is_file():
@@ -132,12 +133,12 @@ def parse_and_validate_pip(pip: Union[str, List[str], Dict]) -> Optional[Dict]:
 
     result = None
     if sys.platform == "win32":
-        raise NotImplementedError(
+        logger.warning(
-            "The 'pip' field in runtime_env "
+            "runtime environment support is experimental on Windows. "
-            "is not currently supported on "
+            "If you run into issues please file a report at "
-            "Windows."
+            "https://github.com/ray-project/ray/issues."
         )
-    elif isinstance(pip, str):
+    if isinstance(pip, str):
         # We have been given a path to a requirements.txt file.
         pip_list = _handle_local_pip_requirement_file(pip)
         result = dict(packages=pip_list, pip_check=False)

python/ray/tests/BUILD

@@ -65,6 +65,7 @@ py_test_module_list(
 py_test_module_list(
   files = [
     "test_client.py",
+    "test_client_reconnect.py",
   ],
   size = "large",
   extra_srcs = SRCS,
@@ -82,7 +83,6 @@ py_test_module_list(
     "test_client_references.py",
     "test_client_warnings.py",
     "test_client_library_integration.py",
-    "test_client_reconnect.py",
   ],
   size = "medium",
   extra_srcs = SRCS,

python/ray/tests/conftest.py

@@ -324,10 +324,10 @@ def call_ray_stop_only():
 # Used to test both Ray Client and non-Ray Client codepaths.
 # Usage: In your test, call `ray.init(address)`.
 @pytest.fixture(scope="function", params=["ray_client", "no_ray_client"])
-def start_cluster(ray_start_cluster, request):
+def start_cluster(ray_start_cluster_enabled, request):
     assert request.param in {"ray_client", "no_ray_client"}
     use_ray_client: bool = request.param == "ray_client"
-    cluster = ray_start_cluster
+    cluster = ray_start_cluster_enabled
     cluster.add_node(num_cpus=4)
     if use_ray_client:
         cluster.head_node._ray_params.ray_client_server_port = "10004"

python/ray/tests/test_runtime_env.py

@@ -145,9 +145,6 @@ def test_container_option_serialize(runtime_env_class):
     assert job_config_serialized.count(b"--name=test") == 1
 
 
-@pytest.mark.skipif(
-    sys.platform == "win32", reason="runtime_env unsupported on Windows."
-)
 @pytest.mark.parametrize("runtime_env_class", [dict, RuntimeEnv])
 def test_no_spurious_worker_startup(shutdown_only, runtime_env_class):
     """Test that no extra workers start up during a long env installation."""
@@ -233,7 +230,7 @@ def test_runtime_env_no_spurious_resource_deadlock_msg(
     assert len(errors) == 0
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="pip not supported on Windows.")
+@pytest.mark.skipif(sys.platform == "win32", reason="Hangs on windows.")
 @pytest.mark.parametrize("runtime_env_class", [dict, RuntimeEnv])
 def test_failed_job_env_no_hang(shutdown_only, runtime_env_class):
     """Test that after a failed job-level env, tasks can still be run."""

python/ray/tests/test_runtime_env_complicated.py

@@ -6,7 +6,7 @@ import sys
 import tempfile
 import time
 from typing import List
-from unittest import mock, skipIf
+from unittest import mock
 import yaml
 
 import ray
@@ -38,9 +38,11 @@ if not os.environ.get("CI"):
 
 REQUEST_VERSIONS = ["2.2.0", "2.3.0"]
 
+_WIN32 = os.name == "nt"
+
 
 @pytest.fixture(scope="session")
-def conda_envs():
+def conda_envs(tmp_path_factory):
     """Creates two conda env with different requests versions."""
     conda_path = get_conda_bin_executable("conda")
     init_cmd = f". {os.path.dirname(conda_path)}" f"/../etc/profile.d/conda.sh"
@@ -50,34 +52,58 @@ def conda_envs():
 
     def create_package_env(env_name, package_version: str):
         delete_env(env_name)
-        subprocess.run(
+        proc = subprocess.run(
-            ["conda", "create", "-n", env_name, "-y", f"python={_current_py_version()}"]
+            [
+                "conda",
+                "create",
+                "-n",
+                env_name,
+                "-y",
+                f"python={_current_py_version()}",
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
         )
+        if proc.returncode != 0:
+            print("conda create failed, returned %d" % proc.returncode)
+            print(proc.stdout.decode())
+            print(proc.stderr.decode())
+            assert False
 
         _inject_ray_to_conda_site(get_conda_env_dir(env_name))
         ray_deps: List[str] = _resolve_install_from_source_ray_dependencies()
         ray_deps.append(f"requests=={package_version}")
-        with tempfile.NamedTemporaryFile("w") as f:
-            f.writelines([line + "\n" for line in ray_deps])
-            f.flush()
 
-            commands = [
+        reqs = tmp_path_factory.mktemp("reqs") / "requirements.txt"
-                init_cmd,
+        with reqs.open("wt") as fid:
-                f"conda activate {env_name}",
+            for line in ray_deps:
-                f"python -m pip install -r {f.name}",
+                fid.write(line)
-                "conda deactivate",
+                fid.write("\n")
-            ]
+
-            proc = subprocess.run(
+        commands = [
-                [" && ".join(commands)],
+            f"conda activate {env_name}",
-                shell=True,
+            f"python -m pip install -r {str(reqs)}",
-                stdout=subprocess.PIPE,
+            "conda deactivate",
-                stderr=subprocess.PIPE,
+        ]
-            )
+        if _WIN32:
-            if proc.returncode != 0:
+            # as a string
-                print("pip install failed")
+            command = " && ".join(commands)
-                print(proc.stdout.decode())
+        else:
-                print(proc.stderr.decode())
+            commands.insert(0, init_cmd)
-                assert False
+            # as a list
+            command = [" && ".join(commands)]
+        proc = subprocess.run(
+            command,
+            shell=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        if proc.returncode != 0:
+            print("conda/pip install failed, returned %d" % proc.returncode)
+            print("command", command)
+            print(proc.stdout.decode())
+            print(proc.stderr.decode())
+            assert False
 
     for package_version in REQUEST_VERSIONS:
         create_package_env(
@@ -279,8 +305,13 @@ def test_get_conda_env_dir(tmp_path):
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
 )
+@pytest.mark.skipif(
+    os.environ.get("CONDA_EXE") is None,
+    reason="Requires properly set-up conda shell",
+)
 def test_conda_create_task(shutdown_only):
-    """Tests dynamic creation of a conda env in a task's runtime env."""
+    """Tests dynamic creation of a conda env in a task's runtime env. Assumes
+    `conda init` has been successfully called."""
     ray.init()
     runtime_env = {
         "conda": {"dependencies": ["pip", {"pip": ["pip-install-test==0.5"]}]}
@@ -305,6 +336,10 @@ def test_conda_create_task(shutdown_only):
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
 )
+@pytest.mark.skipif(
+    os.environ.get("CONDA_EXE") is None,
+    reason="Requires properly set-up conda shell",
+)
 def test_conda_create_job_config(shutdown_only):
     """Tests dynamic conda env creation in a runtime env in the JobConfig."""
 
@@ -499,7 +534,10 @@ def test_pip_job_config(shutdown_only, pip_as_str, tmp_path):
     assert ray.get(f.remote())
 
 
-@skipIf(sys.platform == "win32", "Fail to create temp dir.")
+@pytest.mark.skipif(
+    os.environ.get("CI") and sys.platform == "win32",
+    reason="dirname(__file__) returns an invalid path",
+)
 def test_experimental_package(shutdown_only):
     ray.init(num_cpus=2)
     pkg = ray.experimental.load_package(
@@ -513,7 +551,10 @@ def test_experimental_package(shutdown_only):
     assert ray.get(pkg.my_func.remote()) == "hello world"
 
 
-@skipIf(sys.platform == "win32", "Fail to create temp dir.")
+@pytest.mark.skipif(
+    os.environ.get("CI") and sys.platform == "win32",
+    reason="dirname(__file__) returns an invalid path",
+)
 def test_experimental_package_lazy(shutdown_only):
     pkg = ray.experimental.load_package(
         os.path.join(
@@ -527,7 +568,7 @@ def test_experimental_package_lazy(shutdown_only):
     assert ray.get(pkg.my_func.remote()) == "hello world"
 
 
-@skipIf(sys.platform == "win32", "Fail to create temp dir.")
+@pytest.mark.skipif(_WIN32, reason="requires tar cli command")
 def test_experimental_package_github(shutdown_only):
     ray.init(num_cpus=2)
     pkg = ray.experimental.load_package(
@@ -539,6 +580,7 @@ def test_experimental_package_github(shutdown_only):
     assert ray.get(pkg.my_func.remote()) == "hello world"
 
 
+@pytest.mark.skipif(_WIN32, reason="Fails on windows")
 @pytest.mark.skipif(
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
@@ -587,6 +629,7 @@ def test_client_working_dir_filepath(call_ray_start, tmp_path):
             assert ray.get(f.remote())
 
 
+@pytest.mark.skipif(_WIN32, reason="Hangs on windows")
 @pytest.mark.skipif(
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
@@ -741,6 +784,7 @@ def test_simultaneous_install(shutdown_only):
 CLIENT_SERVER_PORT = 24001
 
 
+@pytest.mark.skipif(_WIN32, reason="Fails on windows")
 @pytest.mark.skipif(
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
@@ -883,6 +927,7 @@ def test_e2e_complex(call_ray_start, tmp_path):
         assert ray.get(a.test.remote()) == "Hello"
 
 
+@pytest.mark.skipif(_WIN32, reason="Fails on windows")
 @pytest.mark.skipif(
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",
@@ -940,6 +985,7 @@ def test_runtime_env_override(call_ray_start):
         ray.shutdown()
 
 
+@pytest.mark.skipif(_WIN32, reason="RecursionError on windows")
 @pytest.mark.skipif(
     os.environ.get("CI") and sys.platform != "linux",
     reason="This test is only run on linux CI machines.",

python/ray/tests/test_runtime_env_conda_and_pip_2.py

@@ -12,6 +12,7 @@ if not os.environ.get("CI"):
     os.environ["RAY_RUNTIME_ENV_LOCAL_DEV_MODE"] = "1"
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Flaky on windows")
 @pytest.mark.parametrize("field", ["conda", "pip"])
 @pytest.mark.parametrize("specify_env_in_init", [True, False])
 @pytest.mark.parametrize("spec_format", ["file", "python_object"])

src/ray/common/ray_config_def.h

@@ -349,7 +349,12 @@ RAY_CONFIG(int64_t, task_rpc_inlined_bytes_limit, 10 * 1024 * 1024)
 RAY_CONFIG(uint64_t, max_pending_lease_requests_per_scheduling_category, 10)
 
 /// Wait timeout for dashboard agent register.
+#ifdef _WIN32
+// agent startup time can involve creating conda environments
+RAY_CONFIG(uint32_t, agent_register_timeout_ms, 100 * 1000)
+#else
 RAY_CONFIG(uint32_t, agent_register_timeout_ms, 30 * 1000)
+#endif
 
 /// If the agent manager fails to communicate with the dashboard agent, we will retry
 /// after this interval.