<commit_before>from typing import Any, Callable, Dict, List, TYPE_CHECKING, Type, Union, Optional

import distutils
import logging
import os
import time
from dataclasses import dataclass
import warnings

from inspect import isclass
from shlex import quote

import ray
import yaml
from ray.tune import TuneError
from ray.tune.callback import Callback
from ray.tune.checkpoint_manager import Checkpoint
from ray.tune.result import NODE_IP
from ray.util import get_node_ip_address
from ray.util.debug import log_once
from ray.tune.cluster_info import get_ssh_key, get_ssh_user
from ray.tune.sync_client import (
    CommandBasedClient,
    get_sync_client,
    get_cloud_sync_client,
    NOOP,
    SyncClient,
)
from ray.util.annotations import PublicAPI

if TYPE_CHECKING:
    from ray.tune.trial import Trial

logger = logging.getLogger(__name__)

# Syncing period for syncing checkpoints between nodes or to cloud.
SYNC_PERIOD = 300

CLOUD_CHECKPOINTING_URL = (
    "https://docs.ray.io/en/master/tune/user-guide.html#using-cloud-storage"
)
_log_sync_warned = False
_syncers = {}


def wait_for_sync():
    for syncer in _syncers.values():
        syncer.wait()


def set_sync_periods(sync_config: "SyncConfig"):
    """Sets sync period from config."""
    global SYNC_PERIOD
    SYNC_PERIOD = int(sync_config.sync_period)


def validate_sync_config(sync_config: "SyncConfig"):
    if sync_config.node_sync_period >= 0 or sync_config.cloud_sync_period >= 0:
        # Until fully deprecated, try to consolidate
        if sync_config.node_sync_period >= 0 and sync_config.cloud_sync_period >= 0:
            sync_period = min(
                sync_config.node_sync_period, sync_config.cloud_sync_period
            )
        else:
            sync_period = max(
                sync_config.node_sync_period, sync_config.cloud_sync_period
            )

        sync_config.sync_period = sync_period
        sync_config.node_sync_period = -1
        sync_config.cloud_sync_period = -1

        warnings.warn(
            "The `node_sync_period` and "
            "`cloud_sync_period` properties of `tune.SyncConfig` are "
            "deprecated. Pass the `sync_period` property instead. "
            "\nFor now, the lower of the two values (if provided) will "
            f"be used as the sync_period. This value is: {sync_period}",
            DeprecationWarning,
        )

    if sync_config.sync_to_cloud or sync_config.sync_to_driver:
        if bool(sync_config.upload_dir):
            syncer = sync_config.sync_to_cloud
            help = "set"
        else:
            syncer = sync_config.sync_to_driver
            help = "not set"

        sync_config.syncer = syncer
        sync_config.sync_to_cloud = None
        sync_config.sync_to_driver = None

        warnings.warn(
            "The `sync_to_cloud` and `sync_to_driver` properties of "
            "`tune.SyncConfig` are deprecated. Pass the `syncer` property "
            "instead. Presence of an `upload_dir` decides if checkpoints "
            "are synced to cloud or not. Syncing to driver is "
            "automatically disabled if an `upload_dir` is given."
            f"\nFor now, as the upload dir is {help}, the respective "
            f"syncer is used. This value is: {syncer}",
            DeprecationWarning,
        )


def log_sync_template(options: str = ""):
    """Template enabling syncs between driver and worker when possible.
    Requires ray cluster to be started with the autoscaler. Also requires
    rsync to be installed.

    Args:
        options: Additional rsync options.

    Returns:
        Sync template with source and target parameters. None if rsync
        unavailable.
    """
    if not distutils.spawn.find_executable("rsync"):
        if log_once("tune:rsync"):
            logger.error("Log sync requires rsync to be installed.")
        return None
    global _log_sync_warned
    ssh_key = get_ssh_key()
    if ssh_key is None:
        if not _log_sync_warned:
            logger.debug("Log sync requires cluster to be setup with `ray up`.")
            _log_sync_warned = True
        return None

    rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no"
    rsh = rsh.format(ssh_key=quote(ssh_key))
    template = "rsync {options} -savz -e {rsh} {{source}} {{target}}"
    return template.format(options=options, rsh=quote(rsh))


@PublicAPI
@dataclass
class SyncConfig:<commit_msg>Don't rsync checkpoint_tmp directories. (#18434)<commit_after>from typing import Any, Callable, Dict, List, TYPE_CHECKING, Type, Union, Optional

import distutils
import logging
import os
import time
from dataclasses import dataclass
import warnings

