def _wait_for_and_process_task(self, task):
        """Wait for a task to be ready and process the task.

        Args:
            task: The task to execute.
        """
        function_descriptor = FunctionDescriptor.from_bytes_list(
            task.function_descriptor_list())
        driver_id = task.driver_id()

        # TODO(rkn): It would be preferable for actor creation tasks to share
        # more of the code path with regular task execution.
        if not task.actor_creation_id().is_nil():
            assert self.actor_id.is_nil()
            self.actor_id = task.actor_creation_id()
            self.actor_creation_task_id = task.task_id()
            actor_class = self.function_actor_manager.load_actor_class(
                driver_id, function_descriptor)
            self.actors[self.actor_id] = actor_class.__new__(actor_class)
            self.actor_checkpoint_info[self.actor_id] = ActorCheckpointInfo(
                num_tasks_since_last_checkpoint=0,
                last_checkpoint_timestamp=int(1000 * time.time()),
                checkpoint_ids=[],
            )

        execution_info = self.function_actor_manager.get_execution_info(
            driver_id, function_descriptor)

        # Execute the task.
        function_name = execution_info.function_name
        extra_data = {"name": function_name, "task_id": task.task_id().hex()}
        if task.actor_id().is_nil():
            if task.actor_creation_id().is_nil():
                title = "ray_worker:{}()".format(function_name)
                next_title = "ray_worker"
            else:
                actor = self.actors[task.actor_creation_id()]
                title = "ray_{}:{}()".format(actor.__class__.__name__,
                                             function_name)
                next_title = "ray_{}".format(actor.__class__.__name__)
        else:
            actor = self.actors[task.actor_id()]
            title = "ray_{}:{}()".format(actor.__class__.__name__,
                                         function_name)
            next_title = "ray_{}".format(actor.__class__.__name__)
        with profiling.profile("task", extra_data=extra_data):
            with _changeproctitle(title, next_title):
                self._process_task(task, execution_info)
            # Reset the state fields so the next task can run.
            self.task_context.current_task_id = TaskID.nil()
            self.task_context.task_index = 0
            self.task_context.put_index = 1
            if self.actor_id.is_nil():
                # Don't need to reset task_driver_id if the worker is an
                # actor. Because the following tasks should all have the
                # same driver id.
                self.task_driver_id = DriverID.nil()
                # Reset signal counters so that the next task can get
                # all past signals.
                ray_signal.reset()

        # Increase the task execution counter.
        self.function_actor_manager.increase_task_counter(
            driver_id, function_descriptor)

        reached_max_executions = (self.function_actor_manager.get_task_counter(
            driver_id, function_descriptor) == execution_info.max_calls)
        if reached_max_executions:
            self.raylet_client.disconnect()
            sys.exit(0)
Wait for a task to be ready and process the task.

        Args:
            task: The task to execute.