Submit a remote task to the scheduler.

        Tell the scheduler to schedule the execution of the function with
        function_descriptor with arguments args. Retrieve object IDs for the
        outputs of the function from the scheduler and immediately return them.

        Args:
            function_descriptor: The function descriptor to execute.
            args: The arguments to pass into the function. Arguments can be
                object IDs or they can be values. If they are values, they must
                be serializable objects.
            actor_id: The ID of the actor that this task is for.
            actor_counter: The counter of the actor task.
            actor_creation_id: The ID of the actor to create, if this is an
                actor creation task.
            actor_creation_dummy_object_id: If this task is an actor method,
                then this argument is the dummy object ID associated with the
                actor creation task for the corresponding actor.
            execution_dependencies: The execution dependencies for this task.
            num_return_vals: The number of return values this function should
                have.
            resources: The resource requirements for this task.
            placement_resources: The resources required for placing the task.
                If this is not provided or if it is an empty dictionary, then
                the placement resources will be equal to resources.
            driver_id: The ID of the relevant driver. This is almost always the
                driver ID of the driver that is currently running. However, in
                the exceptional case that an actor task is being dispatched to
                an actor created by a different driver, this should be the
                driver ID of the driver that created the actor.

        Returns:
            The return object IDs for this task.

def submit_task(self,
                    function_descriptor,
                    args,
                    actor_id=None,
                    actor_handle_id=None,
                    actor_counter=0,
                    actor_creation_id=None,
                    actor_creation_dummy_object_id=None,
                    max_actor_reconstructions=0,
                    execution_dependencies=None,
                    new_actor_handles=None,
                    num_return_vals=None,
                    resources=None,
                    placement_resources=None,
                    driver_id=None):
        """Submit a remote task to the scheduler.

        Tell the scheduler to schedule the execution of the function with
        function_descriptor with arguments args. Retrieve object IDs for the
        outputs of the function from the scheduler and immediately return them.

        Args:
            function_descriptor: The function descriptor to execute.
            args: The arguments to pass into the function. Arguments can be
                object IDs or they can be values. If they are values, they must
                be serializable objects.
            actor_id: The ID of the actor that this task is for.
            actor_counter: The counter of the actor task.
            actor_creation_id: The ID of the actor to create, if this is an
                actor creation task.
            actor_creation_dummy_object_id: If this task is an actor method,
                then this argument is the dummy object ID associated with the
                actor creation task for the corresponding actor.
            execution_dependencies: The execution dependencies for this task.
            num_return_vals: The number of return values this function should
                have.
            resources: The resource requirements for this task.
            placement_resources: The resources required for placing the task.
                If this is not provided or if it is an empty dictionary, then
                the placement resources will be equal to resources.
            driver_id: The ID of the relevant driver. This is almost always the
                driver ID of the driver that is currently running. However, in
                the exceptional case that an actor task is being dispatched to
                an actor created by a different driver, this should be the
                driver ID of the driver that created the actor.

        Returns:
            The return object IDs for this task.
        """
        with profiling.profile("submit_task"):
            if actor_id is None:
                assert actor_handle_id is None
                actor_id = ActorID.nil()
                actor_handle_id = ActorHandleID.nil()
            else:
                assert actor_handle_id is not None

            if actor_creation_id is None:
                actor_creation_id = ActorID.nil()

            if actor_creation_dummy_object_id is None:
                actor_creation_dummy_object_id = ObjectID.nil()

            # Put large or complex arguments that are passed by value in the
            # object store first.
            args_for_raylet = []
            for arg in args:
                if isinstance(arg, ObjectID):
                    args_for_raylet.append(arg)
                elif ray._raylet.check_simple_value(arg):
                    args_for_raylet.append(arg)
                else:
                    args_for_raylet.append(put(arg))

            # By default, there are no execution dependencies.
            if execution_dependencies is None:
                execution_dependencies = []

            if new_actor_handles is None:
                new_actor_handles = []

            if driver_id is None:
                driver_id = self.task_driver_id

            if resources is None:
                raise ValueError("The resources dictionary is required.")
            for value in resources.values():
                assert (isinstance(value, int) or isinstance(value, float))
                if value < 0:
                    raise ValueError(
                        "Resource quantities must be nonnegative.")
                if (value >= 1 and isinstance(value, float)
                        and not value.is_integer()):
                    raise ValueError(
                        "Resource quantities must all be whole numbers.")

            # Remove any resources with zero quantity requirements
            resources = {
                resource_label: resource_quantity
                for resource_label, resource_quantity in resources.items()
                if resource_quantity > 0
            }

            if placement_resources is None:
                placement_resources = {}

            # Increment the worker's task index to track how many tasks
            # have been submitted by the current task so far.
            self.task_context.task_index += 1
            # The parent task must be set for the submitted task.
            assert not self.current_task_id.is_nil()
            # Current driver id must not be nil when submitting a task.
            # Because every task must belong to a driver.
            assert not self.task_driver_id.is_nil()
            # Submit the task to raylet.
            function_descriptor_list = (
                function_descriptor.get_function_descriptor_list())
            assert isinstance(driver_id, DriverID)
            task = ray._raylet.Task(
                driver_id,
                function_descriptor_list,
                args_for_raylet,
                num_return_vals,
                self.current_task_id,
                self.task_context.task_index,
                actor_creation_id,
                actor_creation_dummy_object_id,
                max_actor_reconstructions,
                actor_id,
                actor_handle_id,
                actor_counter,
                new_actor_handles,
                execution_dependencies,
                resources,
                placement_resources,
            )
            self.raylet_client.submit_task(task)

            return task.returns()