from inspect import isclass
from shlex import quote

import ray
import yaml
from ray.tune import TuneError
from ray.tune.callback import Callback
from ray.tune.checkpoint_manager import Checkpoint
from ray.tune.result import NODE_IP
from ray.util import get_node_ip_address
from ray.util.debug import log_once
from ray.tune.cluster_info import get_ssh_key, get_ssh_user
from ray.tune.sync_client import (
    CommandBasedClient,
    get_sync_client,
    get_cloud_sync_client,
    NOOP,
    SyncClient,
)
from ray.util.annotations import PublicAPI

if TYPE_CHECKING:
    from ray.tune.trial import Trial

logger = logging.getLogger(__name__)

# Syncing period for syncing checkpoints between nodes or to cloud.
SYNC_PERIOD = 300

CLOUD_CHECKPOINTING_URL = (
    "https://docs.ray.io/en/master/tune/user-guide.html#using-cloud-storage"
)
_log_sync_warned = False
_syncers = {}


def wait_for_sync():
    for syncer in _syncers.values():
        syncer.wait()


def set_sync_periods(sync_config: "SyncConfig"):
    """Sets sync period from config."""
    global SYNC_PERIOD
    SYNC_PERIOD = int(sync_config.sync_period)


def validate_sync_config(sync_config: "SyncConfig"):
    if sync_config.node_sync_period >= 0 or sync_config.cloud_sync_period >= 0:
        # Until fully deprecated, try to consolidate
        if sync_config.node_sync_period >= 0 and sync_config.cloud_sync_period >= 0:
            sync_period = min(
                sync_config.node_sync_period, sync_config.cloud_sync_period
            )
        else:
            sync_period = max(
                sync_config.node_sync_period, sync_config.cloud_sync_period
            )

        sync_config.sync_period = sync_period
        sync_config.node_sync_period = -1
        sync_config.cloud_sync_period = -1

        warnings.warn(
            "The `node_sync_period` and "
            "`cloud_sync_period` properties of `tune.SyncConfig` are "
            "deprecated. Pass the `sync_period` property instead. "
            "\nFor now, the lower of the two values (if provided) will "
            f"be used as the sync_period. This value is: {sync_period}",
            DeprecationWarning,
        )

    if sync_config.sync_to_cloud or sync_config.sync_to_driver:
        if bool(sync_config.upload_dir):
            syncer = sync_config.sync_to_cloud
            help = "set"
        else:
            syncer = sync_config.sync_to_driver
            help = "not set"

        sync_config.syncer = syncer
        sync_config.sync_to_cloud = None
        sync_config.sync_to_driver = None

        warnings.warn(
            "The `sync_to_cloud` and `sync_to_driver` properties of "
            "`tune.SyncConfig` are deprecated. Pass the `syncer` property "
            "instead. Presence of an `upload_dir` decides if checkpoints "
            "are synced to cloud or not. Syncing to driver is "
            "automatically disabled if an `upload_dir` is given."
            f"\nFor now, as the upload dir is {help}, the respective "
            f"syncer is used. This value is: {syncer}",
            DeprecationWarning,
        )


def log_sync_template(options: str = ""):
    """Template enabling syncs between driver and worker when possible.
    Requires ray cluster to be started with the autoscaler. Also requires
    rsync to be installed.

    Args:
        options: Additional rsync options.

    Returns:
        Sync template with source and target parameters. None if rsync
        unavailable.
    """
    if not distutils.spawn.find_executable("rsync"):
        if log_once("tune:rsync"):
            logger.error("Log sync requires rsync to be installed.")
        return None
    global _log_sync_warned
    ssh_key = get_ssh_key()
    if ssh_key is None:
        if not _log_sync_warned:
            logger.debug("Log sync requires cluster to be setup with `ray up`.")
            _log_sync_warned = True
        return None

    rsh = "ssh -i {ssh_key} -o ConnectTimeout=120s -o StrictHostKeyChecking=no"
    rsh = rsh.format(ssh_key=quote(ssh_key))
    options += " --exclude='checkpoint_tmp*'"
    template = "rsync {options} -savz -e {rsh} {{source}} {{target}}"
    return template.format(options=options, rsh=quote(rsh))


@PublicAPI
@dataclass
class SyncConfig:
{'fraction_non_alphanumeric': 0.07771441350777145, 'fraction_numerical': 0.003242759700324276, 'mean_word_length': 2.7994902293967714, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '782205', 'n_tokens_mistral': 2777, 'n_tokens_neox': 2687, 'n_words': 767}