python/ray/remote_function.py
def _remote(self, args=None, kwargs=None, **task_options):
        """Submit the remote function for execution."""
        # We pop the "max_calls" coming from "@ray.remote" here. We no longer need
        # it in "_remote()".
        task_options.pop("max_calls", None)
        if client_mode_should_convert(auto_init=True):
            return client_mode_convert_function(self, args, kwargs, **task_options)

        worker = ray.worker.global_worker
        worker.check_connected()

        # If this function was not exported in this session and job, we need to
        # export this function again, because the current GCS doesn't have it.
        if (
            not self._is_cross_language
            and self._last_export_session_and_job != worker.current_session_and_job
        ):
            self._function_descriptor = PythonFunctionDescriptor.from_function(
                self._function, self._uuid
            )
            # There is an interesting question here. If the remote function is
            # used by a subsequent driver (in the same script), should the
            # second driver pickle the function again? If yes, then the remote
            # function definition can differ in the second driver (e.g., if
            # variables in its closure have changed). We probably want the
            # behavior of the remote function in the second driver to be
            # independent of whether or not the function was invoked by the
            # first driver. This is an argument for repickling the function,
            # which we do here.
            try:
                self._pickled_function = pickle.dumps(self._function)
            except TypeError as e:
                msg = (
                    "Could not serialize the function "
                    f"{self._function_descriptor.repr}. Check "
                    "https://docs.ray.io/en/master/serialization.html#troubleshooting "
                    "for more information."
                )
                raise TypeError(msg) from e

            self._last_export_session_and_job = worker.current_session_and_job
            worker.function_actor_manager.export(self)

        kwargs = {} if kwargs is None else kwargs
        args = [] if args is None else args

        # fill task required options
        for k, v in ray_option_utils.task_options.items():
            task_options[k] = task_options.get(k, v.default_value)
        # "max_calls" already takes effects and should not apply again.
        # Remove the default value here.
        task_options.pop("max_calls", None)

        # TODO(suquark): cleanup these fields
        name = task_options["name"]
        runtime_env = parse_runtime_env(task_options["runtime_env"])
        placement_group = task_options["placement_group"]
        placement_group_bundle_index = task_options["placement_group_bundle_index"]
        placement_group_capture_child_tasks = task_options[
            "placement_group_capture_child_tasks"
        ]
        scheduling_strategy = task_options["scheduling_strategy"]
        num_returns = task_options["num_returns"]
        max_retries = task_options["max_retries"]
        retry_exceptions = task_options["retry_exceptions"]

        resources = ray._private.utils.resources_from_ray_options(task_options)

        if scheduling_strategy is None or isinstance(
            scheduling_strategy, PlacementGroupSchedulingStrategy
        ):
            if isinstance(scheduling_strategy, PlacementGroupSchedulingStrategy):
                placement_group = scheduling_strategy.placement_group
                placement_group_bundle_index = (
                    scheduling_strategy.placement_group_bundle_index
                )
                placement_group_capture_child_tasks = (
                    scheduling_strategy.placement_group_capture_child_tasks
                )

            if placement_group_capture_child_tasks is None:
                placement_group_capture_child_tasks = (
                    worker.should_capture_child_tasks_in_placement_group
                )
            placement_group = configure_placement_group_based_on_context(
                placement_group_capture_child_tasks,
                placement_group_bundle_index,
                resources,
                {},  # no placement_resources for tasks
                self._function_descriptor.function_name,
                placement_group=placement_group,
            )
            if not placement_group.is_empty:
                scheduling_strategy = PlacementGroupSchedulingStrategy(
                    placement_group,
                    placement_group_bundle_index,
                    placement_group_capture_child_tasks,
                )
            else:
                scheduling_strategy = "DEFAULT"

        serialized_runtime_env_info = None
        if runtime_env is not None:
            serialized_runtime_env_info = get_runtime_env_info(
                runtime_env,
                is_job_runtime_env=False,
                serialize=True,
            )

        def invocation(args, kwargs):
            if self._is_cross_language:
                list_args = cross_language.format_args(worker, args, kwargs)
            elif not args and not kwargs and not self._function_signature:
                list_args = []
            else:
                list_args = ray._private.signature.flatten_args(
                    self._function_signature, args, kwargs
                )

            if worker.mode == ray.worker.LOCAL_MODE:
                assert (
                    not self._is_cross_language
                ), "Cross language remote function cannot be executed locally."
            object_refs = worker.core_worker.submit_task(
                self._language,
                self._function_descriptor,
                list_args,
                name if name is not None else "",
                num_returns,
                resources,
                max_retries,
                retry_exceptions,
                scheduling_strategy,
                worker.debugger_breakpoint,
                serialized_runtime_env_info or "{}",
            )
            # Reset worker's debug context from the last "remote" command
            # (which applies only to this .remote call).
            worker.debugger_breakpoint = b""
            if len(object_refs) == 1:
                return object_refs[0]
            elif len(object_refs) > 1:
                return object_refs

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

        return invocation(args, kwargs)
