Start the Redis global state store.

    Args:
        node_ip_address: The IP address of the current node. This is only used
            for recording the log filenames in Redis.
        redirect_files: The list of (stdout, stderr) file pairs.
        port (int): If provided, the primary Redis shard will be started on
            this port.
        redis_shard_ports: A list of the ports to use for the non-primary Redis
            shards.
        num_redis_shards (int): If provided, the number of Redis shards to
            start, in addition to the primary one. The default value is one
            shard.
        redis_max_clients: If this is provided, Ray will attempt to configure
            Redis with this maxclients number.
        redirect_worker_output (bool): True if worker output should be
            redirected to a file and false otherwise. Workers will have access
            to this value when they start up.
        password (str): Prevents external clients without the password
            from connecting to Redis if provided.
        use_credis: If True, additionally load the chain-replicated libraries
            into the redis servers.  Defaults to None, which means its value is
            set by the presence of "RAY_USE_NEW_GCS" in os.environ.
        redis_max_memory: The max amount of memory (in bytes) to allow each
            redis shard to use. Once the limit is exceeded, redis will start
            LRU eviction of entries. This only applies to the sharded redis
            tables (task, object, and profile tables). By default, this is
            capped at 10GB but can be set higher.
        include_java (bool): If True, the raylet backend can also support
            Java worker.

    Returns:
        A tuple of the address for the primary Redis shard, a list of
            addresses for the remaining shards, and the processes that were
            started.

