Remove all torch packages from the nightly test requirements (#2157)

Co-authored-by: Justin Zhao <[email protected]>

.github/workflows/pytest.yml

@@ -92,7 +92,7 @@ jobs:
           fi
 
           if [ "$PYTORCH" == "nightly" ]; then
-            cat requirements.txt | sed '/^torch[>=<]/d' > requirements-temp && mv requirements-temp requirements.txt
+            cat requirements.txt | sed '/^torch[>=<]/d' | sed '/^torchtext[>=<]/d' | sed '/^torchvision[>=<]/d' | sed '/^torchaudio[>=<]/d' > requirements-temp && mv requirements-temp requirements.txt
             extra_index_url=https://download.pytorch.org/whl/nightly/cpu
             pip install --pre torch torchtext torchvision torchaudio --extra-index-url $extra_index_url
           else

ludwig/hyperopt/execution.py

@@ -10,6 +10,7 @@
 import traceback
 import uuid
 from abc import ABC, abstractmethod
+from distutils.version import LooseVersion
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple, Union
 
@@ -36,7 +37,13 @@
     from ray.tune import register_trainable, Stopper
     from ray.tune.suggest import BasicVariantGenerator, ConcurrencyLimiter, SEARCH_ALG_IMPORT
     from ray.tune.sync_client import CommandBasedClient
-    from ray.tune.syncer import get_cloud_sync_client
+
+    _ray_114 = LooseVersion(ray.__version__) >= LooseVersion("1.14")
+    if _ray_114:
+        from ray.tune.syncer import get_node_to_storage_syncer, SyncConfig
+    else:
+        from ray.tune.syncer import get_cloud_sync_client
+
     from ray.tune.utils import wait_for_gpu
     from ray.tune.utils.placement_groups import PlacementGroupFactory
     from ray.util.queue import Queue as RayQueue
@@ -266,7 +273,13 @@ def _get_sync_client_and_remote_checkpoint_dir(self, trial_dir: Path) -> Optiona
         remote_checkpoint_dir = os.path.join(
             self.sync_config.upload_dir, *_get_relative_checkpoints_dir_parts(trial_dir)
         )
-        return get_cloud_sync_client(remote_checkpoint_dir), remote_checkpoint_dir
+
+        if _ray_114:
+            syncer = get_node_to_storage_syncer(SyncConfig(upload_dir=remote_checkpoint_dir))
+        else:
+            syncer = get_cloud_sync_client(remote_checkpoint_dir)
+
+        return syncer, remote_checkpoint_dir
 
     # For specified [stopped] trial, remove checkpoint marker on any partial checkpoints
     @staticmethod
@@ -288,7 +301,7 @@ def _get_best_model_path(self, trial_path, analysis):
         if sync_info is not None:
             sync_client, remote_checkpoint_dir = sync_info
             sync_client.sync_down(remote_checkpoint_dir, trial_path)
-            sync_client.wait()
+            sync_client.wait_or_retry()
         self._remove_partial_checkpoints(trial_path)  # needed by get_best_checkpoint
         mod_path = None
         try:

tests/integration_tests/test_hyperopt_ray_horovod.py

@@ -17,6 +17,7 @@
 import os.path
 import shutil
 import uuid
+from distutils.version import LooseVersion
 from unittest.mock import patch
 
 import pytest
@@ -31,7 +32,12 @@
 
 try:
     import ray
-    from ray.tune.sync_client import get_sync_client
+
+    _ray_114 = LooseVersion(ray.__version__) >= LooseVersion("1.14")
+    if _ray_114:
+        from ray.tune.syncer import get_node_to_storage_syncer, SyncConfig
+    else:
+        from ray.tune.syncer import get_sync_client
 
     from ludwig.backend.ray import RayBackend
     from ludwig.hyperopt.execution import _get_relative_checkpoints_dir_parts, RayTuneExecutor
@@ -55,10 +61,12 @@
 
 def mock_storage_client(path):
     """Mocks storage client that treats a local dir as durable storage."""
-    client = get_sync_client(LOCAL_SYNC_TEMPLATE, LOCAL_DELETE_TEMPLATE)
     os.makedirs(path, exist_ok=True)
-    client.set_logdir(path)
-    return client
+    if _ray_114:
+        syncer = get_node_to_storage_syncer(SyncConfig(upload_dir=path))
+    else:
+        syncer = get_sync_client(LOCAL_SYNC_TEMPLATE, LOCAL_DELETE_TEMPLATE)
+    return syncer
 
 
 HYPEROPT_CONFIG = {