Start a raylet, which is a combined local scheduler and object manager.

    Args:
        redis_address (str): The address of the primary Redis server.
        node_ip_address (str): The IP address of this node.
        raylet_name (str): The name of the raylet socket to create.
        plasma_store_name (str): The name of the plasma store socket to connect
             to.
        worker_path (str): The path of the Python file that new worker
            processes will execute.
        temp_dir (str): The path of the temporary directory Ray will use.
        num_cpus: The CPUs allocated for this raylet.
        num_gpus: The GPUs allocated for this raylet.
        resources: The custom resources allocated for this raylet.
        object_manager_port: The port to use for the object manager. If this is
            None, then the object manager will choose its own port.
        node_manager_port: The port to use for the node manager. If this is
            None, then the node manager will choose its own port.
        redis_password: The password to use when connecting to Redis.
        use_valgrind (bool): True if the raylet should be started inside
            of valgrind. If this is True, use_profiler must be False.
        use_profiler (bool): True if the raylet should be started inside
            a profiler. If this is True, use_valgrind must be False.
        stdout_file: A file handle opened for writing to redirect stdout to. If
            no redirection should happen, then this should be None.
        stderr_file: A file handle opened for writing to redirect stderr to. If
            no redirection should happen, then this should be None.
        config (dict|None): Optional Raylet configuration that will
            override defaults in RayConfig.
        include_java (bool): If True, the raylet backend can also support
            Java worker.
        java_worker_options (str): The command options for Java worker.
    Returns:
        ProcessInfo for the process that was started.

def start_raylet(redis_address,
                 node_ip_address,
                 raylet_name,
                 plasma_store_name,
                 worker_path,
                 temp_dir,
                 num_cpus=None,
                 num_gpus=None,
                 resources=None,
                 object_manager_port=None,
                 node_manager_port=None,
                 redis_password=None,
                 use_valgrind=False,
                 use_profiler=False,
                 stdout_file=None,
                 stderr_file=None,
                 config=None,
                 include_java=False,
                 java_worker_options=None,
                 load_code_from_local=False):
    """Start a raylet, which is a combined local scheduler and object manager.

    Args:
        redis_address (str): The address of the primary Redis server.
        node_ip_address (str): The IP address of this node.
        raylet_name (str): The name of the raylet socket to create.
        plasma_store_name (str): The name of the plasma store socket to connect
             to.
        worker_path (str): The path of the Python file that new worker
            processes will execute.
        temp_dir (str): The path of the temporary directory Ray will use.
        num_cpus: The CPUs allocated for this raylet.
        num_gpus: The GPUs allocated for this raylet.
        resources: The custom resources allocated for this raylet.
        object_manager_port: The port to use for the object manager. If this is
            None, then the object manager will choose its own port.
        node_manager_port: The port to use for the node manager. If this is
            None, then the node manager will choose its own port.
        redis_password: The password to use when connecting to Redis.
        use_valgrind (bool): True if the raylet should be started inside
            of valgrind. If this is True, use_profiler must be False.
        use_profiler (bool): True if the raylet should be started inside
            a profiler. If this is True, use_valgrind must be False.
        stdout_file: A file handle opened for writing to redirect stdout to. If
            no redirection should happen, then this should be None.
        stderr_file: A file handle opened for writing to redirect stderr to. If
            no redirection should happen, then this should be None.
        config (dict|None): Optional Raylet configuration that will
            override defaults in RayConfig.
        include_java (bool): If True, the raylet backend can also support
            Java worker.
        java_worker_options (str): The command options for Java worker.
    Returns:
        ProcessInfo for the process that was started.
    """
    config = config or {}
    config_str = ",".join(["{},{}".format(*kv) for kv in config.items()])

    if use_valgrind and use_profiler:
        raise Exception("Cannot use valgrind and profiler at the same time.")

    num_initial_workers = (num_cpus if num_cpus is not None else
                           multiprocessing.cpu_count())

    static_resources = check_and_update_resources(num_cpus, num_gpus,
                                                  resources)

    # Limit the number of workers that can be started in parallel by the
    # raylet. However, make sure it is at least 1.
    num_cpus_static = static_resources.get("CPU", 0)
    maximum_startup_concurrency = max(
        1, min(multiprocessing.cpu_count(), num_cpus_static))

    # Format the resource argument in a form like 'CPU,1.0,GPU,0,Custom,3'.
    resource_argument = ",".join(
        ["{},{}".format(*kv) for kv in static_resources.items()])

    gcs_ip_address, gcs_port = redis_address.split(":")

    if include_java is True:
        java_worker_options = (java_worker_options
                               or DEFAULT_JAVA_WORKER_OPTIONS)
        java_worker_command = build_java_worker_command(
            java_worker_options,
            redis_address,
            plasma_store_name,
            raylet_name,
            redis_password,
            os.path.join(temp_dir, "sockets"),
        )
    else:
        java_worker_command = ""

    # Create the command that the Raylet will use to start workers.
    start_worker_command = ("{} {} "
                            "--node-ip-address={} "
                            "--object-store-name={} "
                            "--raylet-name={} "
                            "--redis-address={} "
                            "--temp-dir={}".format(
                                sys.executable, worker_path, node_ip_address,
                                plasma_store_name, raylet_name, redis_address,
                                temp_dir))
    if redis_password:
        start_worker_command += " --redis-password {}".format(redis_password)

    # If the object manager port is None, then use 0 to cause the object
    # manager to choose its own port.
    if object_manager_port is None:
        object_manager_port = 0
    # If the node manager port is None, then use 0 to cause the node manager
    # to choose its own port.
    if node_manager_port is None:
        node_manager_port = 0

    if load_code_from_local:
        start_worker_command += " --load-code-from-local "

    command = [
        RAYLET_EXECUTABLE,
        "--raylet_socket_name={}".format(raylet_name),
        "--store_socket_name={}".format(plasma_store_name),
        "--object_manager_port={}".format(object_manager_port),
        "--node_manager_port={}".format(node_manager_port),
        "--node_ip_address={}".format(node_ip_address),
        "--redis_address={}".format(gcs_ip_address),
        "--redis_port={}".format(gcs_port),
        "--num_initial_workers={}".format(num_initial_workers),
        "--maximum_startup_concurrency={}".format(maximum_startup_concurrency),
        "--static_resource_list={}".format(resource_argument),
        "--config_list={}".format(config_str),
        "--python_worker_command={}".format(start_worker_command),
        "--java_worker_command={}".format(java_worker_command),
        "--redis_password={}".format(redis_password or ""),
        "--temp_dir={}".format(temp_dir),
    ]
    process_info = start_ray_process(
        command,
        ray_constants.PROCESS_TYPE_RAYLET,
        use_valgrind=use_valgrind,
        use_gdb=False,
        use_valgrind_profiler=use_profiler,
        use_perftools_profiler=("RAYLET_PERFTOOLS_PATH" in os.environ),
        stdout_file=stdout_file,
        stderr_file=stderr_file)

    return process_info