我们从Python开源项目中，提取了以下18个代码示例，用于说明如何使用psutil.TimeoutExpired()。
def test_wait_non_children(self): # test wait() against processes which are not our children code = "import sys;" code += "from subprocess import Popen, PIPE;" code += "cmd = ['%s', '-c', 'import time; time.sleep(60)'];" % PYTHON code += "sp = Popen(cmd, stdout=PIPE);" code += "sys.stdout.write(str(sp.pid));" sproc = get_test_subprocess([PYTHON, "-c", code], stdout=subprocess.PIPE) grandson_pid = int(sproc.stdout.read()) grandson_proc = psutil.Process(grandson_pid) try: self.assertRaises(psutil.TimeoutExpired, grandson_proc.wait, 0.01) grandson_proc.kill() ret = grandson_proc.wait() self.assertEqual(ret, None) finally: reap_children(recursive=True)
def test_wait_timeout_0(self): sproc = get_test_subprocess() p = psutil.Process(sproc.pid) self.assertRaises(psutil.TimeoutExpired, p.wait, 0) p.kill() stop_at = time.time() + 2 while True: try: code = p.wait(0) except psutil.TimeoutExpired: if time.time() >= stop_at: raise else: break if POSIX: self.assertEqual(code, signal.SIGKILL) else: self.assertEqual(code, 0) self.assertFalse(p.is_running())
def kill_process(process): """Kill a single process. Args: process (psutil.Process): process to kill. """ if process.is_running() is False: return process.kill() try: process.wait(PROCESS_TERMINATION_TIMEOUT) except psutil.TimeoutExpired: core_log.warning("Process %d failed to terminate", process.pid)
def test_wait_timeout_0(self): sproc = get_test_subprocess() p = psutil.Process(sproc.pid) self.assertRaises(psutil.TimeoutExpired, p.wait, 0) p.kill() stop_at = time.time() + 2 while True: try: code = p.wait(0) except psutil.TimeoutExpired: if time.time() >= stop_at: raise else: break if POSIX: self.assertEqual(code, -signal.SIGKILL) else: self.assertEqual(code, 0) self.assertFalse(p.is_running())
def kill_proc(procname): done = False for proc in psutil.process_iter(): process = psutil.Process(proc.pid) if process.name() == procname: try: process.terminate() process.wait(timeout=3) done = True except psutil.AccessDenied: print "Error: Access Denied to terminate %s" % procname except psutil.NoSuchProcess: pass except psutil.TimeoutExpired: if proz['killcount'] == 2: print "Error: Terminating of %s failed! (tried 3 times)" % procname else: print "Error: Terminating of %s took to long.. retrying" % procname proz['killcount'] += 1 kill_proc(procname) break if done: print "%s terminated!" % procname
def _wait_output(popen): """Returns `True` if we can get output of the command in the `settings.wait_command` time. Command will be killed if it wasn't finished in the time. :type popen: Popen :rtype: bool """ proc = Process(popen.pid) try: proc.wait(settings.wait_command) return True except TimeoutExpired: for child in proc.children(recursive=True): child.kill() proc.kill() return False
def test_timeout_expired__repr__(self, func=repr): self.assertEqual( repr(psutil.TimeoutExpired(321)), "psutil.TimeoutExpired timeout after 321 seconds") self.assertEqual( repr(psutil.TimeoutExpired(321, pid=111)), "psutil.TimeoutExpired timeout after 321 seconds (pid=111)") self.assertEqual( repr(psutil.TimeoutExpired(321, pid=111, name='foo')), "psutil.TimeoutExpired timeout after 321 seconds " "(pid=111, name='foo')")
def _exec_cmd(self, args, shell, timeout): """Executes adb commands. Args: args: string or list of strings, program arguments. See subprocess.Popen() documentation. shell: bool, True to run this command through the system shell, False to invoke it directly. See subprocess.Popen() docs. timeout: float, the number of seconds to wait before timing out. If not specified, no timeout takes effect. Returns: The output of the adb command run if exit code is 0. Raises: AdbError: The adb command exit code is not 0. AdbTimeoutError: The adb command timed out. """ proc = subprocess.Popen( args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=shell) process = psutil.Process(proc.pid) if timeout and timeout <= 0: raise Error('Timeout is not a positive value: %s' % timeout) if timeout and timeout > 0: try: process.wait(timeout=timeout) except psutil.TimeoutExpired: process.terminate() raise AdbTimeoutError(cmd=args, timeout=timeout) (out, err) = proc.communicate() ret = proc.returncode logging.debug('cmd: %s, stdout: %s, stderr: %s, ret: %s', cli_cmd_to_string(args), out, err, ret) if ret == 0: return out else: raise AdbError(cmd=args, stdout=out, stderr=err, ret_code=ret)
def get_is_process_running(pid): process = None if is_int_str(pid): try: process = psutil.Process(pid=int(pid)) exitCode = process.wait(0) log.debug('waited for process.. exitCode: {}'.format(exitCode)) except (psutil.NoSuchProcess, psutil.ZombieProcess, psutil.TimeoutExpired): pass return process
def test_wait(self): # check exit code signal sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.kill() code = p.wait() if POSIX: self.assertEqual(code, signal.SIGKILL) else: self.assertEqual(code, 0) self.assertFalse(p.is_running()) sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.terminate() code = p.wait() if POSIX: self.assertEqual(code, signal.SIGTERM) else: self.assertEqual(code, 0) self.assertFalse(p.is_running()) # check sys.exit() code code = "import time, sys; time.sleep(0.01); sys.exit(5);" sproc = get_test_subprocess([PYTHON, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertFalse(p.is_running()) # Test wait() issued twice. # It is not supposed to raise NSP when the process is gone. # On UNIX this should return None, on Windows it should keep # returning the exit code. sproc = get_test_subprocess([PYTHON, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertIn(p.wait(), (5, None)) # test timeout sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.name() self.assertRaises(psutil.TimeoutExpired, p.wait, 0.01) # timeout < 0 not allowed self.assertRaises(ValueError, p.wait, -1) # XXX why is this skipped on Windows?
def test_wait(self): # check exit code signal sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.kill() code = p.wait() if POSIX: self.assertEqual(code, -signal.SIGKILL) else: self.assertEqual(code, 0) self.assertFalse(p.is_running()) sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.terminate() code = p.wait() if POSIX: self.assertEqual(code, -signal.SIGTERM) else: self.assertEqual(code, 0) self.assertFalse(p.is_running()) # check sys.exit() code code = "import time, sys; time.sleep(0.01); sys.exit(5);" sproc = get_test_subprocess([PYTHON, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertFalse(p.is_running()) # Test wait() issued twice. # It is not supposed to raise NSP when the process is gone. # On UNIX this should return None, on Windows it should keep # returning the exit code. sproc = get_test_subprocess([PYTHON, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertIn(p.wait(), (5, None)) # test timeout sproc = get_test_subprocess() p = psutil.Process(sproc.pid) p.name() self.assertRaises(psutil.TimeoutExpired, p.wait, 0.01) # timeout < 0 not allowed self.assertRaises(ValueError, p.wait, -1) # XXX why is this skipped on Windows?
def kill_process_tree(logger, pid, timeout=DEFAULT_TIME_TO_WAIT_AFTER_SIGTERM): """ TODO(saguziel): also kill the root process after killing descendants Kills the process's descendants. Kills using the `kill` shell command so that it can change users. Note: killing via PIDs has the potential to the wrong process if the process dies and the PID gets recycled in a narrow time window. :param logger: logger :type logger: logging.Logger """ try: root_process = psutil.Process(pid) except psutil.NoSuchProcess: logger.warning("PID: {} does not exist".format(pid)) return # Check child processes to reduce cases where a child process died but # the PID got reused. descendant_processes = [x for x in root_process.children(recursive=True) if x.is_running()] if len(descendant_processes) != 0: logger.info("Terminating descendant processes of {} PID: {}" .format(root_process.cmdline(), root_process.pid)) temp_processes = descendant_processes[:] for descendant in temp_processes: logger.info("Terminating descendant process {} PID: {}" .format(descendant.cmdline(), descendant.pid)) if not kill_using_shell(logger, descendant.pid, signal.SIGTERM): descendant_processes.remove(descendant) logger.info("Waiting up to {}s for processes to exit..." .format(timeout)) try: psutil.wait_procs(descendant_processes, timeout) logger.info("Done waiting") except psutil.TimeoutExpired: logger.warning("Ran out of time while waiting for " "processes to exit") # Then SIGKILL descendant_processes = [x for x in root_process.children(recursive=True) if x.is_running()] if len(descendant_processes) > 0: temp_processes = descendant_processes[:] for descendant in temp_processes: logger.info("Killing descendant process {} PID: {}" .format(descendant.cmdline(), descendant.pid)) if not kill_using_shell(logger, descendant.pid, signal.SIGKILL): descendant_processes.remove(descendant) else: descendant.wait() logger.info("Killed all descendant processes of {} PID: {}" .format(root_process.cmdline(), root_process.pid)) else: logger.debug("There are no descendant processes to kill")
