async def METHOD_NAME(self, subtask: Subtask):
    if self._running_processor is not None:  # pragma: no cover
        running_subtask_id = await self._running_processor.get_running_subtask_id()
        # current subtask is still running
        raise SlotOccupiedAlready(
            f"There is subtask(id: {running_subtask_id}) running in {self.uid} "
            f"at {self.address}, cannot run subtask {subtask.subtask_id}"
        )
    session_id = subtask.session_id
    supervisor_address = await self._get_supervisor_address(session_id)
    if session_id not in self._session_id_to_processors:
        try:
            self._session_id_to_processors[session_id] = await mo.create_actor(
                SubtaskProcessorActor,
                session_id,
                self._band,
                supervisor_address,
                self._worker_address,
                self._subtask_processor_cls,
                uid=SubtaskProcessorActor.gen_uid(session_id),
                address=self.address,
            )
        except mo.ActorAlreadyExist:
            # when recovering actor pools, the actor created in sub pools
            # may be recovered already
            self._session_id_to_processors[session_id] = await mo.actor_ref(
                uid=SubtaskProcessorActor.gen_uid(session_id),
                address=self.address,
            )
    processor = self._session_id_to_processors[session_id]
    try:
        self._running_processor = self._last_processor = processor
        result = yield self._running_processor.run(subtask)
    finally:
        self._running_processor = None
    raise mo.Return(result)