<commit_before>import base64
import collections
import errno
import io
import json
import logging
import multiprocessing
import os
import random
import shutil
import signal
import socket
import subprocess
import sys
import time

import colorama
import psutil
# Ray modules
import ray
import ray.ray_constants as ray_constants
import redis

resource = None
if sys.platform != "win32":
    import resource

EXE_SUFFIX = ".exe" if sys.platform == "win32" else ""

# True if processes are run in the valgrind profiler.
RUN_RAYLET_PROFILER = False
RUN_PLASMA_STORE_PROFILER = False

# Location of the redis server and module.
RAY_HOME = os.path.join(os.path.dirname(os.path.dirname(__file__)), "../..")
RAY_PATH = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
REDIS_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/thirdparty/redis/src/redis-server" + EXE_SUFFIX)
REDIS_MODULE = os.path.join(
    RAY_PATH, "core/src/ray/gcs/redis_module/libray_redis_module.so")

# Location of the plasma object store executable.
PLASMA_STORE_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/plasma/plasma_store_server" + EXE_SUFFIX)

# Location of the raylet executables.
RAYLET_EXECUTABLE = os.path.join(RAY_PATH,
                                 "core/src/ray/raylet/raylet" + EXE_SUFFIX)
GCS_SERVER_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/gcs/gcs_server" + EXE_SUFFIX)

# Location of the cpp default worker executables.
DEFAULT_WORKER_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/cpp/default_worker" + EXE_SUFFIX)

# Logger for this module. It should be configured at the entry point
# into the program using Ray. Ray provides a default configuration at
# entry/init points.
logger = logging.getLogger(__name__)

ProcessInfo = collections.namedtuple("ProcessInfo", [
    "process",
    "stdout_file",
    "stderr_file",
    "use_valgrind",
    "use_gdb",
    "use_valgrind_profiler",
    "use_perftools_profiler",
    "use_tmux",
])


def serialize_config(config):
    config_pairs = []
    for key, value in config.items():
        if isinstance(value, str):
            value = value.encode("utf-8")
        if isinstance(value, bytes):
            value = base64.b64encode(value).decode("utf-8")
        config_pairs.append((key, value))
    config_str = ";".join(["{},{}".format(*kv) for kv in config_pairs])
    assert " " not in config_str, (
        "Config parameters currently do not support "
        "spaces:", config_str)
    return config_str


class ConsolePopen(subprocess.Popen):
    if sys.platform == "win32":

        def terminate(self):
            if isinstance(self.stdin, io.IOBase):
                self.stdin.close()
            if self._use_signals:
                self.send_signal(signal.CTRL_BREAK_EVENT)
            else:
                super(ConsolePopen, self).terminate()

        def __init__(self, *args, **kwargs):
            # CREATE_NEW_PROCESS_GROUP is used to send Ctrl+C on Windows:
            # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
            new_pgroup = subprocess.CREATE_NEW_PROCESS_GROUP
            flags_to_add = 0
            if ray.utils.detect_fate_sharing_support():
                # If we don't have kernel-mode fate-sharing, then don't do this
                # because our children need to be in out process group for
                # the process reaper to properly terminate them.
                flags_to_add = new_pgroup
            flags_key = "creationflags"
            if flags_to_add:
                kwargs[flags_key] = (kwargs.get(flags_key) or 0) | flags_to_add
            self._use_signals = (kwargs[flags_key] & new_pgroup)
            super(ConsolePopen, self).__init__(*args, **kwargs)


def address(ip_address, port):
    return ip_address + ":" + str(port)


def new_port():
    return random.randint(10000, 65535)


