zoukankan      html  css  js  c++  java
  • 卡死进程检杀工具

    [本文出自天外归云的博客园]

    Python环境:Python3.6

    ProcessManager参数

    logger:自己的日志输入logger

    targets:包含目标字符串(目标cmd的一部分,可以用来识别cmd的字符串)的列表

    duration:目标进程的最大存活时间,单位分钟,超过就会被杀掉

    程序运行后每隔60s检查一次

     

    代码如下:

    import psutil
    from os import system as run_cmd
    from os import name as system_name
    from time import time, sleep
    import logging
    from os.path import dirname, abspath, join
    
    
    def get_logger():
        log_dir = abspath(dirname(__file__))
        log_name = f"{__file__}.log"
        log_path = join(log_dir, log_name)
        formatter = logging.Formatter('%(asctime)s %(message)s')
        _logger = logging.getLogger()
        for h in [logging.FileHandler(log_path), logging.StreamHandler()]:
            h.setFormatter(formatter)
            _logger.addHandler(h)
        _logger.setLevel(logging.INFO)
        return _logger
    
    
    my_logger = get_logger()
    
    
    def every(duration: int):
        def keep_run(func):
            def wrapper(*args, **kwargs):
                while True:
                    sleep(duration)
                    print("KEEP RUNNING")
                    func(*args, **kwargs)
            return wrapper
        return keep_run
    
    
    """
        ProcessManager
        Keep checking process with target cmd every 60 seconds 
        If process with target cmd found and lives longer than duration
        Just kill it
    """
    
    
    class ProcessManager:
        def __init__(self, logger, targets, duration):
            """
            :param logger: logging.getLogger()
            :param targets: [strA, strB, ...]
            :param duration: int
            """
            self.log = logger
            self.targets = targets
            self.duration = duration
    
        def p_kill(self, pid):
            try:
                if system_name == "nt":
                    run_cmd(f"taskkill /pid {pid}")
                else:
                    run_cmd(f"kill -9 {pid}")
                self.log.info(f"[PROC KILL] {pid}")
            except Exception as e:
                self.log.info(f"[PROC KILL EXP] {e}")
    
        def is_target_cmd(self, cmd):
            for target in self.targets:
                if target in cmd:
                    self.log.info(f"[TARGET FOUND] {cmd}")
                    return True
            return False
    
        def need_kill(self, p: psutil.Process):
            duration = abs(int(p.create_time())-int(time()))/60
            if int(duration) > self.duration:
                self.log.info(f"[NEED KILL] {p.pid} with duration: {duration}")
                return True
            return False
    
        @every(60)
        def kill_dead_processes(self):
            for p in psutil.process_iter():
                try:
                    cmds = p.cmdline()
                    if len(cmds) > 0:
                        for cmd in cmds:
                            if self.is_target_cmd(cmd) and self.need_kill(p):
                                self.p_kill(p.pid)
                except Exception as e:
                    if type(e) == psutil.AccessDenied:
                        pass
                    elif type(e) == psutil.ZombieProcess:
                        pass
                    elif type(e) == TypeError:
                        pass
                    else:
                        self.log.info(p.pid, e)
    
    
    if __name__ == '__main__':
        my_targets = ["-xctestrun"]
        ProcessManager(
            logger=my_logger,
            targets=my_targets,
            duration=4
        ).kill_dead_processes()
  • 相关阅读:
    Assembly之instruction之CMP
    Assembly之Instruction之Byte and Word
    MSP430之section(1)
    MSP430之software development flow
    openMSP430之Custom linker script
    CDC之Synchronizers
    CDC之Metastability
    Embedded之Stack之三
    Embedded之Stack之二
    Embedded之Stack之一
  • 原文地址:https://www.cnblogs.com/LanTianYou/p/11264707.html
Copyright © 2011-2022 走看看