Submit the remote function for execution.
Submit the remote function for execution.
Submit the remote function for execution .
def _remote(self, args=None, kwargs=None, **task_options):
        task_options.pop("max_calls", None)
        if client_mode_should_convert(auto_init=True):
            return client_mode_convert_function(self, args, kwargs, **task_options)
        worker = ray.worker.global_worker
        worker.check_connected()
        if (
            not self._is_cross_language
            and self._last_export_session_and_job != worker.current_session_and_job
        ):
            self._function_descriptor = PythonFunctionDescriptor.from_function(
                self._function, self._uuid
            )
            try:
                self._pickled_function = pickle.dumps(self._function)
            except TypeError as e:
                msg = (
                    "Could not serialize the function "
                    f"{self._function_descriptor.repr}. Check "
                    "https://docs.ray.io/en/master/serialization.html#troubleshooting "
                    "for more information."
                )
                raise TypeError(msg) from e
            self._last_export_session_and_job = worker.current_session_and_job
            worker.function_actor_manager.export(self)
        kwargs = {} if kwargs is None else kwargs
        args = [] if args is None else args
        for k, v in ray_option_utils.task_options.items():
            task_options[k] = task_options.get(k, v.default_value)
        task_options.pop("max_calls", None)
        name = task_options["name"]
        runtime_env = parse_runtime_env(task_options["runtime_env"])
        placement_group = task_options["placement_group"]
        placement_group_bundle_index = task_options["placement_group_bundle_index"]
        placement_group_capture_child_tasks = task_options[
            "placement_group_capture_child_tasks"
        ]
        scheduling_strategy = task_options["scheduling_strategy"]
        num_returns = task_options["num_returns"]
        max_retries = task_options["max_retries"]
        retry_exceptions = task_options["retry_exceptions"]
        resources = ray._private.utils.resources_from_ray_options(task_options)
        if scheduling_strategy is None or isinstance(
            scheduling_strategy, PlacementGroupSchedulingStrategy
        ):
            if isinstance(scheduling_strategy, PlacementGroupSchedulingStrategy):
                placement_group = scheduling_strategy.placement_group
                placement_group_bundle_index = (
                    scheduling_strategy.placement_group_bundle_index
                )
                placement_group_capture_child_tasks = (
                    scheduling_strategy.placement_group_capture_child_tasks
                )
            if placement_group_capture_child_tasks is None:
                placement_group_capture_child_tasks = (
                    worker.should_capture_child_tasks_in_placement_group
                )
            placement_group = configure_placement_group_based_on_context(
                placement_group_capture_child_tasks,
                placement_group_bundle_index,
                resources,
                {},  
                self._function_descriptor.function_name,
                placement_group=placement_group,
            )
            if not placement_group.is_empty:
                scheduling_strategy = PlacementGroupSchedulingStrategy(
                    placement_group,
                    placement_group_bundle_index,
                    placement_group_capture_child_tasks,
                )
            else:
                scheduling_strategy = "DEFAULT"
        serialized_runtime_env_info = None
        if runtime_env is not None:
            serialized_runtime_env_info = get_runtime_env_info(
                runtime_env,
                is_job_runtime_env=False,
                serialize=True,
            )
        def invocation(args, kwargs):
            if self._is_cross_language:
                list_args = cross_language.format_args(worker, args, kwargs)
            elif not args and not kwargs and not self._function_signature:
                list_args = []
            else:
                list_args = ray._private.signature.flatten_args(
                    self._function_signature, args, kwargs
                )
            if worker.mode == ray.worker.LOCAL_MODE:
                assert (
                    not self._is_cross_language
                ), "Cross language remote function cannot be executed locally."
            object_refs = worker.core_worker.submit_task(
                self._language,
                self._function_descriptor,
                list_args,
                name if name is not None else "",
                num_returns,
                resources,
                max_retries,
                retry_exceptions,
                scheduling_strategy,
                worker.debugger_breakpoint,
                serialized_runtime_env_info or "{}",
            )
            worker.debugger_breakpoint = b""
            if len(object_refs) == 1:
                return object_refs[0]
            elif len(object_refs) > 1:
                return object_refs
        if self._decorator is not None:
            invocation = self._decorator(invocation)
        return invocation(args, kwargs)
