[Tune] Add find_free_port Tune util

python/ray/tune/integration/tensorflow.py

@@ -7,10 +7,9 @@
 from ray.tune.result import RESULT_DUPLICATE
 from ray.tune.function_runner import wrap_function
 from ray.tune.resources import Resources
-from ray.util.sgd.utils import find_free_port
 from ray.util.placement_group import remove_placement_group
 from ray.tune.utils.trainable import PlacementGroupUtil, TrainableUtil
-from ray.tune.utils.util import detect_checkpoint_function
+from ray.tune.utils import detect_checkpoint_function, find_free_port
 from typing import Callable, Dict, Type, Optional
 
 logger = logging.getLogger(__name__)

python/ray/tune/utils/__init__.py

@@ -1,14 +1,14 @@
 from ray.tune.utils.util import (
- deep_update, date_str, flatten_dict, get_pinned_object, merge_dicts,
- pin_in_object_store, unflattened_lookup, UtilMonitor,
+ deep_update, date_str, find_free_port, flatten_dict, get_pinned_object,
+ merge_dicts, pin_in_object_store, unflattened_lookup, UtilMonitor,
  validate_save_restore, warn_if_slow, diagnose_serialization,
  detect_checkpoint_function, detect_reporter, detect_config_single,
  wait_for_gpu)
 
 __all__ = [
- "deep_update", "date_str", "flatten_dict", "get_pinned_object",
- "merge_dicts", "pin_in_object_store", "unflattened_lookup", "UtilMonitor",
- "validate_save_restore", "warn_if_slow", "diagnose_serialization",
- "detect_checkpoint_function", "detect_reporter", "detect_config_single",
- "wait_for_gpu"
+ "deep_update", "date_str", "find_free_port", "flatten_dict",
+ "get_pinned_object", "merge_dicts", "pin_in_object_store",
+ "unflattened_lookup", "UtilMonitor", "validate_save_restore",
+ "warn_if_slow", "diagnose_serialization", "detect_checkpoint_function",
+ "detect_reporter", "detect_config_single", "wait_for_gpu"
 ]

python/ray/tune/utils/util.py

@@ -1,3 +1,5 @@
+import socket
+from contextlib import closing
 from typing import Dict, List, Union
 import copy
 import json
@@ -484,6 +486,14 @@ def atomic_save(state: Dict, checkpoint_dir: str, file_name: str,
  os.replace(tmp_search_ckpt_path, os.path.join(checkpoint_dir, file_name))
 
 
+def find_free_port():
+ """Finds a free port on the current node."""
+ with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+ s.bind(("", 0))
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ return s.getsockname()[1]
+
+
 def load_newest_checkpoint(dirpath: str, ckpt_pattern: str) -> dict:
  """Returns the most recently modified checkpoint.