qq_lib.go

 15class Goer(Navigator):
 16    """
 17    Handles opening a new shell in the job's working directory on the job's main execution node.
 18    """
 19
 20    def ensure_suitable(self) -> None:
 21        """
 22        Verify that the job is in a state where its working directory can be visited.
 23
 24        Raises:
 25            QQNotSuitableError: If the working directory is not expected to exist.
 26        """
 27        if self._is_synchronized() and not self._work_dir_is_input_dir():
 28            raise QQNotSuitableError(
 29                "Job has been completed and was synchronized: working directory no longer exists."
 30            )
 31
 32        if self._is_killed() and not self.has_destination():
 33            raise QQNotSuitableError(
 34                "Job has been killed and no working directory has been created."
 35            )
 36
 37    def go(self) -> None:
 38        """
 39        Open a shell in the job's working directory on the main execution node (if the node is available).
 40
 41        Raises:
 42            QQError: If the working directory or main node is not set and navigation
 43                    cannot proceed.
 44
 45        Notes:
 46            - This method may block while waiting for a queued job to start.
 47        """
 48        if self._is_in_work_dir():
 49            logger.info("You are already in the working directory.")
 50            return
 51
 52        if self._is_killed() and not self._work_dir_is_input_dir():
 53            logger.warning(
 54                "Job has been killed: working directory may no longer exist."
 55            )
 56
 57        elif (
 58            self._is_failed() or self._is_finished()
 59        ) and not self._work_dir_is_input_dir():
 60            logger.warning(
 61                "Job has been completed: working directory may no longer exist."
 62            )
 63
 64        elif self._is_unknown_inconsistent():
 65            logger.warning("Job is in an unknown, unrecognized, or inconsistent state.")
 66
 67        elif self._is_queued():
 68            logger.warning(
 69                f"Job is {str(self._state)}: cannot visit the working directory. Will retry every {CFG.goer.wait_time} seconds."
 70            )
 71
 72            # keep retrying until the job stops being queued
 73            self._wait_queued()
 74            if self._is_in_work_dir():
 75                logger.info("You are already in the working directory.")
 76                return
 77
 78        if not self.has_destination():
 79            raise QQError(
 80                "Host ('main_node') or working directory ('work_dir') are not defined."
 81            )
 82
 83        # hint for type checker
 84        # work_dir and main_node must be set - we check that in self.hasDestination
 85        assert self._work_dir and self._main_node
 86        logger.info(f"Navigating to '{str(self._work_dir)}' on '{self._main_node}'.")
 87        self._batch_system.navigate_to_destination(self._main_node, self._work_dir)
 88
 89    def _wait_queued(self):
 90        """
 91        Wait until the job is no longer in queued/booting/waiting state.
 92
 93        Raises:
 94            QQNotSuitableError: If at any point the job is found to be finished
 95                                or killed without a working directory.
 96
 97        Note:
 98            This is a blocking method and will continue looping until the job
 99            leaves the queued/booting/waiting state or an exception is raised.
100        """
101        while self._is_queued():
102            sleep(CFG.goer.wait_time)
103            self.update()
104            self.ensure_suitable()