# in python/ray/worker.py
    
    _global_node = ray.node.LocalNode()

	# Start the Ray processes. We set shutdown_at_exit=False because we
    # shutdown the node in the ray.shutdown call that happens in the atexit
    # handler. We still spawn a reaper process in case the atexit handler
    # isn't called.
    _global_node = ray.node.Node(
        head=True,
        shutdown_at_exit=False,
        spawn_reaper=True,
        ray_params=ray_params)

  _global_node = ray.node.Node(
            ray_params,
            head=False,
            shutdown_at_exit=False,
            spawn_reaper=False,
            connect_only=True)

    connect(
        _global_node,
        mode=driver_mode,
        log_to_driver=log_to_driver,
        worker=global_worker,
        driver_object_store_memory=driver_object_store_memory,
        job_id=job_id,
        internal_config=json.loads(_internal_config)
        if _internal_config else {})

	# python/ray/node.py

	def __init__(self,
                 ray_params,
                 head=False,
                 shutdown_at_exit=True,
                 spawn_reaper=True,
                 connect_only=False):
        """Start a node.

        Args:
            ray_params (ray.params.RayParams): The parameters to use to
                configure the node.
            head (bool): True if this is the head node, which means it will
                start additional processes like the Redis servers, monitor
                processes, and web UI.
            shutdown_at_exit (bool): If true, spawned processes will be cleaned
                up if this process exits normally.
            spawn_reaper (bool): If true, spawns a process that will clean up
                other spawned processes if this process dies unexpectedly.
            connect_only (bool): If true, connect to the node without starting
                new processes.
        """

return make_decorator(worker=worker)(args[0])

    return make_decorator(
        num_return_vals=num_return_vals,
        num_cpus=num_cpus,
        num_gpus=num_gpus,
        memory=memory,
        object_store_memory=object_store_memory,
        resources=resources,
        max_calls=max_calls,
        max_reconstructions=max_reconstructions,
        max_retries=max_retries,
        worker=worker)

return worker.make_actor(function_or_class, num_cpus, num_gpus,
                         memory, object_store_memory, resources,
                         max_reconstructions)

return ray.remote_function.RemoteFunction(
    function_or_class, num_cpus, num_gpus, memory,
    object_store_memory, resources, num_return_vals, max_calls,
    max_retries)

    return ActorClass._ray_from_modified_class(
        Class, ActorClassID.from_random(), max_reconstructions, num_cpus,
        num_gpus, memory, object_store_memory, resources)

	# in ray.remote()
...
       def invocation(args, kwargs):
            if not args and not kwargs and not self._function_signature:
                list_args = []
            else:
                list_args = ray.signature.flatten_args(
                    self._function_signature, args, kwargs)

            if worker.mode == ray.worker.LOCAL_MODE:
                object_ids = worker.local_mode_manager.execute(
                    self._function, self._function_descriptor, args, kwargs,
                    num_return_vals)
            else:
                object_ids = worker.core_worker.submit_task(
                    self._function_descriptor_list, list_args, num_return_vals,
                    is_direct_call, resources, max_retries)

            if len(object_ids) == 1:
                return object_ids[0]
            elif len(object_ids) > 1:
                return object_ids

        if self._decorator is not None:
            invocation = self._decorator(invocation)

        return invocation(args, kwargs)

        # in ActorClass::_remote()
    	...
    	actor_id = worker.core_worker.create_actor(
                        function_descriptor.get_function_descriptor_list(),
                        creation_args, meta.max_reconstructions, resources,
                        actor_placement_resources, is_direct_call, max_concurrency,
                        detached, is_asyncio)

        actor_handle = ActorHandle(
        actor_id,
        meta.modified_class.__module__,
        meta.class_name,
        meta.actor_method_names,
        meta.method_decorators,
        meta.method_signatures,
        meta.actor_method_num_return_vals,
        actor_method_cpu,
        worker.current_session_and_job,
        original_handle=True)

        if name is not None:
            ray.experimental.register_actor(name, actor_handle)

        return actor_handle

  return actor._actor_method_call(
                self._method_name,
                args=args,
                kwargs=kwargs,
                num_return_vals=num_return_vals)