def start_redis(node_ip_address,
                redirect_files,
                port=None,
                redis_shard_ports=None,
                num_redis_shards=1,
                redis_max_clients=None,
                redirect_worker_output=False,
                password=None,
                use_credis=None,
                redis_max_memory=None,
                include_java=False):
    """Start the Redis global state store.

    Args:
        node_ip_address: The IP address of the current node. This is only used
            for recording the log filenames in Redis.
        redirect_files: The list of (stdout, stderr) file pairs.
        port (int): If provided, the primary Redis shard will be started on
            this port.
        redis_shard_ports: A list of the ports to use for the non-primary Redis
            shards.
        num_redis_shards (int): If provided, the number of Redis shards to
            start, in addition to the primary one. The default value is one
            shard.
        redis_max_clients: If this is provided, Ray will attempt to configure
            Redis with this maxclients number.
        redirect_worker_output (bool): True if worker output should be
            redirected to a file and false otherwise. Workers will have access
            to this value when they start up.
        password (str): Prevents external clients without the password
            from connecting to Redis if provided.
        use_credis: If True, additionally load the chain-replicated libraries
            into the redis servers.  Defaults to None, which means its value is
            set by the presence of "RAY_USE_NEW_GCS" in os.environ.
        redis_max_memory: The max amount of memory (in bytes) to allow each
            redis shard to use. Once the limit is exceeded, redis will start
            LRU eviction of entries. This only applies to the sharded redis
            tables (task, object, and profile tables). By default, this is
            capped at 10GB but can be set higher.
        include_java (bool): If True, the raylet backend can also support
            Java worker.

    Returns:
        A tuple of the address for the primary Redis shard, a list of
            addresses for the remaining shards, and the processes that were
            started.
    """

    if len(redirect_files) != 1 + num_redis_shards:
        raise ValueError("The number of redirect file pairs should be equal "
                         "to the number of redis shards (including the "
                         "primary shard) we will start.")
    if redis_shard_ports is None:
        redis_shard_ports = num_redis_shards * [None]
    elif len(redis_shard_ports) != num_redis_shards:
        raise Exception("The number of Redis shard ports does not match the "
                        "number of Redis shards.")

    processes = []

    if use_credis is None:
        use_credis = ("RAY_USE_NEW_GCS" in os.environ)
    if use_credis:
        if password is not None:
            # TODO(pschafhalter) remove this once credis supports
            # authenticating Redis ports
            raise Exception("Setting the `redis_password` argument is not "
                            "supported in credis. To run Ray with "
                            "password-protected Redis ports, ensure that "
                            "the environment variable `RAY_USE_NEW_GCS=off`.")
        assert num_redis_shards == 1, (
            "For now, RAY_USE_NEW_GCS supports 1 shard, and credis "
            "supports 1-node chain for that shard only.")

    if use_credis:
        redis_executable = CREDIS_EXECUTABLE
        # TODO(suquark): We need credis here because some symbols need to be
        # imported from credis dynamically through dlopen when Ray is built
        # with RAY_USE_NEW_GCS=on. We should remove them later for the primary
        # shard.
        # See src/ray/gcs/redis_module/ray_redis_module.cc
        redis_modules = [CREDIS_MASTER_MODULE, REDIS_MODULE]
    else:
        redis_executable = REDIS_EXECUTABLE
        redis_modules = [REDIS_MODULE]

    redis_stdout_file, redis_stderr_file = redirect_files[0]
    # Start the primary Redis shard.
    port, p = _start_redis_instance(
        redis_executable,
        modules=redis_modules,
        port=port,
        password=password,
        redis_max_clients=redis_max_clients,
        # Below we use None to indicate no limit on the memory of the
        # primary Redis shard.
        redis_max_memory=None,
        stdout_file=redis_stdout_file,
        stderr_file=redis_stderr_file)
    processes.append(p)
    redis_address = address(node_ip_address, port)

    # Register the number of Redis shards in the primary shard, so that clients
    # know how many redis shards to expect under RedisShards.
    primary_redis_client = redis.StrictRedis(
        host=node_ip_address, port=port, password=password)
    primary_redis_client.set("NumRedisShards", str(num_redis_shards))

    # Put the redirect_worker_output bool in the Redis shard so that workers
    # can access it and know whether or not to redirect their output.
    primary_redis_client.set("RedirectOutput", 1
                             if redirect_worker_output else 0)

    # put the include_java bool to primary redis-server, so that other nodes
    # can access it and know whether or not to enable cross-languages.
    primary_redis_client.set("INCLUDE_JAVA", 1 if include_java else 0)

    # Store version information in the primary Redis shard.
    _put_version_info_in_redis(primary_redis_client)

    # Calculate the redis memory.
    system_memory = ray.utils.get_system_memory()
    if redis_max_memory is None:
        redis_max_memory = min(
            ray_constants.DEFAULT_REDIS_MAX_MEMORY_BYTES,
            max(
                int(system_memory * 0.2),
                ray_constants.REDIS_MINIMUM_MEMORY_BYTES))
    if redis_max_memory < ray_constants.REDIS_MINIMUM_MEMORY_BYTES:
        raise ValueError("Attempting to cap Redis memory usage at {} bytes, "
                         "but the minimum allowed is {} bytes.".format(
                             redis_max_memory,
                             ray_constants.REDIS_MINIMUM_MEMORY_BYTES))

    # Start other Redis shards. Each Redis shard logs to a separate file,
    # prefixed by "redis-<shard number>".
    redis_shards = []
    for i in range(num_redis_shards):
        redis_stdout_file, redis_stderr_file = redirect_files[i + 1]
        if use_credis:
            redis_executable = CREDIS_EXECUTABLE
            # It is important to load the credis module BEFORE the ray module,
            # as the latter contains an extern declaration that the former
            # supplies.
            redis_modules = [CREDIS_MEMBER_MODULE, REDIS_MODULE]
        else:
            redis_executable = REDIS_EXECUTABLE
            redis_modules = [REDIS_MODULE]

        redis_shard_port, p = _start_redis_instance(
            redis_executable,
            modules=redis_modules,
            port=redis_shard_ports[i],
            password=password,
            redis_max_clients=redis_max_clients,
            redis_max_memory=redis_max_memory,
            stdout_file=redis_stdout_file,
            stderr_file=redis_stderr_file)
        processes.append(p)

        shard_address = address(node_ip_address, redis_shard_port)
        redis_shards.append(shard_address)
        # Store redis shard information in the primary redis shard.
        primary_redis_client.rpush("RedisShards", shard_address)

    if use_credis:
        # Configure the chain state. The way it is intended to work is
        # the following:
        #
        # PRIMARY_SHARD
        #
        # SHARD_1 (master replica) -> SHARD_1 (member replica)
        #                                        -> SHARD_1 (member replica)
        #
        # SHARD_2 (master replica) -> SHARD_2 (member replica)
        #                                        -> SHARD_2 (member replica)
        # ...
        #
        #
        # If we have credis members in future, their modules should be:
        # [CREDIS_MEMBER_MODULE, REDIS_MODULE], and they will be initialized by
        # execute_command("MEMBER.CONNECT_TO_MASTER", node_ip_address, port)
        #
        # Currently we have num_redis_shards == 1, so only one chain will be
        # created, and the chain only contains master.

        # TODO(suquark): Currently, this is not correct because we are
        # using the master replica as the primary shard. This should be
        # fixed later. I had tried to fix it but failed because of heartbeat
        # issues.
        primary_client = redis.StrictRedis(
            host=node_ip_address, port=port, password=password)
        shard_client = redis.StrictRedis(
            host=node_ip_address, port=redis_shard_port, password=password)
        primary_client.execute_command("MASTER.ADD", node_ip_address,
                                       redis_shard_port)
        shard_client.execute_command("MEMBER.CONNECT_TO_MASTER",
                                     node_ip_address, port)

    return redis_address, redis_shards, processes