def find_redis_address(address=None):
    """
    Currently, this extracts the deprecated --redis-address from the command
    that launched the raylet running on this node, if any. Anyone looking to
    edit this function should be warned that these commands look like, for
    example:
    /usr/local/lib/python3.8/dist-packages/ray/core/src/ray/raylet/raylet
    --redis_address=123.456.78.910 --node_ip_address=123.456.78.910
    --raylet_socket_name=... --store_socket_name=... --object_manager_port=0
    --min_worker_port=10000 --max_worker_port=10999 --node_manager_port=58578
    --redis_port=6379 --num_initial_workers=8 --maximum_startup_concurrency=8
    --static_resource_list=node:123.456.78.910,1.0,object_store_memory,66
    --config_list=plasma_store_as_thread,True
    --python_worker_command=/usr/bin/python
        /usr/local/lib/python3.8/dist-packages/ray/workers/default_worker.py
        --redis-address=123.456.78.910:6379
        --node-ip-address=123.456.78.910 --node-manager-port=58578
        --object-store-name=... --raylet-name=...
        --config-list=plasma_store_as_thread,True --temp-dir=/tmp/ray
        --metrics-agent-port=41856 --redis-password=[MASKED]
        --java_worker_command= --cpp_worker_command= --redis_password=[MASKED]
        --temp_dir=/tmp/ray --session_dir=... --metrics-agent-port=41856
        --metrics_export_port=64229
        --agent_command=/usr/bin/python
        -u /usr/local/lib/python3.8/dist-packages/ray/new_dashboard/agent.py
            --redis-address=123.456.78.910:6379 --metrics-export-port=64229
            --dashboard-agent-port=41856 --node-manager-port=58578
            --object-store-name=... --raylet-name=... --temp-dir=/tmp/ray
            --log-dir=/tmp/ray/session_2020-11-08_14-29-07_199128_278000/logs
            --redis-password=[<PASSWORD>] --object_store_memory=5037192806
            --plasma_directory=/tmp
    Longer arguments are elided with ... but all arguments from this instance
    are included, to provide a sense of what is in these.
    Indeed, we had to pull --redis-address to the front of each call to make
    this readable.
    As you can see, this is very long and complex, which is why we can't simply
    extract all the the arguments using regular expressions and present a dict
    as if we never lost track of these arguments, for example.
    Picking out --redis-address below looks like it might grab the wrong thing,
    but double-checking that we're finding the correct process by checking that
    the contents look like we expect would probably be prone to choking in
    unexpected ways.
    Notice that --redis-address appears twice. This is not a copy-paste error;
    this is the reason why the for loop below attempts to pick out every
    appearance of --redis-address.

    The --redis-address here is what is now called the --address, but it
    appears in the default_worker.py and agent.py calls as --redis-address.
    """
    pids = psutil.pids()
    redis_addresses = set()
    for pid in pids:
        try:
            proc = psutil.Process(pid)
            # HACK: Workaround for UNIX idiosyncrasy
            # Normally, cmdline() is supposed to return the argument list.
            # But it in some cases (such as when setproctitle is called),
            # an arbitrary string resembling a command-line is stored in
            # the first argument.
            # Explanation: https://unix.stackexchange.com/a/432681<commit_msg>[CLI] Fix ray commands when RAY_ADDRESS used (#11989)<commit_after>import base64
import collections
import errno
import io
import json
import logging
import multiprocessing
import os
import random
import shutil
import signal
import socket
import subprocess
import sys
import time

import colorama
import psutil
# Ray modules
import ray
import ray.ray_constants as ray_constants
import redis

resource = None
if sys.platform != "win32":
    import resource

EXE_SUFFIX = ".exe" if sys.platform == "win32" else ""

# True if processes are run in the valgrind profiler.
RUN_RAYLET_PROFILER = False
RUN_PLASMA_STORE_PROFILER = False

# Location of the redis server and module.
RAY_HOME = os.path.join(os.path.dirname(os.path.dirname(__file__)), "../..")
RAY_PATH = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
REDIS_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/thirdparty/redis/src/redis-server" + EXE_SUFFIX)
REDIS_MODULE = os.path.join(
    RAY_PATH, "core/src/ray/gcs/redis_module/libray_redis_module.so")

# Location of the plasma object store executable.
PLASMA_STORE_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/plasma/plasma_store_server" + EXE_SUFFIX)

# Location of the raylet executables.
RAYLET_EXECUTABLE = os.path.join(RAY_PATH,
                                 "core/src/ray/raylet/raylet" + EXE_SUFFIX)
GCS_SERVER_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/gcs/gcs_server" + EXE_SUFFIX)

# Location of the cpp default worker executables.
DEFAULT_WORKER_EXECUTABLE = os.path.join(
    RAY_PATH, "core/src/ray/cpp/default_worker" + EXE_SUFFIX)

# Logger for this module. It should be configured at the entry point
# into the program using Ray. Ray provides a default configuration at
# entry/init points.
logger = logging.getLogger(__name__)

ProcessInfo = collections.namedtuple("ProcessInfo", [
    "process",
    "stdout_file",
    "stderr_file",
    "use_valgrind",
    "use_gdb",
    "use_valgrind_profiler",
    "use_perftools_profiler",
    "use_tmux",
])


def serialize_config(config):
    config_pairs = []
    for key, value in config.items():
        if isinstance(value, str):
            value = value.encode("utf-8")
        if isinstance(value, bytes):
            value = base64.b64encode(value).decode("utf-8")
        config_pairs.append((key, value))
    config_str = ";".join(["{},{}".format(*kv) for kv in config_pairs])
    assert " " not in config_str, (
        "Config parameters currently do not support "
        "spaces:", config_str)
    return config_str