def _remote ( self , args = None , kwargs = None , ** task_options ) : task_options . pop ( "max_calls" , None ) if client_mode_should_convert ( auto_init = True ) : return client_mode_convert_function ( self , args , kwargs , ** task_options ) worker = ray . worker . global_worker worker . check_connected ( ) if ( not self . _is_cross_language and self . _last_export_session_and_job != worker . current_session_and_job ) : self . _function_descriptor = PythonFunctionDescriptor . from_function ( self . _function , self . _uuid ) try : self . _pickled_function = pickle . dumps ( self . _function ) except TypeError as e : msg = ( "Could not serialize the function " f"{self._function_descriptor.repr}. Check " "https://docs.ray.io/en/master/serialization.html#troubleshooting " "for more information." ) raise TypeError ( msg ) from e self . _last_export_session_and_job = worker . current_session_and_job worker . function_actor_manager . export ( self ) kwargs = { } if kwargs is None else kwargs args = [ ] if args is None else args for k , v in ray_option_utils . task_options . items ( ) : task_options [ k ] = task_options . get ( k , v . default_value ) task_options . pop ( "max_calls" , None ) name = task_options [ "name" ] runtime_env = parse_runtime_env ( task_options [ "runtime_env" ] ) placement_group = task_options [ "placement_group" ] placement_group_bundle_index = task_options [ "placement_group_bundle_index" ] placement_group_capture_child_tasks = task_options [ "placement_group_capture_child_tasks" ] scheduling_strategy = task_options [ "scheduling_strategy" ] num_returns = task_options [ "num_returns" ] max_retries = task_options [ "max_retries" ] retry_exceptions = task_options [ "retry_exceptions" ] resources = ray . _private . utils . resources_from_ray_options ( task_options ) if scheduling_strategy is None or isinstance ( scheduling_strategy , PlacementGroupSchedulingStrategy ) : if isinstance ( scheduling_strategy , PlacementGroupSchedulingStrategy ) : placement_group = scheduling_strategy . placement_group placement_group_bundle_index = ( scheduling_strategy . placement_group_bundle_index ) placement_group_capture_child_tasks = ( scheduling_strategy . placement_group_capture_child_tasks ) if placement_group_capture_child_tasks is None : placement_group_capture_child_tasks = ( worker . should_capture_child_tasks_in_placement_group ) placement_group = configure_placement_group_based_on_context ( placement_group_capture_child_tasks , placement_group_bundle_index , resources , { } , self . _function_descriptor . function_name , placement_group = placement_group , ) if not placement_group . is_empty : scheduling_strategy = PlacementGroupSchedulingStrategy ( placement_group , placement_group_bundle_index , placement_group_capture_child_tasks , ) else : scheduling_strategy = "DEFAULT" serialized_runtime_env_info = None if runtime_env is not None : serialized_runtime_env_info = get_runtime_env_info ( runtime_env , is_job_runtime_env = False , serialize = True , ) def invocation ( args , kwargs ) : if self . _is_cross_language : list_args = cross_language . format_args ( worker , args , kwargs ) elif not args and not kwargs and not self . _function_signature : list_args = [ ] else : list_args = ray . _private . signature . flatten_args ( self . _function_signature , args , kwargs ) if worker . mode == ray . worker . LOCAL_MODE : assert ( not self . _is_cross_language ) , "Cross language remote function cannot be executed locally." object_refs = worker . core_worker . submit_task ( self . _language , self . _function_descriptor , list_args , name if name is not None else "" , num_returns , resources , max_retries , retry_exceptions , scheduling_strategy , worker . debugger_breakpoint , serialized_runtime_env_info or "{}" , ) worker . debugger_breakpoint = b"" if len ( object_refs ) == 1 : return object_refs [ 0 ] elif len ( object_refs ) > 1 : return object_refs if self . _decorator is not None : invocation = self . _decorator ( invocation ) return invocation ( args , kwargs )
Submit the remote function for execution.
Submit the remote function for execution .
"""Submit the remote function for execution.""" # We pop the "max_calls" coming from "@ray.remote" here. We no longer need # it in "_remote()". # If this function was not exported in this session and job, we need to # export this function again, because the current GCS doesn't have it. # There is an interesting question here. If the remote function is # used by a subsequent driver (in the same script), should the # second driver pickle the function again? If yes, then the remote # function definition can differ in the second driver (e.g., if # variables in its closure have changed). We probably want the # behavior of the remote function in the second driver to be # independent of whether or not the function was invoked by the # first driver. This is an argument for repickling the function, # which we do here. # fill task required options # "max_calls" already takes effects and should not apply again. # Remove the default value here. # TODO(suquark): cleanup these fields # no placement_resources for tasks # Reset worker's debug context from the last "remote" command # (which applies only to this .remote call).
self
args
kwargs