def terminate()

in torchx/schedulers/local_scheduler.py [0:0]

• 11 lines of code
• 6 McCabe index (conditional complexity)

    def terminate(self) -> None:
        """
        terminates the underlying process for this replica
        closes stdout and stderr file handles
        safe to call multiple times
        """
        # safe to call terminate on a process that already died
        try:
            os.killpg(self.proc.pid, signal.SIGTERM)
        except ProcessLookupError as e:
            log.debug(f"Process {self.proc.pid} already got terminated")

        # close stdout and stderr log file handles
        if self.stdout:
            none_throws(self.stdout).close()
        if self.stderr:
            none_throws(self.stderr).close()
        if self.combined:
            none_throws(self.combined).close()