class ConsolePopen(subprocess.Popen):
    if sys.platform == "win32":

        def terminate(self):
            if isinstance(self.stdin, io.IOBase):
                self.stdin.close()
            if self._use_signals:
                self.send_signal(signal.CTRL_BREAK_EVENT)
            else:
                super(ConsolePopen, self).terminate()

        def __init__(self, *args, **kwargs):
            # CREATE_NEW_PROCESS_GROUP is used to send Ctrl+C on Windows:
            # https://docs.python.org/3/library/subprocess.html#subprocess.Popen.send_signal
            new_pgroup = subprocess.CREATE_NEW_PROCESS_GROUP
            flags_to_add = 0
            if ray.utils.detect_fate_sharing_support():
                # If we don't have kernel-mode fate-sharing, then don't do this
                # because our children need to be in out process group for
                # the process reaper to properly terminate them.
                flags_to_add = new_pgroup
            flags_key = "creationflags"
            if flags_to_add:
                kwargs[flags_key] = (kwargs.get(flags_key) or 0) | flags_to_add
            self._use_signals = (kwargs[flags_key] & new_pgroup)
            super(ConsolePopen, self).__init__(*args, **kwargs)


def address(ip_address, port):
    return ip_address + ":" + str(port)


def new_port():
    return random.randint(10000, 65535)


def find_redis_address(address=None):
    """
    Currently, this extracts the deprecated --redis-address from the command
    that launched the raylet running on this node, if any. Anyone looking to
    edit this function should be warned that these commands look like, for
    example:
    /usr/local/lib/python3.8/dist-packages/ray/core/src/ray/raylet/raylet
    --redis_address=123.456.78.910 --node_ip_address=123.456.78.910
    --raylet_socket_name=... --store_socket_name=... --object_manager_port=0
    --min_worker_port=10000 --max_worker_port=10999 --node_manager_port=58578
    --redis_port=6379 --num_initial_workers=8 --maximum_startup_concurrency=8
    --static_resource_list=node:123.456.78.910,1.0,object_store_memory,66
    --config_list=plasma_store_as_thread,True
    --python_worker_command=/usr/bin/python
        /usr/local/lib/python3.8/dist-packages/ray/workers/default_worker.py
        --redis-address=123.456.78.910:6379
        --node-ip-address=123.456.78.910 --node-manager-port=58578
        --object-store-name=... --raylet-name=...
        --config-list=plasma_store_as_thread,True --temp-dir=/tmp/ray
        --metrics-agent-port=41856 --redis-password=[MASKED]
        --java_worker_command= --cpp_worker_command= --redis_password=[MASKED]
        --temp_dir=/tmp/ray --session_dir=... --metrics-agent-port=41856
        --metrics_export_port=64229
        --agent_command=/usr/bin/python
        -u /usr/local/lib/python3.8/dist-packages/ray/new_dashboard/agent.py
            --redis-address=123.456.78.910:6379 --metrics-export-port=64229
            --dashboard-agent-port=41856 --node-manager-port=58578
            --object-store-name=... --raylet-name=... --temp-dir=/tmp/ray
            --log-dir=/tmp/ray/session_2020-11-08_14-29-07_199128_278000/logs
            --redis-password=[<PASSWORD>] --object_store_memory=5037192806
            --plasma_directory=/tmp
    Longer arguments are elided with ... but all arguments from this instance
    are included, to provide a sense of what is in these.
    Indeed, we had to pull --redis-address to the front of each call to make
    this readable.
    As you can see, this is very long and complex, which is why we can't simply
    extract all the the arguments using regular expressions and present a dict
    as if we never lost track of these arguments, for example.
    Picking out --redis-address below looks like it might grab the wrong thing,
    but double-checking that we're finding the correct process by checking that
    the contents look like we expect would probably be prone to choking in
    unexpected ways.
    Notice that --redis-address appears twice. This is not a copy-paste error;
    this is the reason why the for loop below attempts to pick out every
    appearance of --redis-address.

    The --redis-address here is what is now called the --address, but it
    appears in the default_worker.py and agent.py calls as --redis-address.
    """

    if "RAY_ADDRESS" in os.environ:
        return os.environ["RAY_ADDRESS"]

    pids = psutil.pids()
    redis_addresses = set()
    for pid in pids:
        try:
            proc = psutil.Process(pid)
            # HACK: Workaround for UNIX idiosyncrasy
            # Normally, cmdline() is supposed to return the argument list.
            # But it in some cases (such as when setproctitle is called),
            # an arbitrary string resembling a command-line is stored in
            # the first argument.
            # Explanation: https://unix.stackexchange.com/a/432681
{'fraction_non_alphanumeric': 0.10346457756466536, 'fraction_numerical': 0.02870264064293915, 'mean_word_length': 2.838258164852255, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7297031', 'n_tokens_mistral': 4933, 'n_tokens_neox': 4594, 'n_words': 1200}